说实话，第一次看到 main.py 只有 37 行的时候，我有点意外。一个 AI Agent 框架的入口，能有多复杂？

但仔细读完整个调用链，才发现这 37 行代码背后，藏着一整套经过深思熟虑的架构设计。今天就来聊聊这些值得学习的设计亮点，以及 LLM 如何响应问题、选择工具的核心机制。

---

调用流程时序图

先看整体调用流程，心里有个全貌：

---

入口代码长什么样

import argparse
import asyncio

from app.agent.manus import Manus
from app.logger import logger


async def main():
    parser = argparse.ArgumentParser(description="Run Manus agent with a prompt")
    parser.add_argument("--prompt", type=str, required=False)
    args = parser.parse_args()

    agent = await Manus.create()
    try:
        prompt = args.prompt if args.prompt else input("Enter your prompt: ")
        if not prompt.strip():
            logger.warning("Empty prompt provided.")
            return

        logger.warning("Processing your request...")
        await agent.run(prompt)
        logger.info("Request processing completed.")
    except KeyboardInterrupt:
        logger.warning("Operation interrupted.")
    finally:
        await agent.cleanup()


if __name__ == "__main__":
    asyncio.run(main())

37 行，但每个部分都有讲究。

---

亮点一：异步工厂方法

agent = await Manus.create()

为什么不用 Manus() 直接实例化？因为 Manus 的初始化需要连接 MCP 服务器——这是异步操作。Pydantic 的 __init__ 是同步的，没法做异步初始化。

所以 Manus 实现了异步工厂方法：

@classmethod
async def create(cls, **kwargs) -> "Manus":
    instance = cls(**kwargs)              # 同步实例化
    await instance.initialize_mcp_servers() # 异步初始化
    instance._initialized = True
    return instance

实例化和初始化分离，用户拿到的是一个完全就绪的对象。

---

亮点二：三层继承体系

BaseAgent (基础抽象类)
    ↓
ReActAgent (ReAct 模式抽象)
    ↓
ToolCallAgent (工具调用能力)
    ↓
Manus (具体实现)

BaseAgent 定义骨架：状态管理、内存管理、执行循环。

ReActAgent 引入 ReAct 范式：每一步都是 think() → act()。

ToolCallAgent 实现工具调用逻辑：调用 LLM、执行工具、处理结果。

Manus 是最终实现：定义工具集、系统提示词、MCP 连接。

---

亮点三：状态机 + 上下文管理器

@asynccontextmanager
async def state_context(self, new_state: AgentState):
    previous_state = self.state
    self.state = new_state
    try:
        yield
    except Exception as e:
        self.state = AgentState.ERROR
        raise e
    finally:
        self.state = previous_state

无论执行成功还是异常，状态都会自动恢复。

---

核心机制：LLM 如何选择工具

这是整个框架最核心的部分：LLM 是怎么"知道"该调用哪个工具的？

OpenAI Function Calling 协议

1. 工具定义

每个工具需要描述清楚：名字、功能、参数：

class PythonExecute(BaseTool):
    name: str = "python_execute"
    description: str = "Executes Python code string..."
    parameters: dict = {
        "type": "object",
        "properties": {
            "code": {"type": "string", "description": "The Python code to execute."},
        },
        "required": ["code"],
    }

2. 转换为 OpenAI 格式

def to_param(self) -> Dict:
    return {
        "type": "function",
        "function": {
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters,
        },
    }

3. 发送给 LLM

response = await self.client.chat.completions.create(
    model=self.model,
    messages=messages,
    tools=tools,              # 工具定义列表
    tool_choice=tool_choice,  # auto/none/required
)

4. LLM 响应工具调用

response.tool_calls = [
    {
        "id": "call_abc123",
        "type": "function",
        "function": {
            "name": "python_execute",
            "arguments": '{"code": "print(1+1)"}'
        }
    }
]

---

亮点四：Tool 泛化调用机制

这是整个工具系统的精髓：如何用统一的代码调用任意工具？

问题：工具差异巨大

看看这几个工具，差异很明显：

# Python 执行工具
class PythonExecute(BaseTool):
    async def execute(self, code: str, timeout: int = 5) -> Dict:
        ...

# 终止工具
class Terminate(BaseTool):
    async def execute(self, status: str) -> str:
        ...

# 询问人类工具
class AskHuman(BaseTool):
    async def execute(self, inquire: str) -> str:
        ...

方法签名完全不同：参数名不同、参数数量不同、返回类型不同。

如果为每个工具写专门的调用代码，那太蠢了——新增工具就要改调用逻辑。

解决方案：统一接口 + 动态分发

第一步：BaseTool 定义统一接口

class BaseTool(ABC, BaseModel):
    name: str
    description: str
    parameters: Optional[dict] = None

    async def __call__(self, **kwargs) -> Any:
        """让工具对象可调用"""
        return await self.execute(**kwargs)

    @abstractmethod
    async def execute(self, **kwargs) -> Any:
        """子类实现具体逻辑"""

关键点：**kwargs 接收任意参数，子类按需提取。

第二步：ToolCollection 管理所有工具

class ToolCollection:
    def __init__(self, *tools: BaseTool):
        self.tools = tools
        self.tool_map = {tool.name: tool for tool in tools}  # 名称 → 实例映射

    async def execute(self, *, name: str, tool_input: Dict[str, Any] = None) -> ToolResult:
        tool = self.tool_map.get(name)  # 根据名称找到工具
        if not tool:
            return ToolFailure(error=f"Tool {name} is invalid")
        try:
            result = await tool(**tool_input)  # 统一调用
            return result
        except ToolError as e:
            return ToolFailure(error=e.message)

第三步：execute_tool 解析 LLM 返回的参数

async def execute_tool(self, command: ToolCall) -> str:
    name = command.function.name

    # 关键：LLM 返回的是 JSON 字符串，需要解析
    args = json.loads(command.function.arguments or "{}")

    # 统一调用，不管是什么工具
    result = await self.available_tools.execute(name=name, tool_input=args)

    return f"Observed output of cmd `{name}` executed:\n{str(result)}"

完整调用链

LLM 返回 tool_call
    ↓
json.loads(arguments) → {"code": "print(1+1)"}
    ↓
ToolCollection.execute(name="python_execute", tool_input={"code": "print(1+1)"})
    ↓
tool_map["python_execute"] → PythonExecute 实例
    ↓
await tool(**{"code": "print(1+1)"})  →  tool.execute(code="print(1+1)")
    ↓
返回 ToolResult

为什么这种设计是"泛化"的？

新增工具零改动：只需继承 BaseTool，定义 name、description、parameters，实现 execute 方法。调用链完全不用动。

# 新增一个工具，只要这样写
class MyNewTool(BaseTool):
    name: str = "my_new_tool"
    description: str = "这是一个新工具"
    parameters: dict = {...}

    async def execute(self, **kwargs):
        # 实现逻辑
        return "done"

# 注册到 Manus
available_tools: ToolCollection = ToolCollection(
    PythonExecute(),
    MyNewTool(),  # 加这一行就行
    ...
)

# 调用链完全不变！LLM 会自动学习这个新工具

统一错误处理：所有工具异常都在 ToolCollection 层捕获，返回 ToolFailure。

统一结果格式：所有工具返回 ToolResult，包含 output、error、base64_image 等字段。

---

完整调用代码示例

import asyncio
from app.agent.manus import Manus
from app.schema import Message

async def demo_tool_call():
    # 1. 创建 Agent
    agent = await Manus.create()

    # 2. 模拟用户请求
    user_prompt = "用 Python 计算 123 * 456 的结果"
    agent.update_memory("user", user_prompt)

    # 3. think() - 调用 LLM
    response = await agent.llm.ask_tool(
        messages=agent.messages,
        system_msgs=[Message.system_message(agent.system_prompt)],
        tools=agent.available_tools.to_params(),
        tool_choice="auto",
    )

    # 4. LLM 返回工具调用意图
    print(f"LLM 决定调用: {response.tool_calls[0].function.name}")
    print(f"参数: {response.tool_calls[0].function.arguments}")

    # 5. act() - 泛化执行（不管是什么工具，代码一样）
    for tool_call in response.tool_calls:
        result = await agent.execute_tool(tool_call)  # 泛化调用
        print(f"执行结果: {result}")

    await agent.cleanup()

asyncio.run(demo_tool_call())

执行过程日志：

✨ Manus's thoughts: 我需要用 python_execute 工具来计算
🛠️ Manus selected 1 tools to use
🧰 Tools being prepared: ['python_execute']
🔧 Tool arguments: {"code": "print(123 * 456)"}
🔧 Activating tool: 'python_execute'...
🎯 Tool 'python_execute' completed its mission! Result: 56088

---

亮点五：MCP 动态工具扩展

async def connect_mcp_server(self, server_url: str, server_id: str = ""):
    await self.mcp_clients.connect_sse(server_url, server_id)
    new_tools = [t for t in self.mcp_clients.tools if t.server_id == server_id]
    self.available_tools.add_tools(*new_tools)  # 动态添加

---

亮点六：资源清理的完整性

finally:
    await agent.cleanup()

级联清理：浏览器关闭、MCP 断开、沙箱清理。

---

写在最后

看完 main.py 和背后的架构，最大的感受是：代码的简洁来自架构的清晰。

这套工具泛化调用机制尤其值得学习：

• BaseTool 统一接口，**kwargs 接收任意参数
• ToolCollection 集中管理，名称映射 + 统一调用
• execute_tool 解析 JSON，动态分发到具体工具

新增工具只需继承 BaseTool，调用链完全不用动——这才是真正的"开闭原则"。

有问题欢迎评论区交流。