オープンソース「OpenManus」、その背景にあるAIエージェント・アーキテクチャを解説

オープニング：マヌス火災とオープンマヌスの故障

最近、AI界で起きている大きな出来事のひとつは マヌス 強力な機能と柔軟な使い方で注目を集めているAIエージェント「Manus」。簡単に言えば、プログラミング、情報検索、文書処理、ネットサーフィンなど、あらゆる場面で活躍してくれる万能アシスタントだ。

しかし、Manusを使うのはそう簡単ではなく、招待コードが必要だ。そのため、多くの開発者や研究者がManusを使うことができない。誰もが途方に暮れていたその時、オープンソースコミュニティが介入しました！MetaGPTチームの@mannaandpoem、@XiangJinyu、@MoshiQAQ、@didiforgithubは、わずか3時間かけてOpenManusというオープンソースプロジェクトを立ち上げました。これで招待コードなしでManusのパワーを体験できる！さらにエキサイティングなのは、OpenManusがオープンソースプロジェクトであること！

OpenManusの出現は、より多くの人々にAIエージェントの魅力を体験する機会を与えるだけでなく、AIエージェントの開発に新たな活力を注入しています。技術に携わる私たちにとって、OpenManusは良いツールであるだけでなく、優れた学習リソースでもあります。そのコードを研究することで、AI Agentのフレームワーク設計と実装の詳細についてより深い理解を得ることができます。

AIエージェントのフレームワーク：オープンマーナスの設計思想

OpenManusのコード構造は非常に明快で、異なる機能モジュールを組み合わせるビルディングブロックのようなモジュール設計を採用しています。この設計の利点は、コードの再利用性と拡張性が高く、各モジュールの責任が明確であることです。

オープンマーナスのコア・コンポーネントには以下のものがある：

OpenManus
├── Agent (代理层)
│   ├── BaseAgent (基础抽象类)
│   ├── ReActAgent (思考-行动模式)
│   ├── ToolCallAgent (工具调用能力)
│   ├── PlanningAgent (规划能力)
│   ├── SWEAgent (软件工程能力)
│   └── Manus (通用代理)
├── LLM (语言模型层)
├── Memory (记忆层)
├── Tool (工具层)
│   ├── BaseTool (工具基类)
│   ├── PlanningTool (规划工具)
│   ├── PythonExecute (Python 执行)
│   ├── GoogleSearch (搜索工具)
│   ├── BrowserUseTool (浏览器工具)
│   └── ... (其他工具)
├── Flow (工作流层)
│   ├── BaseFlow (基础流程)
│   └── PlanningFlow (规划流程)
└── Prompt (提示词层)

LLMの構成要素：エージェントの頭脳

エージェントを人に例えるなら、LLM（Large Language Model）はエージェントの頭脳であり、ユーザーのコマンドを理解し、レスポンスを生成し、意思決定を行う役割を担っています。オープンマーナスはLLMクラスを通してLLMとの対話をカプセル化します。

class LLM:
_instances: Dict[str, "LLM"] = {}  # 单例模式实现
def __init__(
self, config_name: str = "default", llm_config: Optional[LLMSettings] = None
):
if not hasattr(self, "client"):  # 只初始化一次
llm_config = llm_config or config.llm
llm_config = llm_config.get(config_name, llm_config["default"])
self.model = llm_config.model
self.max_tokens = llm_config.max_tokens
self.temperature = llm_config.temperature
self.client = AsyncOpenAI(
api_key=llm_config.api_key, base_url=llm_config.base_url
)

LLMクラスは2つのコア・メソッドを提供する：

• ask:: 一般的な対話のリクエストを送る
• ask_tool:: ツールの呼び出しでリクエストを送信

async def ask_tool(
self,
messages: List[Union[dict, Message]],
system_msgs: Optional[List[Union[dict, Message]]] = None,
timeout: int = 60,
tools: Optional[List[dict]] = None,
tool_choice: Literal["none", "auto", "required"] = "auto",
temperature: Optional[float] = None,
**kwargs,
):
# 格式化消息
if system_msgs:
system_msgs = self.format_messages(system_msgs)
messages = system_msgs + self.format_messages(messages)
else:
messages = self.format_messages(messages)
# 发送请求
response = await self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature or self.temperature,
max_tokens=self.max_tokens,
tools=tools,
tool_choice=tool_choice,
timeout=timeout,
**kwargs,
)

メモリー・コンポーネント：エージェントのメモリー

Memoryコンポーネントは、Agentのノートブックのようなもので、Agentのダイアログ履歴を記録・管理します。Memoryがあれば、Agentは以前に言ったことを思い出し、ダイアログの一貫性を保つことができます。

class Memory(BaseModel):
"""Stores and manages agent's conversation history."""
messages: List[Message] = Field(default_factory=list)
def add_message(self, message: Union[Message, dict]) -> None:
"""Add a message to memory."""
if isinstance(message, dict):
message = Message(**message)
self.messages.append(message)
def get_messages(self) -> List[Message]:
"""Get all messages in memory."""
return self.messages

メモリー・コンポーネントはエージェントのコア・コンポーネントであり、BaseAgentの update_memory メソッドを使って新しいメッセージを追加します：

def update_memory(
self,
role: Literal["user", "system", "assistant", "tool"],
content: str,
**kwargs,
) -> None:
"""Add a message to the agent's memory."""
message_map = {
"user": Message.user_message,
"system": Message.system_message,
"assistant": Message.assistant_message,
"tool": lambda content, **kw: Message.tool_message(content, **kw),
}
if role not in message_map:
raise ValueError(f"Unsupported message role: {role}")
msg_factory = message_map[role]
msg = msg_factory(content, **kwargs) if role == "tool" else msg_factory(content)
self.memory.add_message(msg)

ツールコンポーネント：エージェントのツールボックス

Tools コンポーネントは Agent と外の世界との橋渡し役で、OpenManus は Agent が様々なツールを呼び出してタスクを実行できる柔軟なツールシステムを実装しています。

class BaseTool(ABC, BaseModel):
name: str
description: str
parameters: Optional[dict] = None
async def __call__(self, **kwargs) -> Any:
"""Execute the tool with given parameters."""
return await self.execute(**kwargs)
    @abstractmethod
async def execute(self, **kwargs) -> Any:
"""Execute the tool with given parameters."""
def to_param(self) -> Dict:
"""Convert tool to function call format."""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
},
}

ツールの実行結果は ToolResult クラスで表現する：

class ToolResult(BaseModel):
"""Represents the result of a tool execution."""
output: Any = Field(default=None)
error: Optional[str] = Field(default=None)
system: Optional[str] = Field(default=None)

OpenManus は以下のような多くの組み込みツールを提供します。 PlanningTool::

class PlanningTool(BaseTool):
"""
A planning tool that allows the agent to create and manage plans for solving complex tasks.
The tool provides functionality for creating plans, updating plan steps, and tracking progress.
"""
name: str = "planning"
description: str = _PLANNING_TOOL_DESCRIPTION
parameters: dict = {
"type": "object",
"properties": {
"command": {
"description": "The command to execute. Available commands: create, update, list, get, set_active, mark_step, delete.",
"enum": [
"create",
"update",
"list",
"get",
"set_active",
"mark_step",
"delete",
],
"type": "string",
},
# 其他参数...
},
"required": ["command"],
}

プランニングコンポーネント：エージェントのプランニング能力

プランニング・コンポーネントは、オープンマーナスが複雑なタスクを処理する能力の鍵です。エージェントが計画を立て、複雑なタスクを小さなステップごとのタスクに分解し、ひとつずつ完了できるようにします。

プランニング・コンポーネントは、主に2つの部分で構成されている：

1. PlanningToolプランの作成、更新、追跡機能を提供。
2. PlanningAgent使用 PlanningTool タスクの計画と実行を行う。

class PlanningAgent(ToolCallAgent):
"""
An agent that creates and manages plans to solve tasks.
This agent uses a planning tool to create and manage structured plans,
and tracks progress through individual steps until task completion.
"""
name: str = "planning"
description: str = "An agent that creates and manages plans to solve tasks"
system_prompt: str = PLANNING_SYSTEM_PROMPT
next_step_prompt: str = NEXT_STEP_PROMPT
available_tools: ToolCollection = Field(
default_factory=lambda: ToolCollection(PlanningTool(), Terminate())
)
# 步骤执行跟踪器
step_execution_tracker: Dict[str, Dict] = Field(default_factory=dict)
current_step_index: Optional[int] = None

PlanningAgent コアとなる方法論には以下のようなものがある：

async def think(self) -> bool:
"""Decide the next action based on plan status."""
prompt = (
f"CURRENT PLAN STATUS:n{await self.get_plan()}nn{self.next_step_prompt}"
if self.active_plan_id
else self.next_step_prompt
)
self.messages.append(Message.user_message(prompt))
# 获取当前步骤索引
self.current_step_index = await self._get_current_step_index()
result = await super().think()
# 关联工具调用与当前步骤
if result and self.tool_calls:
# ...关联逻辑...
return result

フローコンポーネント：エージェントのコラボレーション機能

フローコンポーネントの役割は、より複雑なタスクを達成するために一緒に働く複数のエージェントを調整することである。

class BaseFlow(BaseModel, ABC):
"""Base class for execution flows supporting multiple agents"""
agents: Dict[str, BaseAgent]
tools: Optional[List] = None
primary_agent_key: Optional[str] = None
    @property
def primary_agent(self) -> Optional[BaseAgent]:
"""Get the primary agent for the flow"""
return self.agents.get(self.primary_agent_key)
    @abstractmethod
async def execute(self, input_text: str) -> str:
"""Execute the flow with given input"""

PlanningFlow は、タスクの計画と実行のための具体的なFlowの実装である：

class PlanningFlow(BaseFlow):
"""A flow that manages planning and execution of tasks using agents."""
llm: LLM = Field(default_factory=lambda: LLM())
planning_tool: PlanningTool = Field(default_factory=PlanningTool)
executor_keys: List[str] = Field(default_factory=list)
active_plan_id: str = Field(default_factory=lambda: f"plan_{int(time.time())}")
current_step_index: Optional[int] = None
async def execute(self, input_text: str) -> str:
"""Execute the planning flow with agents."""
try:
# 创建初始计划
if input_text:
await self._create_initial_plan(input_text)
# 执行计划步骤
while await self._has_next_step():
# 获取当前步骤
step_info = await self._get_current_step()
# 选择合适的执行者
executor = self.get_executor(step_info.get("type"))
# 执行步骤
result = await self._execute_step(executor, step_info)
# 更新步骤状态
await self._update_step_status(step_info["index"], "completed")
# 完成计划
return await self._finalize_plan()
except Exception as e:
# 处理异常
return f"Error executing flow: {str(e)}"

OpenManusのエージェント実装：レイヤードアーキテクチャ

オープンマヌス社のエージェントは階層型アーキテクチャを採用しており、基本的な機能から特殊なアプリケーションまでレイヤーごとに構築されています。この設計の利点は、高いコードの再利用性、高い拡張性、各レベルでの明確な責任です。

BaseAgent (抽象基类)
└── ReActAgent (思考-行动模式)
└── ToolCallAgent (工具调用能力)
├── PlanningAgent (规划能力)
├── SWEAgent (软件工程能力)
└── Manus (通用代理)

BaseAgent: ベースのベース

BaseAgent は、フレームワーク全体の基礎であり、エージェントのコアプロパティとメソッドを定義します：

class BaseAgent(BaseModel, ABC):
"""Abstract base class for managing agent state and execution."""
# 核心属性
name: str = Field(..., description="Unique name of the agent")
description: Optional[str] = Field(None, description="Optional agent description")
# 提示词
system_prompt: Optional[str] = Field(None, description="System-level instruction prompt")
next_step_prompt: Optional[str] = Field(None, description="Prompt for determining next action")
# 依赖组件
llm: LLM = Field(default_factory=LLM, description="Language model instance")
memory: Memory = Field(default_factory=Memory, description="Agent's memory store")
state: AgentState = Field(default=AgentState.IDLE, description="Current agent state")
# 执行控制
max_steps: int = Field(default=10, description="Maximum steps before termination")
current_step: int = Field(default=0, description="Current step in execution")

ReActAgent：考えるエージェント

ReActAgent エージェントの実行プロセスを2つのフェーズに分けた「シンク・アクト」モデルが実現されている：

class ReActAgent(BaseAgent, ABC):
    @abstractmethod
async def think(self) -> bool:
"""Process current state and decide next action"""
    @abstractmethod
async def act(self) -> str:
"""Execute decided actions"""
async def step(self) -> str:
"""Execute a single step: think and act."""
should_act = await self.think()
if not should_act:
return "Thinking complete - no action needed"
return await self.act()

ToolCallAgent：ツール対応エージェント

ToolCallAgent エージェントにツールを使用する機能を追加：

class ToolCallAgent(ReActAgent):
"""Base agent class for handling tool/function calls with enhanced abstraction"""
available_tools: ToolCollection = ToolCollection(
CreateChatCompletion(), Terminate()
)
tool_choices: Literal["none", "auto", "required"] = "auto"
async def think(self) -> bool:
# 获取 LLM 响应和工具选择
response = await self.llm.ask_tool(
messages=self.messages,
system_msgs=[Message.system_message(self.system_prompt)]
if self.system_prompt
else None,
tools=self.available_tools.to_params(),
tool_choice=self.tool_choices,
)
self.tool_calls = response.tool_calls
# 处理响应和工具调用
# ...
async def act(self) -> str:
# 执行工具调用
results = []
for command in self.tool_calls:
result = await self.execute_tool(command)
# 添加工具响应到内存
# ...
results.append(result)
return "nn".join(results)

PlanningAgent：計画を立てるエージェント。

PlanningAgent タスクのプランニングと実行のトラッキングが達成された：

class PlanningAgent(ToolCallAgent):
"""
An agent that creates and manages plans to solve tasks.
This agent uses a planning tool to create and manage structured plans,
and tracks progress through individual steps until task completion.
"""
# 步骤执行跟踪器
step_execution_tracker: Dict[str, Dict] = Field(default_factory=dict)
current_step_index: Optional[int] = None
async def think(self) -> bool:
"""Decide the next action based on plan status."""
prompt = (
f"CURRENT PLAN STATUS:n{await self.get_plan()}nn{self.next_step_prompt}"
if self.active_plan_id
else self.next_step_prompt
)
self.messages.append(Message.user_message(prompt))
# 获取当前步骤索引
self.current_step_index = await self._get_current_step_index()
result = await super().think()
# 关联工具调用与当前步骤
if result and self.tool_calls:
# ...关联逻辑...
return result

マヌス：全能のエージェント

Manus はOpenManusの中核となるエージェントで、さまざまなタスクを処理するためのツールや機能を統合しています：

class Manus(ToolCallAgent):
"""
A versatile general-purpose agent that uses planning to solve various tasks.
This agent extends PlanningAgent with a comprehensive set of tools and capabilities,
including Python execution, web browsing, file operations, and information retrieval
to handle a wide range of user requests.
"""
name: str = "manus"
description: str = "A versatile general-purpose agent"
system_prompt: str = SYSTEM_PROMPT
next_step_prompt: str = NEXT_STEP_PROMPT
available_tools: ToolCollection = Field(
default_factory=lambda: ToolCollection(
PythonExecute(), GoogleSearch(), BrowserUseTool(), FileSaver(), Terminate()
)
)

プロンプト：エージェントの行動指針

プロンプトはエージェントシステムの構築において重要な役割を果たし、エージェントに何をすべきかを指示する取扱説明書として機能します。

システムプロンプト：エージェントの役割の定義

システムプロンプトは、エージェントの基本的な役割と動作ガイドラインを設定します：

SYSTEM_PROMPT = "You are OpenManus, an all-capable AI assistant, aimed at solving any task presented by the user. You have various tools at your disposal that you can call upon to efficiently complete complex requests. Whether it's programming, information retrieval, file processing, or web browsing, you can handle it all."

プロンプトは、エージェントがユーザーの要求を満たすために様々なツールを使用できる万能AIアシスタントであることを伝えます。

プランニング・プロンプト：エージェントをプランニングに導く

プランニングプロンプトは、複雑なタスクを実行プランのある小さなタスクに分割する方法をエージェントに伝えます：

PLANNING_SYSTEM_PROMPT = """
You are an expert Planning Agent tasked with solving complex problems by creating and managing structured plans.
Your job is:
1. Analyze requests to understand the task scope
2. Create clear, actionable plans with the `planning` tool
3. Execute steps using available tools as needed
4. Track progress and adapt plans dynamically
5. Use `finish` to conclude when the task is complete
Available tools will vary by task but may include:
- `planning`: Create, update, and track plans (commands: create, update, mark_step, etc.)
- `finish`: End the task when complete
Break tasks into logical, sequential steps. Think about dependencies and verification methods.
"""

このプロンプトは、エージェントがプランニングの専門家であることを示し、プランニングの専門家であることを示す。 planning 計画を作成、更新、追跡するためのツール。

ツール使用プロンプト： エージェントにツールの使用方法を指示します。

ツール使用プロンプトは、エージェントが適切なツールを選択できるように、各ツールの機能と使用シナリオを詳細に説明します：

NEXT_STEP_PROMPT = """You can interact with the computer using PythonExecute, save important content and information files through FileSaver, open browsers with BrowserUseTool, and retrieve information using GoogleSearch.
PythonExecute: Execute Python code to interact with the computer system, data processing, automation tasks, etc.
FileSaver: Save files locally, such as txt, py, html, etc.
BrowserUseTool: Open, browse, and use web browsers.If you open a local HTML file, you must provide the absolute path to the file.
GoogleSearch: Perform web information retrieval
Based on user needs, proactively select the most appropriate tool or combination of tools. For complex tasks, you can break down the problem and use different tools step by step to solve it. After using each tool, clearly explain the execution results and suggest the next steps.
"""

ダイナミック・プロンプト：エージェントをより柔軟に

OpenManus のプロンプトは静的なものだけでなく、動的に生成することもできます。たとえば PlanningAgent をプロンプトに入力すると、システムは現在のプランステータスをプロンプトに追加します：

async def think(self) -> bool:
"""Decide the next action based on plan status."""
prompt = (
f"CURRENT PLAN STATUS:n{await self.get_plan()}nn{self.next_step_prompt}"
if self.active_plan_id
else self.next_step_prompt
)
self.messages.append(Message.user_message(prompt))

このダイナミックなプロンプトにより、エージェントは現在の状況に基づいてより合理的な判断を下すことができます。

総括：オープンマーナスからの洞察

オープンマーナスのコードを分析することで、AIエージェントフレームワークのいくつかの主要なコンポーネントを要約することができる：

1. 代理店ベーシックからプロフェッショナルまで、さまざまなレベルの能力を達成するための階層的デザイン。
   • BaseAgent基本的な状態管理と実行ループを提供。
   • ReActAgentシンク・アクト・モデルの実現。
   • ToolCallAgentツール呼び出し機能を追加。
   • プロフェッショナル・エージェント：例 PlanningAgentそしてSWEAgent 歌で応える Manus.
2. LLM大規模な言語モデルとのインタラクションをカプセル化し、対話とツールの呼び出し機能を提供します。
   • 通常の対話とツールコールをサポート。
   • リトライ機構とエラー処理を実装する。
   • ストリーミング対応。
3. メモリー対話の歴史と背景を管理する
   • メッセージの保存と検索
   • 対話の文脈を維持する
4. 工具外界とのインターフェイスを提供する。
   • 基本的なツールの抽象化。
   • そのための複数の専用ツール。
   • ツール結果の処理。
5. プランニングタスクの計画と実行の追跡を可能にする。
   • プランの作成と管理
   • ステップステータスのトラッキング。
   • ダイナミックな調整プログラム
6. フロー複数のエージェントのコラボレーションを管理する。
   • タスキング。
   • 結果の統合。
   • プロセス制御。
7. プロンプトエージェントの行動と意思決定を導く。
   • システム・プロンプトは役割を定義する。
   • プロフェッショナル・プロンプトが決断の指針
   • 動的プロンプト生成。

明確な設計と構造化されたコードを持つOpenManusは、AIエージェントの実装について学ぶための優れた例です。そのモジュール設計により、開発者は簡単にエージェントを拡張し、カスタマイズすることができます。

OpenManusは、AIエージェントについてもっと学びたい、あるいは独自のエージェントシステムを構築したいという開発者に、良い出発点を提供します。そのアーキテクチャと実装について学ぶことで、AIエージェントがどのように機能し、どのように設計されているかをよりよく理解することができます。