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

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

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

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

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

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

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

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

オープンマウス
エージェント（エージェント層）
├─ BaseAgent（ベース抽象クラス）
│ ├── ReActAgent（シンクアクトモード）
├── ToolCallAgent（ツール呼び出し機能）
├── PlanningAgent（プランニング機能）
│ ├── SWEAgent（ソフトウェアエンジニアリング機能）
│ └─ Manus（ジェネリック・エージェント）
├── LLM（言語モデリング層） │ ├── Memory（言語モデル層）
メモリー（メモリー層）
ツール（ツール層）
├─ BaseTool（ツール基本クラス）
プランニングツール
PythonExecute（Python実行） │ ├── PythonExecute（Python実行
├── GoogleSearch（検索ツール）
├── BrowserUseTool（ブラウザツール）
│ └── ... (その他のツール）
├─ Flow（ワークフローレイヤー） │ ├─ BaseFlow
├─ BaseFlow（ベースフロー）
└─ PlanningFlow（プランニングフロー）
└─ Prompt（プロンプト・ワード・レイヤー） │ └─ 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 または 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つのコア・メソッドを提供する：

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

非同期 def ask_tool(
self、
messages: List[Union[dict, Message]], system_msgs: Optional[List[Union[dict, Message]] = None, async
system_msgs: Optional[List[Union[dict, Message]] = None, timeout: int = 60, async ask_tool(
system_msgs: Optional[List[Union[dict, Message]] = None, timeout: int = 60、

tool_choice: Literal["none", "auto", "required"] = "auto", temperature: Optional[float] = None
temperature: オプション[float] = なし、
**kwargs.
).
# メッセージのフォーマット
if system_msgs.
system_msgs = self.format_messages(system_msgs)
messages = system_msgs + self.format_messages(messages)
messages = self.format_messages(system_msgs)
messages = self.format_messages(messages)
# リクエストを送信する
response = await self.client.chat.completions.create(
model=self.model、
messages=messages, temperature=temperature, or self.chat.completions.create(
temperature=temperature or self.temperature, max_tokens=self.temperature, max_tokens=self.temperature
max_tokens=self.max_tokens, tools=tools、
max_tokens=self.max_tokens, tools=tools, tool_choice=tool_choice
tool_choice=tool_choice、
timeout=timeout, **kwargs, **kwargs, **kwargs
**kwargs。
)

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

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

class Memory(BaseModel).
"""エージェントの会話履歴を保存・管理する。""""
messages: List[Message] = Field(default_factory=list)
def add_message(self, message: Union[Message, dict]) -> None.
"""メッセージをメモリに追加する。"""
if isinstance(message, dict).
message = Message(**message)
self.messages.append(message)
def get_messages(self) -> List[Message].
"""メモリ内のすべてのメッセージを取得する。"""
return self.messages

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

def update_memory(
self.
role: Literal["user", "system", "assistant", "tool"]、
content: str, **kwargs, **kwargs, **kwargs.
**を返します。
) -> なし。
"""エージェントのメモリにメッセージを追加します。""""
message_map = {
"user": Message.user_message, "" "エージェントのメモリにメッセージを追加します。""
"system": Message.system_message, "assistant": Message.
"assistant": Message.assistant_message, "tool": lambda content, message_map
"tool": lambda content, **kw: Message.tool_message(content, **kw)、
}
if role not in message_map.
raise ValueError(f "サポートされていないメッセージの役割：{role}")
msg_factory = message_map[ロール］
msg = msg_factory(content, **kwargs) if role == "tool" else msg_factory(content)
self.memory.add_message(msg)

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

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

クラス BaseTool(ABC, BaseModel).
name: str
説明: str
parameters: Optional[dict] = なし
async def __call__(self, **kwargs) -> Any.
"""与えられたパラメータでツールを実行する。""""
return await self.execute(**kwargs)。
    抽象メソッド
async def execute(self, **kwargs) -> Any.
"""与えられたパラメータでツールを実行する。""""
def to_param(self) -> Dict.
"""ツールを関数呼び出し形式に変換する。""""
return {

"function": {
「name": self.name、"description": self.description、"function": {。
description": self.description、"parameters": self.parameters、"function": { "name": self.name
「parameters": self.parameters, }, return
}, }
}

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

class ToolResult(BaseModel).
"""ツールの実行結果を表す。"""
output: Any = Field(default=None)。
error: Optional[str] = Field(default=None)
system: オプション[str] = Field(default=None)

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

クラス PlanningTool(BaseTool).
"""
エージェントが複雑なタスクを解決するための計画を作成し、管理できるようにする計画ツール。
このツールは、計画の作成、計画ステップの更新、進行状況の追跡のための機能を提供します。
"""
name: str = "プランニング"
説明: str = _PLANNING_TOOL_DESCRIPTION
パラメータ: dict = {
"タイプ": "オブジェクト", "プロパティ": {
「プロパティ": {
"命令": {
「利用可能なコマンド: create、update、list、get、set_active、mark_step、delete、
"enum": [
"create"、
"create"、"update"、
"list", "get", "set_active", delete.
"get"、"set_active"、"set_step"、"list"、"get"。


「delete"。
]、"type"。
「type": "string", }, "type".
}, .
# その他のパラメータ...
}, "required".
"required": ["command"], }, #その他のパラメータ....
}

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

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

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

1. プランニングツールプランの作成、更新、追跡機能を提供。
2. プランニング・エージェント使用 プランニングツール タスクの計画と実行を行う。

class PlanningAgent(ToolCallAgent).
"""
タスクを解決するための計画を作成・管理するエージェント。
このエージェントは、計画ツールを使用して、構造化された計画を作成および管理し、タスク完了まで個々のステップを通して進捗を追跡します。
このエージェントは、計画ツールを使って構造化された計画を作成・管理し、タスク完了まで個々のステップの進捗を追跡する。
"""
name: str = "プランニング"
name: str = "planning" description: str = "タスクを解決するための計画を作成し、管理するエージェント"
システムプロンプト: str = PLANNING_SYSTEM_PROMPT
next_step_prompt: str = NEXT_STEP_PROMPT
available_tools: ToolCollection = フィールド(
default_factory=lambda: ToolCollection(PlanningTool(), Terminate())
)
#ステップ実行トラッカー
step_execution_tracker: Dict[str, Dict] = Field(default_factory=dict)
current_step_index: オプション[int] = なし

プランニング・エージェント コアとなる方法論には以下のようなものがある：

async def think(self) -> bool.
"""プランの状態に基づいて次のアクションを決定する。""""
prompt = (
f "現在のプランステータス: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：
# ...ロジックを関連付ける
結果を返す

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

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

クラス BaseFlow(BaseModel, ABC)。
"""複数のエージェントをサポートする実行フローの基本クラス"""
agents: Dict[str, BaseAgent].
tools: Optional[List] = None
primary_agent_key: オプション[str] = None
    プロパティ
def primary_agent(self) -> Optional[BaseAgent].
"""フローのプライマリエージェントを取得する"""
return self.agents.get(self.primary_agent_key)
    抽象メソッド
async def execute(self, input_text: str) -> str.
"""与えられた入力でフローを実行する"""

プランニングフロー は、タスクの計画と実行のための具体的なFlowの実装である：

クラス PlanningFlow(BaseFlow).
"""エージェントを使ってタスクの計画と実行を管理するフロー""""
llm: LLM = Field(default_factory=lambda: LLM())
planning_tool: PlanningTool = Field(default_factory=PlanningTool)
executor_keys: リスト[str] = Field(default_factory=list)
active_plan_id: str = Field(default_factory=lambda: f "plan_{int(time.time())}")
current_step_index: オプション[int] = なし
async def execute(self, input_text: str) -> str.
"""エージェントによる計画フローを実行する"""""
try.
# 初期プランを作成する
if input_text: await self._create_initial_plan
await self._create_initial_plan(入力_テキスト)
# 計画ステップの実行
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._finalise_plan()
except Exception as e.
# 例外処理
return f "フロー実行エラー: {str(e)}"

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

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

BaseAgent（抽象ベースクラス）
ReActAgent（シンクアクトモード）
└─ ToolCallAgent（ツール呼び出し機能）
├─ PlanningAgent（プランニング機能）
├── SWEAgent（ソフトウェアエンジニアリング機能）
└─ Manus（汎用エージェント）

BaseAgent: ベースのベース

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

class BaseAgent(BaseModel, ABC).
"""エージェントの状態と実行を管理するための抽象ベースクラス"""
#コアプロパティ
name: str = Field(...説明="エージェントの固有名")
description: Optional[str] = Field(なし, description="オプションのエージェントの説明")
# プロンプト
system_prompt: オプション[str] = Field(なし, description="システムレベルの命令プロンプト")
next_step_prompt: オプション[str] = Field(なし、説明="次のアクションを決定するプロンプト")
#の依存関係
llm: LLM = Field(default_factory=LLM, description="言語モデルインスタンス")
memory: Memory = Field(default_factory=Memory, description="エージェントのメモリストア")
state: AgentState = Field(default=AgentState.IDLE, description="現在のエージェント状態")
#実行制御
max_steps: int = Field(default=10、description="終了までの最大ステップ数")
current_step: int = Field(default=0、description="現在の実行ステップ")

ReActAgent：考えるエージェント

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

class ReActAgent(BaseAgent, ABC).
    抽象メソッド
async def think(self) -> bool.
"""現在の状態を処理し、次のアクションを決定する""""
    抽象メソッド
async def act(self) -> str.
"""決定したアクションを実行する"""
async def step(self) -> str.
"""1つのステップを実行する：考えて行動する"""""
should_act = await self.think()
if not should_act: return "思考完了 - アクションなし。
return "思考完了 - アクション不要"
return await self.act()

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

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

クラス ToolCallAgent(ReActAgent).
"""抽象化を強化したツール/ファンクションコールを処理するためのベースエージェントクラス""""
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(
メッセージ=self.messages、
system_msgs=[Message.system_message(self.system_prompt)]。
if self.system_prompt
メッセージ、system_msgs=[メッセージ.system_message(self.system_prompt)
tools=self.available_tools.to_params()、
tool_choice=self.tool_choices, )
)
self.tool_calls = response.tool_calls
# レスポンスとツールコールの処理
# ...
async def act(self) -> str.
# ツールコールを実行する
結果 = [].
for command in self.tool_calls：
results = await self.execute_tool(command)
# ツールのレスポンスをメモリに追加
# ...
results.append(結果)
return "nn".join(results)

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

プランニング・エージェント タスクのプランニングと実行のトラッキングが達成された：

class PlanningAgent(ToolCallAgent).
"""
タスクを解決するための計画を作成・管理するエージェント。
このエージェントは、計画ツールを使用して、構造化された計画を作成および管理し、タスク完了まで個々のステップを通して進捗を追跡します。
このエージェントは、計画ツールを使って構造化された計画を作成・管理し、タスク完了まで個々のステップの進捗を追跡する。
"""
# ステップ実行トラッカー
step_execution_tracker: Dict[str, Dict] = Field(default_factory=dict)
current_step_index: Optional[int] = なし
async def think(self) -> bool.
"""プランのステータスに基づいて次のアクションを決定する""""
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：
# ...ロジックを関連付ける
結果を返す

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

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

クラス Manus(ToolCallAgent).
"""
様々なタスクを解決するためにプランニングを使用する多目的な汎用エージェント。
このエージェントは、Python実行、ウェブブラウジング、ファイル操作、情報管理を含む包括的なツールと機能のセットでPlanningAgentを拡張します。
このエージェントは、Pythonの実行、Webブラウジング、ファイル操作、情報管理を含む包括的なツールと機能のセットでPlanningAgentを拡張します。 回収
幅広いユーザーの要求に対応するため
"""
name: str = "manus"
description: str = "多目的な汎用エージェント"
システムプロンプト： str = SYSTEM_PROMPT
next_step_prompt: str = NEXT_STEP_PROMPT
available_tools: ToolCollection = フィールド(
default_factory=lambda: ToolCollection(
PythonExecute(), GoogleSearch(), BrowserUseTool(), FileSaver(), Terminate()
)
)

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

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

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

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

SYSTEM_PROMPT = "あなたはOpenManusです。ユーザーから提示されたどんなタスクでも解決することを目的とした、あらゆる能力を備えたAIアシスタントです。 あなたは自由に使える様々なツールを持っています。あなたは、複雑な要求を効率的に完了するために呼び出すことができる様々なツールを自由に使うことができます。プログラミングであれ、情報検索であれ、ファイル処理であれ、ウェブブラウジングであれ、あなたはすべてを処理することができます。"

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

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

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

planning_system_prompt = """
あなたは、構造化された計画を作成・管理することで複雑な問題を解決することを任務とする専門家プランニング・エージェントです。
あなたの仕事は
1.リクエストを分析してタスクの範囲を理解する
2.`planning`ツールを使って明確で実行可能なプランを作成する
3.必要に応じて、利用可能なツールを使ってステップを実行する。
4.進捗を追跡し、計画を動的に適応させる
5.タスクが完了したら `finish` を使って結論を出す
利用可能なツールはタスクによって異なるが、以下のようなものがある。
- planning`: 計画の作成、更新、追跡 (コマンド: create、update、mark_step など)
- 完了`: 完了したらタスクを終了する
タスクを論理的で連続したステップに分割する。 依存関係と検証方法を考える。
"""

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

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

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

NEXT_STEP_PROMPT = """次のような対話ができる。 コンピュータ PythonExecuteを使い、FileSaverで重要なコンテンツや情報ファイルを保存し、BrowserUseToolでブラウザを開き、GoogleSearchで情報を取得する。GoogleSearchを使って情報を検索する。
PythonExecute：Pythonコードを実行し、コンピュータシステムとの対話、データ処理、自動化タスクなどを実行します。
FileSaver: txt、py、htmlなどのファイルをローカルに保存します。
BrowserUseTool: 開く、 ブラウズローカルのHTMLファイルを開く場合は、ファイルへの絶対パスを指定する必要があります。
GoogleSearch：ウェブ情報検索の実行
複雑なタスクの場合は、問題を分解し、段階的に異なるツールを使用して解決することができます。複雑なタスクについては、問題を分解し、段階的に異なるツールを使用して解決することができます。 各ツールを使用した後、実行結果を明確に説明し、次のステップを提案する。
"各ツールを使用した後、実行結果を明確に説明し、次のステップを提案する。

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

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

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

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

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