【SpringAI】第七弹：深入解析 Al 智能体概念 Cot 、 ReAct 并解析 OpenManus 关键源码源码、基于分层架构实现了拥有自主规划能力的 AI 智能体
深入解析 Al 智能体概念 Cot 、 ReAct 并解析 OpenManus 关键源码源码、基于分层架构实现了拥有自主规划能力的 AI 智能体
本节重点
学习 AI 智能体的概念和构建方法，掌握如何实现具有自主决策能力的 AI 系统。通过 CoT 思维链、ReAct 模式等核心技术，深入理解智能体的工作原理，并参考 OpenManus 开源框架实战开发能够自主规划执行任务的 AI 智能体。
具体内容包括：
- AI 智能体概念与特点
- 智能体实现关键技术
- 使用 AI 智能体的多种方式
- OpenManus 实现原理
- 自主实现 Manus 智能体
- 智能体工作流编排
- A2A 协议
一、什么是智能体？
智能体（Agent）是一个能够感知环境、进行推理、制定计划、做出决策并自主采取行动以实现特定目标的 AI 系统。
它以大语言模型为核心，集成 记忆、知识库和工具 等能力为一体，构造了完整的决策能力、执行能力和记忆能力，就像一个有主观能动性的人类一样。
与普通的 AI 大模型不同，智能体能够：
- 感知环境：通过各种输入渠道获取信息（多模态），理解用户需求和环境状态
- 自主规划任务步骤：将复杂任务分解为可执行的子任务，并设计执行顺序
- 主动调用工具完成任务：根据需要选择并使用各种外部工具和 API，扩展自身能力边界
- 进行多步推理：通过思维链（Chain of Thought）逐步分析问题并推导解决方案
- 持续学习和记忆过去的交互：保持上下文连贯性，利用历史交互改进决策
- 根据环境反馈调整行为：根据执行结果动态调整策略，实现闭环优化
大多数同学第一次感受到智能体应该是 “深度思考” 功能，这是 AI 逐步智能化的体现：
AI 会像人类一样先思考后回答，让答案更准确：
智能体的分类
跟人的生长阶段一样，智能体也是可以不断进化的。按照自主性和规划能力，智能体可以分为几个层次：
1）反应式智能体：仅根据当前输入和固定规则做出反应，类似简单的聊天机器人，没有真正的规划能力。23 年时的大多数 AI 聊天机器人应用，几乎都是反应式智能体。
2）有限规划智能体：能进行简单地多步骤执行，但执行路径通常是预设的或有严格限制的。鉴定为 “能干事、但干不了复杂的大事”。24 年流行的很多可联网搜索内容、调用知识库和工具的 AI 应用，都属于这类智能体。比如 ChatGPT + Plugins：
3）自主规划智能体：也叫目标导向智能体，能够根据任务目标自主分解任务、制定计划、选择工具并一步步执行，直到完成任务。
比如 25 年初很火的 Manus 项目，它的核心亮点在于其 “自主执行” 能力。据官方介绍，Manus 能够在虚拟机中调用各种工具（如编写代码、爬取数据）完成任务。其应用场景覆盖旅行规划、股票分析、教育内容生成等 40 余个领域，所以在当时给了很多人震撼感。
但其实早在这之前，就有类似的项目了，比如 AutoGPT，所以 Manus 大火的同时也被人诟病 “会营销而已”。甚至没隔多久就有小团队开源了 Manus 的复刻版 —— OpenManus，这类智能体通过 “思考-行动-观察” 的循环模式工作，能够持续推进任务直至完成目标。
需要注意，自主规划能力是智能体发展的重要方向，但并非所有应用场景都需要完全的自主规划能力。在某些场景中，限制智能体的自主性反而能提高效率和安全性。
二、智能体实现关键技术
在自主开发智能体前，我们要先了解一下智能体的关键实现技术，也就是方案设计阶段做的事情。
CoT 思维链
CoT（Chain of Thought）思维链是一种让 AI 像人类一样 “思考” 的技术，帮助 AI 在处理复杂问题时能够按步骤思考。对于复杂的推理类问题，先思考后执行，效果往往更好。而且还可以让模型在生成答案时展示推理过程，便于我们理解和优化 AI。
CoT 的实现方式其实很简单，可以在输入 Prompt 时，给模型提供额外的提示或引导，比如 “让我们一步一步思考这个问题”，让模型以逐步推理的方式生成回答。还可以运用 Prompt 的优化技巧 few shot，给模型提供包含思维链的示例问题和答案，让模型学习如何构建自己的思维链。
在 OpenManus 早期版本中，可以看到实现 CoT 的系统提示词：
You are an assistant focused on Chain of Thought reasoning. For each question, please follow these steps:
1. Break down the problem: Divide complex problems into smaller, more manageable parts
2. Think step by step: Think through each part in detail, showing your reasoning process
3. Synthesize conclusions: Integrate the thinking from each part into a complete solution
4. Provide an answer: Give a final concise answer
Your response should follow this format:
Thinking: [Detailed thought process, including problem decomposition, reasoning for each step, and analysis]
Answer: [Final answer based on the thought process, clear and concise]
Remember, the thinking process is more important than the final answer, as it demonstrates how you reached your conclusion.
我们来测试一下：
Agent Loop 执行循环
Agent Loop 是智能体最核心的工作机制，指智能体在没有用户输入的情况下，自主重复执行推理和工具调用的过程。
在传统的聊天模型中，每次用户提问后，AI 回复一次就结束了。但在智能体中，AI 回复后可能会继续自主执行后续动作（如调用工具、处理结果、继续推理），形成一个自主执行的循环，直到任务完成（或者超出预设的最大步骤数）。
Agent Loop 的实现很简单，参考代码如下：
public String execute() {
List<String> results = new ArrayList<>();
while (currentStep < MAX_STEPS && !isFinished) {
currentStep++;
// 这里实现具体的步骤逻辑
String stepResult = executeStep();
results.add("步骤 " + currentStep + ": " + stepResult);
}
if (currentStep >= MAX_STEPS) {
results.add("达到最大步骤数: " + MAX_STEPS);
}
return String.join("\n", results);
}
ReAct 模式
ReAct（Reasoning + Acting）是一种结合推理和行动的智能体架构，它模仿人类解决问题时 ”思考 - 行动 - 观察” 的循环，目的是通过交互式决策解决复杂任务，是目前最常用的智能体工作模式之一。
核心思想：
- 推理（Reason）：将原始问题拆分为多步骤任务，明确当前要执行的步骤，比如 “第一步需要打开编程导航网站”。
- 行动（Act）：调用外部工具执行动作，比如调用搜索引擎、打开浏览器访问网页等。
- 观察（Observe）：获取工具返回的结果，反馈给智能体进行下一步决策。比如将打开的网页代码输入给 AI。
- 循环迭代：不断重复上述 3 个过程，直到任务完成或达到终止条件。
ReAct 流程如图：
示例实现代码：
void executeReAct(String task) {
String state = "开始";
while (!state.equals("完成")) {
// 1. 推理 (Reason)
String thought = "思考下一步行动";
System.out.println("推理: " + thought);
// 2. 行动 (Act)
String action = "执行具体操作";
System.out.println("行动: " + action);
// 3. 观察 (Observe)
String observation = "观察执行结果";
System.out.println("观察: " + observation);
// 更新状态
state = "完成";
}
}
所需支持系统
除了基本的工作机制外，智能体的实现还依赖于很多支持系统。
1. 首先是 AI 大模型，这个就不多说了，大模型提供了思考、推理和决策的核心能力，越强的 AI 大模型通常执行任务的效果越好。
2. 记忆系统
智能体需要记忆系统来存储对话历史、中间结果和执行状态，这样它才能够进行连续对话并根据历史对话分析接下来的工作步骤。之前我们学习过如何使用 Spring AI 的 ChatMemory 实现对话记忆。
3. 知识库
尽管大语言模型拥有丰富的参数知识，但针对特定领域的专业知识往往需要额外的知识库支持。之前我们学习过，通过 RAG 检索增强生成 + 向量数据库等技术，智能体可以检索并利用专业知识回答问题。
4. 工具调用
工具是扩展智能体能力边界的关键，**智能体通过工具调用可以访问搜索引擎、数据库、API 接口等外部服务，**极大地增强了其解决实际问题的能力。当然，MCP 也可以算是工具调用的一种。
💡 这里分享一种特殊的工具调用 —— Compute Use，**它允许智能体直接与计算环境交互，比如执行代码、操作文件系统等。**目前 Claude 等平台已提供此类能力，感兴趣的同学可以阅读下面的资料学习：
- Claude 支持 Compute Use：https://docs.anthropic.com/en/docs/agents-and-tools/computer-use
- Compute Use 实现示例：https://github.com/anthropics/anthropic-quickstarts/tree/main/computer-use-demo
- Compute Use 开源实现：https://github.com/e2b-dev/open-computer-use、https://github.com/showlab/computer_use_ootb
使用 Compute Use 功能时需要注意，这些操作会在我们自己的系统上执行，建议在虚拟环境中运行，防止意外操作影响实际系统。
综合上面 4 类技术，并且结合 CoT、Agent Loop、ReAct 等机制（可以总称为 “规划执行机制”），我们就可以构建一个完整的、有自主规划能力的智能体系统啦。
三、使用 AI 智能体
有 3 种方式可以使用 AI 智能体，之前的教程中其实我们已经有接触过，这里再快速复习一遍。
1、平台中使用
许多 AI 大模型开发平台已经提供了智能体创建和使用功能，最简单的方式就是直接在这些平台上创建和使用智能体。
比如可以在阿里云百炼、Dify 等平台上创建智能体：
2、软件中使用
AI 开发工具 Cursor 就集成了 AI 智能体，可以帮我们生成完整的项目代码、或者解释项目代码。选择 Agent 模式就可以直接使用了，建议同时开启 Thinking 深度思考：
运行效果如图，典型的 CoT + Agent Loop + ReAct 模式：
3、程序中使用
之前有讲解过，如果使用 AI 大模型开发平台创建了智能体，那么可以直接通过 SDK 或 API 在自己的程序中调用智能体。
当然也可以完全通过编程自主开发智能体，之前我们通过 Spring AI 创建的恋爱大师 APP 就已经是个智能体了，包含了大模型决策、记忆能力、知识库集成和工具调用。
ChatResponse response = chatClient
.prompt()
.user(message)
// 记忆
.advisors(spec -> spec.param(CHAT_MEMORY_CONVERSATION_ID_KEY, chatId)
.param(CHAT_MEMORY_RETRIEVE_SIZE_KEY, 10))
// 开启日志，便于观察效果
.advisors(new MyLoggerAdvisor())
// 知识库
.advisors(new QuestionAnswerAdvisor(loveAppVectorStore))
// 工具调用
.tools(toolCallbackProvider)
.call()
.chatResponse();
但是，这个实现还不具备完整的自主规划和执行能力，下面我们将学习更高级的自主规划智能体架构。
四、OpenManus 实现原理
在我们开发超级智能体前，可以先学习下优秀的开源项目。
下面以 OpenManus 项目 为例，煮波会带大家从 0 开始阅读项目源码，学到超级智能体实现方式的同时，帮助大家掌握快速学习开源项目的方法。
快速运行
学习开源项目的第一步是阅读官方提供的 README.md 项目介绍文档，先把项目跑起来，借此熟悉业务流程，而不是一上来就啃源码。
OpenManus 官方提供了一个 Demo 视频，我们就不必再自己搭建环境来跑项目了：
下面让我们看一个 OpenManus 的对话示例，帮助我们了解它的工作流程。假设用户要完成一个任务 —— 在编程导航网站上寻找热门文章，对话如下：
用户：帮我在编程导航网站上找到最近的几篇热门文章
系统：OpenManus 已启动，正在处理您的请求...
助手：我会帮您在编程导航网站查找最近的热门文章。让我先用浏览器工具访问编程导航网站。
系统：正在调用浏览器工具...
助手：我成功打开了浏览器并访问了编程导航网站。从网站内容来看，我现在需要导航到热门文章区域，然后提取这些文章的信息。
系统：正在执行浏览器操作...
助手：我已在编程导航网站找到了热门文章区域，以下是最近的几篇热门文章：
1. 《Spring Boot 实战：构建微服务应用》- 阅读量：3200
2. 《前端框架对比：React vs Vue vs Angular》- 阅读量：2850
3. 《Python 数据分析入门：从 Pandas 到可视化》- 阅读量：2400
4. 《Docker 容器化应用部署最佳实践》- 阅读量：2150
5. 《算法学习路线：从入门到精通》- 阅读量：1980
这些文章是根据网站显示的阅读量排序的热门文章。
系统：正在调用终止工具...
助手：已完成任务，成功找到编程导航网站上的热门文章。如果您对某篇文章感兴趣或有其他需求，请随时告诉我。
在这个示例中，OpenManus 首先理解用户任务，然后选择合适的工具（浏览器工具）来执行任务，最后整理信息并返回结果。整个过程体现了 ReAct 模式的核心要素：思考（Thinking）、行动（Action）和观察（Observation）的循环，直到任务完成。
OpenManus 整体架构
了解了 OpenManus 的核心功能后，先 从整体到局部 地观察 OpenManus 的架构，简单来说就是看文件夹的组织形式、文件的命名规则、以及文件之间的关联，暂时不用深入源码。
1、整体文件夹组织形式
先看文件夹的组织形式，见名知意：
2、agent 目录
agent 目录是 OpenManus 实现的核心，采用了分层的代理架构，不同层次的代理负责不同的功能，这样更利于系统的扩展。
OpenManus 的代理架构主要包含以下几层：
- BaseAgent：最基础的代理抽象类，定义了所有代理的基本状态管理和执行循环
- ReActAgent：实现 ReAct 模式的代理，具有思考（Think）和行动（Act）两个主要步骤
- ToolCallAgent：能够调用工具的代理，继承自 ReActAgent 并扩展了工具调用能力
- Manus：具体实现的智能体实例，集成了所有能力并添加了更多专业工具
还有更多适用于特定领域的智能体实例，比如 DataAnalysis 数据分析 Agent、SWE 软件开发工程师 Agent、MCP 服务交互 Agent、Browser 浏览器操作 Agent，它们都继承了 ToolCallAgent。
3、tool 目录
tool 目录定义了各种各样的工具，比如网页搜索、文件操作、询求用户帮助、代码执行器等等：
4、prompt 目录
prompt 目录定义了整个项目中可能会用到的提示词。从下图中我们可以看到提示词写的比较专业，这块也是比较值得学习的。
5、其他支持
为了实现完整的智能体功能，OpenManus 依赖以下关键组件：
- 记忆系统：使用 Memory 类存储对话历史和中间状态
- LLM 大模型：通过 LLM 类提供思考和决策能力
- 工具系统：提供 BaseTool 和 ToolCollection 类扩展智能体的能力边界
- 流程控制：通过 AgentState 和执行循环管理状态转换和任务流程
在 OpenManus 中，这些都是自主实现的：
AI 智能体核心实现
了解整体架构后，我们重点学习 Agent 分层代理架构。
1、BaseAgent
BaseAgent 是所有代理的基础，定义了代理状态管理和执行循环的核心逻辑，就像一个通用模板，规定了代理工作的基本流程：
- 启动前检查：只有代理处于 “空闲” 状态，才能开始工作，否则报错。
- 接收任务：如果有用户请求，就先存到 “记忆” 里。
- 循环工作：用
while循环一步步做事，直到完成任务或达到最大步数。每步会调用step方法（这是个抽象方法，具体做什么由子类决定），并记录结果。 - 防卡死：过程中会检查是否卡住，卡住了就处理。
- 结束工作：完成后重置状态，返回所有步骤的结果。
简单说，它定好了代理 “怎么跑起来、怎么循环工作、怎么收尾” 的规矩，具体每步干啥留给子类自己实现。
查看 base.py 文件，关键代码就是 Agent Loop 的实现，通过 while 实现循环，并且定义了死循环检查机制：
class BaseAgent(BaseModel, ABC):
async def run(self, request: Optional[str] = None) -> str:
"""执行代理的主循环"""
if self.state != AgentState.IDLE:
raise RuntimeError(f"Cannot run agent from state: {self.state}")
if request:
self.update_memory("user", request)
results: List[str] = []
async with self.state_context(AgentState.RUNNING):
while (self.current_step < self.max_steps and
self.state != AgentState.FINISHED):
self.current_step += 1
step_result = await self.step()
# 检查是否陷入循环
if self.is_stuck():
self.handle_stuck_state()
results.append(f"Step {self.current_step}: {step_result}")
if self.current_step >= self.max_steps:
self.current_step = 0
self.state = AgentState.IDLE
results.append(f"Terminated: Reached max steps ({self.max_steps})")
return "\n".join(results) if results else "No steps executed"
@abstractmethod
async def step(self) -> str:
"""执行单步操作，必须由子类实现"""
这里其实使用了模板方法设计模式，父类定义执行流程，具体的执行方法（step）交给子类实现。
2、ReActAgent
ReActAgent 实现了 ReAct 模式，将代理的执行过程分为思考（Think）和行动（Act）两个关键步骤。
简单来说，ReActAgent 是基于 BaseAgent 的一个子类，它把代理的 “一步工作” 拆得更细了：
它在 BaseAgent 规定的 step 方法（每步要做的事）里，加了两个步骤：
- 先调用 think 方法（思考）：判断现在该做什么
- 如果需要行动，就调用 act 方法（行动）：执行具体操作
就像 BaseAgent 要求子类实现 step 一样，ReActAgent 要求子类必须实现 think（怎么思考）和 act（怎么行动）。
简单说，BaseAgent 定了 “要一步步做事” 的规矩，ReActAgent 则把 “每一步” 明确分为 “先想清楚再动手”，让代理的行为更符合人类 “思考 - 行动” 的模式。
查看 react.py 文件：
class ReActAgent(BaseAgent, ABC):
@abstractmethod
async def think(self) -> bool:
"""处理当前状态并决定下一步行动"""
@abstractmethod
async def act(self) -> str:
"""执行决定的行动"""
async def step(self) -> str:
"""执行单步：思考和行动"""
should_act = await self.think()
if not should_act:
return "Thinking complete - no action needed"
return await self.act()
上述代码同样运用了模板方法设计模式，这种设计体现了 ReAct 模式的核心思想，也就是 “思考 - 行动 - 观察” 的循环过程。但是具体怎么思考、怎么行动，交给子类去实现。
3、ToolCallAgent
ToolCallAgent 在 ReAct 模式的基础上增加了工具调用能力，是 OpenManus 最重要的一个层次。
具体的，ToolCallAgent 是 ReActAgent 的子类，继承了 “思考 - 行动” 的模式，还增加了使用工具的能力，整体逻辑承上启下：
它遵循 BaseAgent 规定的 “一步步工作” 框架，也按 ReActAgent 要求把每步拆成 “思考” 和 “行动”，但让这两步更具体：
- think（思考）：通过 AI 决定该用什么工具（比如查资料、发消息等），并记录下要调用的工具
- act（行动）：按思考的结果，实际去调用这些工具，执行操作后保存结果
简单说，BaseAgent 定了 “步骤循环”，ReActAgent 分了 “思考 - 行动”，让每个步骤先想清楚做什么再行动，ToolCallAgent 则让思考能 “选工具”、行动能 “用工具”，让代理能借助各种工具完成更复杂的任务。
查看 toolcall.py 文件，虽然代码比较复杂，但原理很简单，就是工具调用机制的具体实现：
- think：和 AI 交互思考使用什么工具
- act：程序执行工具
- observe：将结果返回给 AI
class ToolCallAgent(ReActAgent):
"""能够执行工具调用的代理类"""
available_tools: ToolCollection = ToolCollection(
CreateChatCompletion(), Terminate()
)
tool_choices: TOOL_CHOICE_TYPE = ToolChoice.AUTO
special_tool_names: List[str] = Field(default_factory=lambda: [Terminate().name])
async def think(self) -> bool:
"""处理当前状态并使用工具决定下一步行动"""
# 添加下一步提示到用户消息(拼接提示词)
if self.next_step_prompt:
user_msg = Message.user_message(self.next_step_prompt)
self.messages += [user_msg]
# 请求 LLM 选择工具(AI 从能使用的工具列表中选择, 将 AI 选择结果作为响应的其中一个参数)
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,
)
# 处理工具调用 (拿到 AI 返回的调用工具响应)
self.tool_calls = tool_calls = (
response.tool_calls if response and response.tool_calls else []
)
content = response.content if response and response.content else ""
# 添加助手消息到记忆 (从响应中查看 AI 想调用的工具)
assistant_msg = (
Message.from_tool_calls(content=content, tool_calls=self.tool_calls)
if self.tool_calls
else Message.assistant_message(content)
)
self.memory.add_message(assistant_msg)
# 决定是否应该执行行动
return bool(self.tool_calls or content)
async def act(self) -> str:
"""执行工具调用并处理结果"""
if not self.tool_calls:
# 返回最后一条消息内容，如果没有工具调用
return self.messages[-1].content or "No content or commands to execute"
results = []
for command in self.tool_calls:
# 执行工具并获取结果
result = await self.execute_tool(command)
# 把工具调用结果存入记忆 (对话上下文)
tool_msg = Message.tool_message(
content=result,
tool_call_id=command.id,
name=command.function.name,
base64_image=self._current_base64_image,
)
# 收集每一步工具调用的结果, 到最终的结果列表中
self.memory.add_message(tool_msg)
results.append(result)
# 把结果列表 (也就是每个步骤调用的工具的结果)整合后返回
return "\n\n".join(results)
总结：从 BaseAgent 搭好 “一步步循环做事” 的基础框架，到 ReActAgent 把 “每一步” 细化成 “先想后做” 的流程，再到 ToolCallAgent 给 “想” 和 “做” 加上 “选工具”“用工具” 的能力，层层递进的关系
4、Manus
Manus 类是 OpenManus 的核心智能体实例，集成了各种工具和能力。
Manus 是 ToolCallAgent 的子类，完全继承了
- BaseAgent（步骤循环）、
- ReActAgent（思考 - 行动）、
- ToolCallAgent（选工具 - 用工具）
的核心能力，是 OpenManus 能直接用的 “成品智能体”，它像 BaseAgent 那样能一步步循环工作，像 ReActAgent 那样会 “先想后做”，像 ToolCallAgent 那样能调用工具，并且具体配备了多种实用工具（比如运行代码、浏览网页、编辑文字、询问人类等）。
具体逻辑如下：
1. 基础信息与核心工具配置
- 给智能体定了名字（Manus）和描述，还内置了一批常用工具（比如运行 Python 代码、用浏览器、编辑文字、询问人类、结束任务）；
- 同时加了 MCP 客户端，让它不仅能用完内置工具，还能调用 MCP 提供的远程工具，能力更全面。
2. 便捷的创建与初始化方式
- 提供了
create工厂方法，创建 Manus 实例时，会自动初始化 MCP 服务器，不用手动操作，方便直接启动使用。
3. 优化的 “思考” 逻辑
- 重写了 ReActAgent 传下来的
think方法：本质还是调用父类（ToolCallAgent）的 “思考” 逻辑来选工具，但在这之前会做两件关键事：- 确保 MCP 服务器已初始化（保证远程工具能用）
- 如果正在用浏览器，就自动添加浏览器相关的特殊上下文（让思考更贴合当前工具场景）
- 思考结束后再恢复原始配置，既保留了父类的核心能力，又适配了自身工具的使用需求。
简单说，前面的类都是 “基础框架”，而 Manus 是基于这些框架打造的 “成品智能体”，有了具体名字和实用工具，可以直接用来处理各种实际任务。
先简单了解一下源码的实现细节和优点（我们可以先看加粗的实现细节，思考一下这样的实现有什么优点，在看高亮位置标注的优点加深理解），再看源码：
-
这段代码定义了 Manus 智能体的名称、描述和可直接使用的工具列表，还补充了 MCP 客户端配置，让 Manus 不仅能调用本地工具，还能使用 MCP 提供的远程工具，进一步扩展了可用工具的范围。
-
Manus 重写了
think方法，从super().think()能看出，它本质还是依赖父类（ToolCallAgent）的 “思考” 逻辑来决定用什么工具，但在调用父类方法前，额外做了 MCP 服务器初始化（确保远程工具能正常用）、添加浏览器特殊上下文（让思考更适配当前工具场景）的操作， 既没丢父类的核心能力，又针对自身工具做了优化。 -
这种分层架构让 agent 拓展很方便：如果我们要开发新 agent，想直接用工具调用能力，就继承 ToolCallAgent；如果觉得 ToolCallAgent 不够，想自己改 “思考 - 行动” 逻辑，就继承 ReActAgent；若只需基础步骤循环，就继承 BaseAgent。 按需选择合适的父类，既能复用已有能力，又能让新 agent 的代码更清晰、拓展性更强。
查看 manus.py 文件：
class Manus(ToolCallAgent):
"""多功能通用智能体，支持本地和 MCP 工具"""
name: str = "Manus"
description: str = "A versatile agent that can solve various tasks using multiple tools"
# 添加各种通用工具到工具集合
available_tools: ToolCollection = Field(
default_factory=lambda: ToolCollection(
PythonExecute(),
BrowserUseTool(),
StrReplaceEditor(),
AskHuman(),
Terminate(),
)
)
# MCP 客户端用于远程工具访问
mcp_clients: MCPClients = Field(default_factory=MCPClients)
@classmethod
async def create(cls, **kwargs) -> "Manus":
"""工厂方法创建并初始化 Manus 实例"""
instance = cls(**kwargs)
await instance.initialize_mcp_servers()
instance._initialized = True
return instance
async def think(self) -> bool:
"""处理当前状态并根据上下文决定下一步行动"""
if not self._initialized:
await self.initialize_mcp_servers()
self._initialized = True
# 为浏览器工具添加特殊上下文
original_prompt = self.next_step_prompt
browser_in_use = self._check_browser_in_use()
if browser_in_use:
self.next_step_prompt = (
await self.browser_context_helper.format_next_step_prompt()
)
result = await super().think()
# 恢复原始提示
self.next_step_prompt = original_prompt
return result
关键实现细节
学完了超级智能体的核心实现后，我们再学习一些项目中比较微妙的实现细节，对我们自己开发项目也会很有帮助。
1、工具系统设计
(1) 工具抽象层 BaseTool
所有工具均继承自 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:
"""执行工具的具体逻辑，由子类实现"""
def to_param(self) -> Dict:
"""将工具转换为函数调用格式"""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
},
}
💡以下说明，煮波建议请重点先理解例子，再结合 例子 理解实现细节，进而先自己思考这样实现有什么优点，再结合优点 加深理解；
怎么理解上面的设计呢？我们用一个生活例子理解：
- 就像手机厂商做 “充电接口”—— 所有手机都用 Type-C 接口（统一调用入口）
- 但每个手机的充电速度、功能（快充 / 数据传输）可以不一样（子类自己实现
execute）； - 同时接口上会标清楚 “这是 Type-C，支持 65W 快充”（
name/description/parameters标签）；
这个设计有什么好处呢？具体好处有 3 个：
1. 对人友好：不用记一堆不同的用法
- 不管用什么工具，调用方式都一样（都是「工具实例 ()」）；
- 不用记 “浏览器工具用
open()、Python 工具用run()”，就像 Java 里用接口，不管是ArrayList还是LinkedList，调用add()方法的方式都一样。
2. 对 AI 友好：AI 能轻松理解工具
- 所有工具的信息都通过
to_param转成统一格式，AI 不用分析每个工具的代码，只要看这个结构化数据，就知道 “这个工具叫啥、能干嘛、要传什么参数” - 避免 AI 因为工具格式不统一而 “不会用”。
3. 对开发友好：新增工具很简单
要做一个新工具（比如 “画图工具”），只需要：
-
继承 BaseTool；
-
填好
name（“画图工具”）、description（“生成图片”）、parameters（“图片关键词”）； -
写
execute方法（具体怎么生成图片）。
不用再考虑 “调用方式怎么设计”、“怎么让 AI 看懂”—— 框架已经把通用规则定好了，开发只需要关注 “这个工具具体干啥”。
总结
BaseTool 的核心思路就是**「定规矩、做统一」**：
- 定死所有工具必须有的基础信息（身份标签）；
- 定死所有工具的调用方式（统一入口）；
- 定死工具信息给 AI 的格式（统一沟通语言）；
- 只留 “工具具体干啥” 给子类自己实现（灵活扩展）。
这和 Java 里用「抽象类 + 接口」设计通用组件的思路完全一致，因此，这种设计既保证了整体的规范性，又不限制单个组件的灵活性；使得每个工具都有统一的调用方式，同时具有规范化的参数描述，便于 LLM 理解工具的使用方法。
(2) 终止工具 Terminate
Terminate 工具是一个特殊的工具，允许智能体通过 AI 大模型自主决定何时结束任务，避免无限循环或者过早结束。
具体的代码说明如下：
💡以下说明，煮波建议请重点先理解例子，再结合 例子 理解实现细节，进而先自己思考这样实现有什么优点，再结合优点 加深理解；
1. 给工具 “贴标签”：固定名字 + 明确用途，一眼知道是什么、什么时候用；
- 这段代码设计了一个“终止工具”，用于让智能体决定何时结束任务，防止任务陷入无限循环或过早结束。
- 代码给这个工具取了
固定名字“terminate”，并详细描述了它的用途：任务完成或者进行不下去时使用。 - 就好比给家里的“总开关”贴上了清晰的标签，让人一看便知何时使用。
2. 给结果 “设选项”：只能选 “成功 / 失败”，不允许模糊表述；
- 在“终止结果”方面，只设定了“success”（成功）和“failure”（失败）这两个选项，避免了模糊不清的情况。
- 这就像完成一件事情后，只能给出“做成了”或者“没做成”这两种明确答复。
3. 给操作 “封流程”：用专门的方法处理终止，逻辑不重复、结果可追溯；
- “execute”方法专门用来处理终止逻辑，接收任务的结束状态，执行终止操作，并返回结果。
- 就如同按下“总开关”后，开关不仅切断电源，还会告知你“电源已关，原因是XXX”。
4. 给特殊工具 “列清单”：单独管理，避免和普通工具混用。
- 此外，代码中还有个“special_tool_names”变量，用于列出特殊工具，对“终止工具”这类特殊工具进行单独管理，防止在程序中被误操作。
- 就像把家里的“总开关”单独列在特殊工具清单里，使用时特别留意。
class Terminate(BaseTool):
name: str = "terminate"
description: str = """Terminate the interaction when the request is met OR
if the assistant cannot proceed further with the task.
When you have finished all the tasks, call this tool to end the work."""
parameters: dict = {
"type": "object",
"properties": {
"status": {
"type": "string",
"description": "The finish status of the interaction.",
"enum": ["success", "failure"],
}
},
"reqzuired": ["status"],
}
async def execute(self, status: str) -> str:
"""完成当前执行"""
return f"The interaction has been completed with status: {status}"
在 agent 源码中，有一个 special_tool_names 变量，用于指定终止工具等特殊工具：
在 toolcall.py 源码中，对特殊工具进行处理：
(3) 询问工具 AskHuman
这个工具到底是干嘛的？
这个工具本质是给智能体加了个“求助按钮”——就像人遇到不懂的事会问别人一样，程序卡壳时也能主动找人类要信息。
我们可以把它想象成“程序的请教功能”。比如程序要帮你订机票，却发现你没说“要订哪天的”，它没法继续，就会弹出一句：“请问你想订哪一天的机票呀？”——这个“提问+等你回答”的过程，就是AskHuman工具在工作。
核心解决一个问题：程序不是万能的，遇到“缺信息”“拿不准”的情况，不瞎猜、不卡住，而是主动找人类要帮助。
总结：AskHuman 工具允许智能体在遇到无法自主解决的问题时向人类寻求帮助，也就是给用户一个输入框，让我们能够更好地干预智能体完成任务的过程。
我们先来看看源码，再对源码的精妙设计进行说明：
class AskHuman(BaseTool):
"""Add a tool to ask human for help."""
name: str = "ask_human"
description: str = "Use this tool to ask human for help."
parameters: str = {
"type": "object",
"properties": {
"inquire": {
"type": "string",
"description": "The question you want to ask human.",
}
},
"required": ["inquire"],
}
async def execute(self, inquire: str) -> str:
return input(f"""Bot: {inquire}\n\nYou: """).strip()
和之前的“终止工具”思路相通，AskHuman 工具都是用简单规则让“求助”这件事清晰、不混乱，Java里完全能复用这些逻辑。下面，我们来看看源码里的3个“精妙细节”，
细节1：固定名字+直白描述（name和description）
name: str = "ask_human" # 工具名
description: str = "Use this tool to ask human for help." # 用途
我们通俗理解一下这个细节，就像给手机APP里的一个按钮贴标签：
- 按钮名：“问用户”（程序一看就知道这是“求助按钮”）；
- 按钮说明：“遇到解决不了的问题，点这个按钮问用户”（写代码的人不会用错）。
这样设计有什么优点：
- 程序不认错：不管有多少工具，只要看到“ask_human”，就知道是“要找人类帮忙”，不会和“计算”“查询”等工具搞混；
- 人不滥用：不会在“程序自己能解决”（比如算1+1）的时候，还调用这个工具去问人。
如果要对应 Java 代码，就是用常量定名字，注释写用途：
// 工具名：固定为“ask_human”
public static final String TOOL_NAME = "ask_human";
// 用途说明：遇到无法自主解决的问题时，调用此工具向人类求助
/**
* Use this tool to ask human for help when the agent can't solve the problem by itself.
*/
细节2：必须传“明确的问题”（inquire参数+required）
"inquire": {
"type": "string",
"description": "The question you want to ask human.", # 要问人类的具体问题
"required": ["inquire"] # 必须填这个问题
}
我们通俗理解一下这个细节：
- 程序想“求助”时，不能只说“我卡住了”，必须说具体问题——比如不能说“我不知道怎么订”，得说“请问你要订经济舱还是商务舱？”。
- 而且这个问题是“必填项”，程序不能“空着手求助”（比如没写问题就弹出输入框，你都不知道它要问啥）。
这样设计有什么优点：
- 人类能快速回应：不会让你猜“程序到底要什么信息”（比如问“要哪天的机票”，你直接答日期就行）；
- 程序不做无用功：避免“没问清楚就等回答”，减少卡住的概率。
如果要对应 Java 代码，就是用方法参数强制传“问题”，比如：
// 调用求助工具时，必须传入要问人类的具体问题（inquire）
public String askHuman(String inquire) {
if (inquire == null || inquire.trim().isEmpty()) {
throw new IllegalArgumentException("必须填写要问人类的具体问题");
}
// 后续逻辑：显示问题、接收人类回答
return ...;
}
细节3：“提问+等回答”一步到位（execute方法）
async def execute(self, inquire: str) -> str:
return input(f"""Bot: {inquire}\n\nYou: """).strip()
我们通俗理解一下这个细节：
-
程序调用这个方法时，会自动做两件事：
- 把问题给你看**（比如屏幕上显示“Bot：请问你要订哪天的机票？”）**；
- 等你输入回答**（你输“10月1日”，它就把这个回答存起来，继续往下处理）**。
-
整个过程不用额外写代码，调用一次方法就完成“提问→收回答”。
这样设计有什么优点：
- 逻辑封装：不管后续要问多少次（比如问完日期问舱位），都不用重复写“显示问题、等输入”的代码，直接调用 execute 就行；
- 结果直接用：收到的回答能直接传给程序下一步（比如你答“10月1日”，程序就用这个日期查机票）。
如果要对应 Java 代码，就是用方法封装“显示问题+读输入”的逻辑，比如：
import java.util.Scanner;
public class AskHumanTool {
private Scanner scanner = new Scanner(System.in);
// 执行求助：显示问题，接收人类回答并返回
public String execute(String inquire) {
// 1. 显示程序的问题
System.out.println("Bot: " + inquire);
System.out.print("You: ");
// 2. 读取人类的回答
String answer = scanner.nextLine().trim();
// 3. 返回回答，给程序后续使用
return answer;
}
}
为什么说这个看似简单设计“提升实用性和安全性”？看似简单的工具，其实解决了程序的两个大问题：
1. 提升实用性：不让程序“卡壳”
比如程序要帮你整理文件，却发现“有两个同名文件，不知道删哪个”——如果没有这个工具，程序可能一直卡在这；有了它，程序会主动问“两个‘报告.doc’，要删修改时间早的还是晚的？”，你答完它就能继续干活。
2. 提升安全性：不让程序“瞎猜”
比如程序处理转账，却发现“你没说转多少钱”——如果程序瞎猜“转100块”，可能造成损失；有了这个工具，程序会问“请问你要转账多少金额？”，确认后再操作，避免出错。
(4) 工具集合 ToolCollection
这个工具到底是干嘛的？
-
举个例子，就像一个工具箱，里面可以放锤子、螺丝刀、扳手等各种工具。当你需要拧螺丝时，直接从工具箱里找到螺丝刀使用；需要敲钉子时，找到锤子使用。
-
ToolCollection 类就是这样一个"工具箱"，它用来
管理各种工具（BaseTool 的实例），提供了添加工具、查找工具和执行工具的功能，让使用者可以方便地统一管理和使用不同的工具。
OpenManus 设计了 ToolCollection 类来管理多个工具实例，提供统一的工具注册和执行接口：
class ToolCollection:
"""A collection of defined tools."""
def __init__(self, *tools: BaseTool):
self.tools = tools
self.tool_map = {tool.name: tool for tool in tools}
def to_params(self) -> List[Dict[str, Any]]:
return [tool.to_param() for tool in self.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)
def add_tools(self, *tools: BaseTool):
"""Add multiple tools to the collection."""
for tool in tools:
self.add_tool(tool)
return self
这个工具有什么细节？
细节一：初始化方法（init）
def __init__(self, *tools: BaseTool):
self.tools = tools
self.tool_map = {tool.name: tool for tool in tools}
我们通俗理解一下这个细节，就像：
- 你买了一个新工具箱，第一次使用时，你把家里现有的锤子、螺丝刀、扳手一起放进去。同时，你在工具箱内部做了一个标签对照表，上面写着"锤子"对应哪个工具、"螺丝刀"对应哪个工具。
- 当创建 ToolCollection 这个"工具箱"时，
__init__方法会接收传入的多个工具（就像把工具放进工具箱），然后把这些工具存到self.tools里。同时，会创建一个self.tool_map（就像标签对照表），用工具的名字作为"标签"，对应到具体的工具，方便后续快速找到工具。
这样设计有什么优点？
- 通过创建 tool_map，后续要查找某个工具时，不需要逐个检查所有工具，直接通过工具名称就能快速找到对应的工具，提高了查找效率。
细节二：转换为参数列表的方法（to_params）
def to_params(self) -> List[Dict[str, Any]]:
return [tool.to_param() for tool in self.tools]
我们通俗理解一下这个细节，就像：
- 你的工具箱里有各种工具，现在你需要给别人一张清单，说明每个工具的基本信息（比如锤子的重量、螺丝刀的型号等）。你逐个查看每个工具，把它们的信息整理成一条条记录，然后汇总成一张清单。
- to_params 方法会遍历工具箱里的每个工具，调用每个工具的 to_param() 方法获取该工具的参数信息（比如工具名称、功能描述等），然后把所有工具的参数信息整理成一个列表返回，就像生成了一张工具信息清单。
这样设计有什么优点？
- 比如在需要向用户展示有哪些工具可用时，就可以用这个方法快速获取所有工具的信息。
- 因此，这样设计可以统一获取所有工具的参数信息，方便对外展示或传递工具的基本信息；
细节三：执行工具的方法（execute）
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_map 中找到对应的工具，如果找不到就返回错误信息。找到工具后，会传入工具需要的参数（tool_input）并执行该工具，返回执行结果；如果执行过程中出现错误，就返回错误信息。
这样设计有什么优点？
- 提供了统一的工具执行接口，使用者不需要关心每个工具具体的执行方式，只需知道工具名称和参数就能执行；
- 同时还能统一处理工具不存在或执行出错的情况，简化了工具的使用流程。
细节四：添加工具的方法（add_tools）
def add_tools(self, *tools: BaseTool):
"""Add multiple tools to the collection."""
for tool in tools:
self.add_tool(tool)
return self
我们通俗理解一下这个细节，就像：
- 你又买了新的工具（比如钳子、卷尺），想把它们放进已有的工具箱里。你逐个把这些新工具放进工具箱，同时更新工具箱内部的标签对照表，添加上新工具的名称和对应工具。
- 说明：add_tools 方法用于向工具箱中添加多个新工具。它会逐个处理每个要添加的工具，调用 add_tool 方法（虽然代码中没显示 add_tool 的实现，但可以推测是将工具加入 tools 列表并更新 tool_map），添加完成后返回自身，方便进行链式调用（比如连续添加多批工具）。
这样设计有什么优点？
- 允许随时向工具集合中添加新工具，实现了工具的动态扩展，让工具系统更加灵活，符合可插拔性的设计理念。
- 同时支持一次添加多个工具，提高了使用便利性。
总结：这种设计使得 OpenManus 可以灵活地添加、移除和管理工具，实现了工具系统的可插拔性。我们之前利用 Spring AI 开发工具调用时，也写了个类似的工具注册类。
2、MCP 协议支持
(1) MCP 与工具系统的集成
这个工具到底是干嘛的？
- 就像一个特殊的"网络工具箱"，它继承了普通工具箱的所有功能，但专门用来管理连接到不同网络服务器的工具。比如，它可以同时连接到文件服务器、数据库服务器，并管理这些服务器提供的各种操作工具。
- 还记得么，之前提到过 “MCP 的本质就是工具调用”，OpenManus 的实现也是遵循了这一思想。通过
MCPClients类（继承自 ToolCollection）将 MCP 服务集成到现有工具系统中。 - MCPClients 类是一种特殊的工具集合，它继承了之前的 ToolCollection（工具箱）的所有功能，专门用于连接多个 MCP 服务器，并通过 Model Context Protocol 管理这些服务器提供的工具，让用户可以统一操作不同服务器上的工具。
查看 tool/mcp.py 的源码：
class MCPClients(ToolCollection):
"""
A collection of tools that connects to multiple MCP servers and manages available tools through the Model Context Protocol.
"""
sessions: Dict[str, ClientSession] = {}
exit_stacks: Dict[str, AsyncExitStack] = {}
description: str = "MCP client tools for server interaction"
def __init__(self):
super().__init__() # Initialize with empty tools list
self.name = "mcp" # Keep name for backward compatibility
这个工具有什么细节？
细节一：类的继承关系
class MCPClients(ToolCollection):
我们通俗理解一下这个细节，就像：
- 就像"智能手机"继承了"手机"的所有功能（能打电话、发短信），但又增加了智能系统、上网等新功能。
- MCPClients 类继承了 ToolCollection 类，这意味着它自动拥有了 ToolCollection 的所有功能（比如管理工具、执行工具等），同时还可以添加自己特有的功能（比如连接 MCP 服务器）。
这样设计有什么优点？
- 不需要重新编写 ToolCollection 已经实现的功能（如工具管理、执行等），只需专注于 MCP 服务器相关的特有功能;
- 节省了开发时间，同时保证了与原有工具系统的兼容性。
细节二：类属性定义
sessions: Dict[str, ClientSession] = {}
exit_stacks: Dict[str, AsyncExitStack] = {}
description: str = "MCP client tools for server interaction"
我们通俗理解一下这个细节，就像：
- 一个专门的"快递柜"会有几个固定的部分：存放不同快递的格子（sessions）、管理格子开关的控制系统（exit_stacks）、以及贴在柜子上的说明文字（description）。
- 说明：这些是 MCPClients 类自带的"部件"。sessions 用来存储与各个 MCP 服务器的连接会话（像记录每个服务器的连接信息）；exit_stacks 用来管理这些连接的关闭和资源释放；description 则是对这个工具集合的简单描述，说明它是用于与服务器交互的 MCP 客户端工具。
这样设计有什么优点？
- 提前定义好这些属性，让类的功能更清晰，其他开发者一看就知道这个类会管理服务器连接、资源释放等信息，同时也为后续的功能实现打下基础。
细节三：初始化方法
def __init__(self):
super().__init__() # Initialize with empty tools list
self.name = "mcp" # Keep name for backward compatibility
我们通俗理解一下这个细节，就像：
- 你买了一个继承了普通工具箱功能的"网络工具箱"，第一次使用时，先按照普通工具箱的初始化方式准备好内部结构（比如空的工具存放区），然后给这个特殊的工具箱贴上一个标签"mcp"，方便大家识别。
- 初始化 MCPClients 实例时，首先通过 super().init() 调用父类 ToolCollection 的初始化方法，创建一个空的工具列表。然后设置自己的名字为"mcp"，这样即使系统升级，老的代码还能通过这个名字识别它，保证了兼容性。
这样设计有什么优点？
- 既保证了继承父类的基础功能（如工具管理的初始状态）；
- 又通过设置名字确保了与旧系统的兼容性，让新工具可以平滑融入现有系统。
(2) 动态工具代理
这个工具到底是干嘛的？
- 就像你通过家里的智能音箱（代理）控制远方公司的打印机。你不需要直接接触打印机，只需告诉智能音箱"打印文件"，它会帮你连接公司的打印机并完成操作，最后告诉你结果。
- MCPClientTool 是一个"远程工具代理"，它能代表客户端连接到 MCP 服务器上的工具。当你需要使用服务器上的工具时，不用直接操作服务器，只需通过这个代理发送请求，它会帮你完成远程调用并返回结果。
- 每当连接到 MCP 服务器时，OpenManus 会动态创建
MCPClientTool实例（继承自 BaseTool）作为每个远程工具的代理。
查看 tool/mcp.py 的源码，通过向 MCP 服务器发送远程请求来执行工具：
class MCPClientTool(BaseTool):
"""Represents a tool proxy that can be called on the MCP server from the client side."""
session: Optional[ClientSession] = None
server_id: str = "" # Add server identifier
original_name: str = ""
async def execute(self, **kwargs) -> ToolResult:
"""Execute the tool by making a remote call to the MCP server."""
if not self.session:
return ToolResult(error="Not connected to MCP server")
try:
result = await self.session.call_tool(self.original_name, kwargs)
content_str = ", ".join(
item.text for item in result.content if isinstance(item, TextContent)
)
return ToolResult(output=content_str or "No output returned.")
except Exception as e:
return ToolResult(error=f"Error executing tool: {str(e)}")
这个工具有什么细节
细节一：类的继承与基本属性
class MCPClientTool(BaseTool):
"""Represents a tool proxy that can be called on the MCP server from the client side."""
session: Optional[ClientSession] = None
server_id: str = "" # Add server identifier
original_name: str = ""
我们通俗理解一下这个细节，就像：
- 这就像一个"远程控制遥控器"，它继承了普通遥控器的基本功能（如发送指令），但增加了几个特殊属性：连接电视的信号线（session）、电视的编号（server_id）、以及电视上原功能的名称（original_name，比如"播放"按钮）。
- MCPClientTool 继承了 BaseTool（基础工具类），所以它具备工具的基本特性。同时它有三个关键属性：session 是与 MCP 服务器的连接（类似信号线），server_id 是服务器的标识（类似电视编号），original_name 是服务器上对应工具的原名（类似电视上的功能名称）。
这样设计有什么优点？
- 通过继承基础工具类，保证了它能融入现有的工具系统；
- 而三个属性分别存储了连接信息、服务器标识和工具原名，为远程调用提供了必要的基础信息，让代理知道该连接哪个服务器、调用哪个工具。
细节二：执行远程工具的方法（execute）
async def execute(self, **kwargs) -> ToolResult:
"""Execute the tool by making a remote call to the MCP server."""
if not self.session:
return ToolResult(error="Not connected to MCP server")
try:
result = await self.session.call_tool(self.original_name, kwargs)
content_str = ", ".join(
item.text for item in result.content if isinstance(item, TextContent)
)
return ToolResult(output=content_str or "No output returned.")
except Exception as e:
return ToolResult(error=f"Error executing tool: {str(e)}")
我们通俗理解一下这个细节，就像：
- 你用"远程控制遥控器"（MCPClientTool）让公司的打印机打印文件。首先检查遥控器是否连上了打印机（session是否存在），没连上就提示"未连接"。如果连上了，就通过信号线（session）告诉打印机（服务器）执行"打印"功能（original_name），并传入打印参数（比如文件名，即kwargs）。打印机完成后返回结果，遥控器把结果整理成文字告诉你；如果中途出错（比如纸没了），也会告诉你具体错误。
- execute方法是代理执行远程工具的核心逻辑。首先检查是否已连接服务器，没连接就返回错误。连接正常的话，通过session调用服务器上名为original_name的工具，并传入参数kwargs。然后把服务器返回的结果整理成文字，最后返回执行结果；如果过程中出错，就返回错误信息。
这样设计有什么优点？
- 提供了统一的远程工具调用流程，自动处理连接检查、结果整理和错误捕获，让用户使用远程工具就像使用本地工具一样简单，同时保证了调用过程的稳定性（出错时能明确反馈原因）。
(3) Manus 中的 MCP 集成机制
这个工具到底是干嘛的？
-
就像你的手机连接新的智能设备（如智能灯泡），连接成功后，手机的控制中心会自动添加这个灯泡的控制工具（开关、调光等），之后你就能像操作手机原有功能一样控制这个新设备。
-
connect_mcp_server 是一个用于连接 MCP 服务器并集成其工具的功能。当连接成功后，它会把服务器提供的工具自动添加到本地可用工具集合中，这样大模型使用这些远程工具就和使用本地工具一样方便。
-
Manus 智能体 通过工具调用 实现了与 MCP 服务器的无缝集成。
查看 agent/manus.py 的源码，发现本质上是把 MCP 服务提供的工具动态添加到可用的工具集合中：
async def connect_mcp_server(
self,
server_url: str,
server_id: str = "",
use_stdio: bool = False,
stdio_args: List[str] = None,
) -> None:
"""Connect to an MCP server and add its tools."""
if use_stdio:
await self.mcp_clients.connect_stdio(server_url, stdio_args or [], server_id)
else:
await self.mcp_clients.connect_sse(server_url, server_id)
# 关键实现：动态添加该服务器的工具到可用工具集合
new_tools = [
tool for tool in self.mcp_clients.tools if tool.server_id == server_id
]
self.available_tools.add_tools(*new_tools)
这个工具有什么细节
细节一：连接服务器的两种方式
if use_stdio:
await self.mcp_clients.connect_stdio(server_url, stdio_args or [], server_id)
else:
await self.mcp_clients.connect_sse(server_url, server_id)
我们通俗理解一下这个细节，就像：
- 你连接公司的打印机有两种方式，一种是用 USB 线直接连接（类似 stdio），另一种是通过无线网络连接（类似 sse）。根据实际情况选择其中一种方式，最终都能成功连接并使用打印机。
- 这段代码根据 use_stdio 参数选择不同的连接方式。当 use_stdio 为 True 时，用 connect_stdio 方法连接（类似有线连接）；否则用 connect_sse 方法连接（类似无线连接）。两种方式都能实现与 MCP 服务器的连接。
这样设计有什么优点？
- 支持多种连接方式，能适应不同的使用场景和服务器配置，提高了系统的灵活性和兼容性，无论服务器支持哪种连接方式都能正常对接。
细节二：动态添加服务器工具到本地集合
new_tools = [
tool for tool in self.mcp_clients.tools if tool.server_id == server_id
]
self.available_tools.add_tools(*new_tools)
我们通俗理解一下这个细节，就像：
- 你连接上新的智能家电后，手机会从这个家电支持的功能中筛选出属于它的控制工具（比如只添加"智能灯泡"的控制工具，而不是其他设备的），然后把这些工具添加到手机的控制中心，方便你统一管理和使用。
- 这段代码先从已连接的 MCP 服务器工具中，筛选出属于当前服务器（通过 server_id 匹配）的工具，然后把这些工具添加到本地的可用工具集合（available_tools）中。这样本地就有了这些远程工具，能像使用本地工具一样调用它们。
这样设计有什么优点？
- 实现了远程工具的动态集成，大模型无需区分工具是本地还是远程，调用方式完全一致；
- 既复用了现有的工具调用代码，又让系统能灵活扩展新的服务器工具，简化了开发和使用成本。
这样一来，对大模型来说，MCP 工具与本地工具的调用方式一致，很巧妙地复用了代码。
3、其他值得学习的源码
(1) Python 代码执行沙箱
PythonExecute 工具实现了一个安全的 Python 代码执行环境，这是一个值得学习的安全实现。查看 tool/python_execute.py 源码：
class PythonExecute(BaseTool):
name: str = "python_execute"
async def execute(self, code: str, timeout: int = 5) -> Dict:
"""安全执行 Python 代码"""
with multiprocessing.Manager() as manager:
result = manager.dict({"observation": "", "success": False})
# 安全的全局环境
safe_globals = {"__builtins__": __builtins__.__dict__.copy()}
# 使用独立进程执行代码
proc = multiprocessing.Process(
target=self._run_code, args=(code, result, safe_globals)
)
proc.start()
proc.join(timeout)
# 超时处理
if proc.is_alive():
proc.terminate()
proc.join(1)
return {
"observation": f"Execution timeout after {timeout} seconds",
"success": False,
}
return dict(result)
这段代码展示了几个安全编程的最佳实践：
- 使用独立进程隔离代码执行
- 实现了超时机制防止无限循环
- 截获和处理所有异常
- 重定向标准输出以捕获打印内容
(2) 状态管理与上下文切换机制
BaseAgent 实现了一个优雅的状态管理和上下文切换机制：
@asynccontextmanager
async def state_context(self, new_state: AgentState):
"""Context manager for safe agent state transitions."""
if not isinstance(new_state, AgentState):
raise ValueError(f"Invalid state: {new_state}")
previous_state = self.state
self.state = new_state
try:
yield
except Exception as e:
self.state = AgentState.ERROR # Transition to ERROR on failure
raise e
finally:
self.state = previous_state # Revert to previous state
这个上下文管理器确保了状态转换的安全性和可靠性，即使在异常情况下也能正确恢复状态，是一个值得学习的设计模式。
(3) 工具结果统一表示与组合
OpenManus 设计了 ToolResult 类来统一表示工具执行结果，并支持结果组合。查看 tool/base.py 源码：
class ToolResult(BaseModel):
"""Represents the result of a tool execution."""
output: Any = Field(default=None)
error: Optional[str] = Field(default=None)
base64_image: Optional[str] = Field(default=None)
system: Optional[str] = Field(default=None)
def __add__(self, other: "ToolResult"):
"""组合两个工具结果"""
def combine_fields(field: Optional[str], other_field: Optional[str], concatenate: bool = True):
if field and other_field:
if concatenate:
return field + other_field
raise ValueError("Cannot combine tool results")
return field or other_field
return ToolResult(
output=combine_fields(self.output, other.output),
error=combine_fields(self.error, other.error),
base64_image=combine_fields(self.base64_image, other.base64_image, False),
system=combine_fields(self.system, other.system),
)
这种设计使得工具结果处理更加统一和灵活，特别是在需要组合多个工具结果或处理异常情况时。
五、自主实现 Manus 智能体
虽然 OpenManus 代码量很大，但其实很多代码都是在实现智能体所需的支持系统，比如调用大模型、会话记忆、工具调用能力等。
如果使用 AI 开发框架，这些能力都不需要我们自己实现，代码量会简单很多。
下面就让我们基于 Spring AI 框架，实现一个简化版的 Manus 智能体。
定义数据模型
新建 agent.model 包，将所有用到的数据模型（实体类、枚举类等）都放到该包下。
目前我们只需要定义 Agent 的状态枚举，用于控制智能体的执行。AgentState 代码如下：
/**
* 代理执行状态的枚举类
*/
public enum AgentState {
/**
* 空闲状态
*/
IDLE,
/**
* 运行中状态
*/
RUNNING,
/**
* 已完成状态
*/
FINISHED,
/**
* 错误状态
*/
ERROR
}
核心架构开发
首先定义智能体的核心架构，包括以下类：
- BaseAgent：智能体基类，定义基本信息和多步骤执行流程
- ReActAgent：实现思考和行动两个步骤的智能体
- ToolCallAgent：实现工具调用能力的智能体
- YuManus：最终可使用的 Manus 实例
1、开发基础 Agent 类
参考 OpenManus 的实现方式，BaseAgent 的代码如下：
/**
* 抽象基础代理类，用于管理代理状态和执行流程。
*
* 提供状态转换、内存管理和基于步骤的执行循环的基础功能。
* 子类必须实现step方法。
*/
@Data
@Slf4j
public abstract class BaseAgent {
// 核心属性
private String name;
// 提示
private String systemPrompt;
private String nextStepPrompt;
// 状态
private AgentState state = AgentState.IDLE;
// 执行控制
private int maxSteps = 10;
private int currentStep = 0;
// LLM
private ChatClient chatClient;
// Memory（需要自主维护会话上下文）
private List<Message> messageList = new ArrayList<>();
/**
* 运行代理
*
* @param userPrompt 用户提示词
* @return 执行结果
*/
public String run(String userPrompt) {
if (this.state != AgentState.IDLE) {
throw new RuntimeException("Cannot run agent from state: " + this.state);
}
if (StringUtil.isBlank(userPrompt)) {
throw new RuntimeException("Cannot run agent with empty user prompt");
}
// 更改状态
state = AgentState.RUNNING;
// 记录消息上下文
messageList.add(new UserMessage(userPrompt));
// 保存结果列表
List<String> results = new ArrayList<>();
try {
for (int i = 0; i < maxSteps && state != AgentState.FINISHED; i++) {
int stepNumber = i + 1;
currentStep = stepNumber;
log.info("Executing step " + stepNumber + "/" + maxSteps);
// 单步执行
String stepResult = step();
String result = "Step " + stepNumber + ": " + stepResult;
results.add(result);
}
// 检查是否超出步骤限制
if (currentStep >= maxSteps) {
state = AgentState.FINISHED;
results.add("Terminated: Reached max steps (" + maxSteps + ")");
}
return String.join("\n", results);
} catch (Exception e) {
state = AgentState.ERROR;
log.error("Error executing agent", e);
return "执行错误" + e.getMessage();
} finally {
// 清理资源
this.cleanup();
}
}
/**
* 执行单个步骤
*
* @return 步骤执行结果
*/
public abstract String step();
/**
* 清理资源
*/
protected void cleanup() {
// 子类可以重写此方法来清理资源
}
}
上述代码中，我们要注意 3 点：
- 包含 chatClient 属性，由调用方传入具体调用大模型的对象，而不是写死使用的大模型，更灵活
- 包含 messageList 属性，用于维护消息上下文列表
- 通过 state 属性来控制智能体的执行流程
2、开发 ReActAgent 类
参考 OpenManus 的实现方式，继承自 BaseAgent，并且将 step 方法分解为 think 和 act 两个抽象方法。ReActAgent 的代码如下：
/**
* ReAct (Reasoning and Acting) 模式的代理抽象类
* 实现了思考-行动的循环模式
*/
@EqualsAndHashCode(callSuper = true)
@Data
public abstract class ReActAgent extends BaseAgent {
/**
* 处理当前状态并决定下一步行动
*
* @return 是否需要执行行动，true表示需要执行，false表示不需要执行
*/
public abstract boolean think();
/**
* 执行决定的行动
*
* @return 行动执行结果
*/
public abstract String act();
/**
* 执行单个步骤：思考和行动
*
* @return 步骤执行结果
*/
@Override
public String step() {
try {
boolean shouldAct = think();
if (!shouldAct) {
return "思考完成 - 无需行动";
}
return act();
} catch (Exception e) {
// 记录异常日志
e.printStackTrace();
return "步骤执行失败: " + e.getMessage();
}
}
}
3、开发 ToolCallAgent 类
(1) 选择方案
ToolCallAgent 负责实现工具调用能力，继承自 ReActAgent，具体实现了 think 和 act 两个抽象方法。
我们有 3 种方案来实现 ToolCallAgent：
1. 基于 Spring AI 的工具调用能力，手动控制工具执行。
- 其实 Spring 的 ChatClient 已经支持选择工具进行调用（内部完成了 think、act、observe）
- 但这里我们要自主实现，可以使用 Spring AI 提供的 手动控制工具执行。
2. 基于 Spring AI 的工具调用能力，简化调用流程。
- 由于 Spring AI 完全托管了工具调用；
- 我们可以直接把所有工具调用的代码作为 think 方法，而 act 方法不定义任何动作。
3. 自主实现工具调用能力。
- 也就是工具调用章节提到的实现原理：自己写 Prompt，引导 AI 回复想要调用的工具列表和调用参数，然后再执行工具并将结果返送给 AI 再次执行。
使用哪种方案呢？
- 如果是为了学习 ReAct 模式，让流程更清晰，推荐第一种；
- 如果只是为了快速实现，推荐第二种；
- 不建议采用第三种方案，过于原生，开发成本高。
下面我们采用第一种方案实现 ToolCallAgent；
(2) 定义 ToolCallAgent
先定义所需的属性和构造方法：
/**
* 处理工具调用的基础代理类，具体实现了 think 和 act 方法，可以用作创建实例的父类
*/
@EqualsAndHashCode(callSuper = true)
@Data
@Slf4j
public class ToolCallAgent extends ReActAgent {
// 可用的工具
private final ToolCallback[] availableTools;
// 保存了工具调用信息的响应
private ChatResponse toolCallChatResponse;
// 工具调用管理者
private final ToolCallingManager toolCallingManager;
// 禁用内置的工具调用机制，自己维护上下文
private final ChatOptions chatOptions;
public ToolCallAgent(ToolCallback[] availableTools) {
super();
this.availableTools = availableTools;
this.toolCallingManager = ToolCallingManager.builder().build();
// 禁用 Spring AI 内置的工具调用机制，自己维护选项和消息上下文
this.chatOptions = DashScopeChatOptions.builder()
.withProxyToolCalls(true)
.build();
}
}
(3) 禁止 Spring AI 托管工具调用
注意，在上述代码中，我们通过将 DashScopeChatOptions 的 withProxyToolCalls 选项设置为 true，来禁止 Spring AI 托管工具调用，而是我们自主实现。
虽然官方提供的示例代码是设置 internalToolExecutionEnabled 为 false 来禁用 Spring AI 托管工具调用，但是由于我们使用的是阿里的 DashScopeChatModel 大模型客户端，如果按照下面的方式，会直接导致工具调用失效！
ChatOptions chatOptions = ToolCallingChatOptions.builder()
.toolCallbacks(new CustomerTools())
.internalToolExecutionEnabled(false)
.build();
(4) 实现 think 方法
下面我们实现 think 方法，传入工具列表并调用大模型，得到需要调用的工具列表：
/**
* 处理当前状态并决定下一步行动
*
* @return 是否需要执行行动
*/
@Override
public boolean think() {
if (getNextStepPrompt() != null && !getNextStepPrompt().isEmpty()) {
UserMessage userMessage = new UserMessage(getNextStepPrompt());
getMessageList().add(userMessage);
}
List<Message> messageList = getMessageList();
Prompt prompt = new Prompt(messageList, chatOptions);
try {
// 获取带工具选项的响应
ChatResponse chatResponse = getChatClient().prompt(prompt)
.system(getSystemPrompt())
.tools(availableTools)
.call()
.chatResponse();
// 记录响应，用于 Act
this.toolCallChatResponse = chatResponse;
AssistantMessage assistantMessage = chatResponse.getResult().getOutput();
// 输出提示信息
String result = assistantMessage.getText();
List<AssistantMessage.ToolCall> toolCallList = assistantMessage.getToolCalls();
log.info(getName() + "的思考: " + result);
log.info(getName() + "选择了 " + toolCallList.size() + " 个工具来使用");
String toolCallInfo = toolCallList.stream()
.map(toolCall -> String.format("工具名称：%s，参数：%s",
toolCall.name(),
toolCall.arguments())
)
.collect(Collectors.joining("\n"));
log.info(toolCallInfo);
if (toolCallList.isEmpty()) {
// 只有不调用工具时，才记录助手消息
getMessageList().add(assistantMessage);
return false;
} else {
// 需要调用工具时，无需记录助手消息，因为调用工具时会自动记录
return true;
}
} catch (Exception e) {
log.error(getName() + "的思考过程遇到了问题: " + e.getMessage());
getMessageList().add(
new AssistantMessage("处理时遇到错误: " + e.getMessage()));
return false;
}
}
(5) 实现 act 方法
最后实现 act 方法，执行工具调用列表，得到返回结果，并将工具的响应添加到消息列表中：
/**
* 执行工具调用并处理结果
*
* @return 执行结果
*/
@Override
public String act() {
if (!toolCallChatResponse.hasToolCalls()) {
return "没有工具调用";
}
// 调用工具
Prompt prompt = new Prompt(getMessageList(), chatOptions);
ToolExecutionResult toolExecutionResult = toolCallingManager.executeToolCalls(prompt, toolCallChatResponse);
// 记录消息上下文，conversationHistory 已经包含了助手消息和工具调用返回的结果
setMessageList(toolExecutionResult.conversationHistory());
// 当前工具调用的结果
ToolResponseMessage toolResponseMessage = (ToolResponseMessage) CollUtil.getLast(toolExecutionResult.conversationHistory());
String results = toolResponseMessage.getResponses().stream()
.map(response -> "工具 " + response.name() + " 完成了它的任务！结果: " + response.responseData())
.collect(Collectors.joining("\n"));
log.info(results);
return results;
}
注意维护消息上下文，不要重复添加了消息，因为 toolExecutionResult.conversationHistory() 方法已经包含了助手消息和工具调用返回的结果。
(6) 维护消息上下文原理
可以带看下 toolCallingManager.executeToolCalls 的源码，会自动追加助手消息到上下文中：
4、定义任务终止工具
千万别忘了，我们还需要定义一个终止工具，让智能体可以自行决定任务结束。
1. 定义终止工具 TerminateTool
在 tools 包下新建 TerminateTool：
public class TerminateTool {
@Tool(description = """
Terminate the interaction when the request is met OR if the assistant cannot proceed further with the task.
"When you have finished all the tasks, call this tool to end the work.
""")
public String doTerminate() {
return "任务结束";
}
}
2. 注册终止工具
修改 ToolRegistration，注册中止工具：
TerminateTool terminateTool = new TerminateTool();
return ToolCallbacks.from(
fileOperationTool,
webSearchTool,
webScrapingTool,
resourceDownloadTool,
terminalOperationTool,
pdfGenerationTool,
terminateTool
);
3. 完善 act 方法
我们补充 act 方法的逻辑，当调用了终止工具时，修改 agent 的状态为 “已结束”：
// 当前工具调用的结果
ToolResponseMessage toolResponseMessage = (ToolResponseMessage) CollUtil.getLast(toolExecutionResult.conversationHistory());
String results = toolResponseMessage.getResponses().stream()
.map(response -> "工具 " + response.name() + " 完成了它的任务！结果: " + response.responseData())
.collect(Collectors.joining("\n"));
// 判断是否调用了终止工具
boolean terminateToolCalled = toolResponseMessage.getResponses().stream()
.anyMatch(response -> "doTerminate".equals(response.name()));
if (terminateToolCalled) {
setState(AgentState.FINISHED);
}
log.info(results);
return results;
5、开发 LeiManus 类
YuManus 是可以直接提供给其他方法调用的 AI 超级智能体实例，继承自 ToolCallAgent，需要给智能体设置各种参数，比如对话客户端 chatClient、工具调用列表等。
代码如下：
@Component
public class LeiManus extends ToolCallAgent {
public LeiManus(ToolCallback[] allTools, ChatModel dashscopeChatModel) {
super(allTools);
this.setName("yuManus");
String SYSTEM_PROMPT = """
You are YuManus, 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.
""";
this.setSystemPrompt(SYSTEM_PROMPT);
String NEXT_STEP_PROMPT = """
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.
If you want to stop the interaction at any point, use the `terminate` tool/function call.
""";
this.setNextStepPrompt(NEXT_STEP_PROMPT);
this.setMaxSteps(20);
// 初始化客户端
ChatClient chatClient = ChatClient.builder(dashscopeChatModel)
.defaultAdvisors(new MyLoggerAdvisor())
.build();
this.setChatClient(chatClient);
}
}
测试智能体
编写单元测试类，让超级智能体完成一个较为复杂的任务：
@SpringBootTest
class LeiManusTest {
@Resource
private LeiManus LeiManus;
@Test
void run() {
String userPrompt = """
我的另一半居住在上海杨浦区，请帮我找到 5 公里内合适的约会地点，
并结合一些网络图片，制定一份详细的约会计划，
并以 PDF 格式输出""";
String answer = LeiManus.run(userPrompt);
Assertions.assertNotNull(answer);
}
}
以 Debug 模式运行，能够查看到 AI 每一步的思考过程：
执行完成后，可以看到每一步的执行结果，并且可以在本地查看到已下载的图片和生成的 PDF。
注意，由于 AI 是存在随机性的，执行的步骤数也不固定。
经过测试大家会发现，拥有自主规划能力的超级智能体虽然能够完成复杂的任务，但缺点是 非常浪费 tokens，并且可能陷入无限循环。 所以还是要按需使用。
感兴趣的同学可以参考 OpenManus 的实现来进一步优化超级智能体，比如防止陷入无限循环、能够和用户交互等。
六、扩展知识
智能体工作流
当我们面对复杂任务时，单一智能体可能无法满足需求。因此智能体工作流（Agent Workflow）应运而生，通过简单的编排，允许多个专业智能体协同工作，各司其职。
智能体工作流编排的精髓在于 将复杂任务分解为连贯的节点链，每个节点由最适合的智能体处理，节点间通过条件路由灵活连接，形成一个高效、可靠的执行网络。
Anthropic 曾经在一篇 研究报告《Building effective agents》 中提到了多种不同的智能体工作模式，大家需要了解每种模式的特点和适用场景。
智能体工作模式
1. Prompt Chaining 提示链工作流
Prompt Chaining 是最常见的智能体工作流模式之一。它的核心思想是将一个复杂任务拆解为一系列有序的子任务，每一步由 LLM 处理前一步的输出，逐步推进任务完成。
比如在内容生成场景中，可以先让模型生成大纲，再根据大纲生成详细内容，最后进行润色和校对。每一步都可以插入校验和中间检查，确保流程正确、输出更精准。
这种模式结构清晰，易于调试，非常适合任务可以被自然分解为多个阶段的场景。
2. Routing 路由分流工作流
Routing 工作流模式则更像是一个智能的路由器。系统会根据输入内容的类型或特征，将任务分发给最合适的下游智能体或处理流程。非常适合多样化输入和多种处理策略的场景。
比如在客服系统中，可以将常见问题、退款请求、技术支持等分流到不同的处理模块；在多模型系统中，可以将简单问题分配给小模型，复杂问题交给大模型。这样既提高了处理效率，也保证了每类问题都能得到最优解答。
3. Parallelization 并行化工作流
在 Parallelization 并行化模式下，任务会被拆分为多个可以并行处理的子任务，最后聚合各自的结果。
比如在代码安全审查场景中，可以让多个智能体分别对同一段代码进行安全审查，最后 “投票” 决定是否有问题。又比如在处理长文档时，可以将文档分段，每段由不同智能体并行总结。这种模式可以显著提升处理速度，并通过 “投票” 机制提升结果的准确度。
4. Orchestrator-Workers 协调器-执行者工作流
对于复杂的任务、参与任务的智能体增多时，我们可以引入一位 “管理者”，会根据任务动态拆解出多个子任务，并将这些子任务分配给多个 “工人” 智能体，最后再整合所有工人的结果。这种中央协调机制提高了复杂系统的整体效率，适合任务结构不确定、需要动态分解的复杂场景。
5. Evaluator-Optimizer 评估-优化循环工作流
Evaluator-Optimizer 模式模拟了人类 “写 => 评 => 改” 的过程。一个智能体负责生成初步结果，另一个智能体负责评估和反馈，二者循环迭代优化输出。
举个例子，在机器翻译场景中，先由翻译智能体输出，再由评审智能体给出改进建议，反复迭代直到达到满意的质量。这种模式特别适合需要多轮打磨和质量提升的任务。
智能体工作流编排框架
如果要实现上述工作流，我们可以采用 Activiti、LiteFlow 这种工作流框架，把智能体处理当做工作流的一个节点。也可以直接使用专门的 AI 智能体工作流编排框架，比如 LangGraph 和 Spring AI Alibaba Graph。这两个框架大家简单了解一下，直到怎么组合工作流即可，实际学习成本都不高。
1. LangGraph
LangGraph 是 LangChain 团队开发的前沿工作流编排框架，专为大语言模型应用设计，是构建复杂 AI 系统的首选工具。LangGraph 的核心理念是将智能体工作流表示为状态转换图，每个节点可以是函数、智能体或子工作流，边则代表状态转换条件。
相比于传统的工作流框架，LangGraph 的独特之处在于它对大语言模型工作流的深度优化。框架提供了丰富的功能来满足大模型应用开发场景，比如动态分支和并行执行、思维链支持、对话管理、内置监控与可视化等。开发者可以轻松实现复杂的推理路径，让模型在需要时进行反思、规划和纠错。
前面讲到的几种工作流模式，以及普通的工具调用代理场景，使用 LangGraph 都能轻松完成，具体用法和代码示例可以参考 官方文档。
下面给大家一段简单的工作流代码示例：
from langgraph.graph import StateGraph
from typing import TypedDict, List, Annotated
# 定义工作流状态
class AgentState(TypedDict):
messages: List[dict]
task: str
context: dict
# 创建状态图
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("understand", understand_task)
workflow.add_node("research", research_information)
workflow.add_node("analyze", analyze_data)
workflow.add_node("generate", generate_response)
# 定义边和条件
workflow.add_edge("understand", "research")
workflow.add_conditional_edges(
"research",
decide_next_step,
{
"needs_more_info": "research",
"ready_for_analysis": "analyze",
"skip_analysis": "generate"
}
)
workflow.add_edge("analyze", "generate")
# 设置起始节点
workflow.set_entry_point("understand")
这段代码利用 LangGraph 框架定义了一个多阶段的智能体工作流：
- 首先通过 AgentState 描述了每一步的状态结构；
- 然后用 StateGraph 创建了一个状态流转图，依次添加了 “理解任务”、“信息检索”、“数据分析” 和 “结果生成” 四个节点，并通过有向边和条件分支将它们串联起来，实现了流程的跳转和循环；
- 最后指定工作流从 “理解任务” 节点开始，整体上实现了一个可自动决策、可循环的智能体任务处理流程。
2. Spring AI Alibaba Graph
目前 Spring AI 官方还没有提供工作流编排能力，但是国内的 Spring AI Alibaba 已经提供了工作流编排框架 Spring AI Alibaba Graph。
虽然目前还处于 Demo 阶段，已经支持了：
- 基础工作流的编排
- ReAct Agent 模式
- Orchestrator-Workers 协调器-工作者模式
像我们前面自主开发的 ReAct 模式的 Manus 智能体，使用 Spring AI Alibaba Graph，几行代码就搞定了：
ReactAgent reactAgent = ReactAgent.builder()
.name("React Agent Demo")
.prompt("请完成接下来用户输入给你的任务。")
.chatClient(chatClient)
.resolver(resolver)
.maxIterations(10)
.build();
reactAgent.invoke(Map.of("messages", new UserMessage(query)))
让我们通过一个实际案例来感受像 Spring AI Alibaba Graph 这种框架的作用，比如编写一个数据分析和可视化报告生成任务的工作流，示例代码如下（仅供参考）：
public AIGraph<String, String> dataAnalysisWorkflow(
PlanningAgent planningAgent,
DataCollectionAgent dataCollectionAgent,
AnalysisAgent analysisAgent,
VisualizationAgent visualizationAgent,
ReportingAgent reportingAgent) {
return AIGraph.builder(String.class, String.class)
// 任务规划节点
.addNode("planning", input -> planningAgent.run(input), Map.of(
"success", "data_collection"
))
// 数据收集节点
.addNode("data_collection", input -> dataCollectionAgent.run(input), Map.of(
"complete", "analysis",
"incomplete", "planning"
))
// 数据分析节点
.addNode("analysis", input -> analysisAgent.run(input), Map.of(
"success", "visualization",
"failure", "data_collection"
))
// 数据可视化节点
.addNode("visualization", input -> visualizationAgent.run(input), Map.of(
"success", "reporting"
))
// 报告生成节点
.addNode("reporting", input -> reportingAgent.run(input), Map.of())
.withStartAt("planning")
.build();
}
这个工作流展现了数据分析的完整生命周期：
- 从规划智能体制定分析策略开始；
- 到数据收集智能体获取必要数据；
- 然后是分析智能体处理这些数据；
- 接着由可视化智能体创建直观图表；
- 最后由报告智能体整合所有发现生成完整报告。
除了上面这些工作流开发框架外，之前我们还学习过一种可视化的智能体工作流开发方法 —— 使用 AI 大模型应用开发平台创建工作流应用。
目前很多公司都在用这种方法构建复杂的智能体工作流，确实很方便~
OWL 框架
前面提到过 AI 智能体依赖工具调用能力来拓展能力边界，这里再给大家分享一个很实用的开源项目 OWL。
OWL (Optimized Workforce Learning) 是由 CAMEL-AI 团队开源的一款面向 多智能体协作与真实世界任务自动化 的前沿框架。
通过 OWL，AI 智能体可以执行终端命令、访问网络资源、运行各种编程语言的代码、使用各种开发工具等等。
它的主要特点：
1. 多智能体协作：
- OWL 支持多个智能体之间的动态交互与协作，能够模拟真实团队协作场景，适合解决需要多角色、多技能配合的复杂任务。
2. 丰富的工具集成：
- 内置了浏览器自动化、代码执行、文档处理、音视频分析、搜索引擎等多种工具包，支持多模态任务（如网页操作、图片/视频/音频分析等）。
3. MCP 协议支持：
- 通过 MCP，OWL 能够与外部工具和数据源标准化对接，极大扩展了智能体的能力边界。
4. 可定制与易用性：
- 用户可以根据实际需求灵活配置和组合所需工具，优化性能和资源消耗。
5. Web 可视化界面：
- 提供基于 Gradio 的本地 Web UI，支持模型选择、环境变量管理、交互历史查看等功能，方便开发和调试。
其实 OWL 的本质就是利用 AI 来增强传统的自动化办公场景，像自动化数据分析和处理、自动化网页信息检索，都能够用 OWL 轻松完成。
下图是官方介绍的 Demo，OWL 可以自动打开我们的浏览器访问网页，执行滚动、点击、打字操作，并且实时抓取网页信息：
A2A 协议
什么是 A2A 协议？
A2A（Agent to Agent）也是最近很热门的一个概念，简单来说，A2A 协议 就是为 “智能体之间如何直接交流和协作” 制定的一套标准。
A2A 协议的核心，是让每个智能体都能像 “网络节点” 一样，拥有自己的身份、能力描述和通信接口。它不仅规定了消息的格式和传递方式，还包括了身份认证、能力发现、任务委托、结果回传等机制。这样一来，智能体之间就可以像人类团队一样，互相打招呼、询问对方能做什么、请求协助。
可以把 A2A 类比为智能体世界里的 HTTP协议：
- HTTP 协议让全球不同服务器和电脑之间能够交换数据，A2A 协议则是让不同厂商、不同平台、不同能力的智能体能够像团队成员一样互相理解、协作和分工。
- 如果说 HTTP 协议让互联网成为了一个开放、互联的世界，那么 A2A 协议则让智能体世界变得开放、协作和高效。
A2A 协议的应用场景
A2A 协议的应用非常广泛，总结下来 4 个字就是 开放互联：
- 比如在自动驾驶领域，不同车辆的智能体可以实时交换路况信息，协同避障和规划路线；
- 在制造车间，生产线上的各类机器人智能体可以根据任务动态分工，互相补位；
- 在金融风控、智能客服等场景，不同的智能体可以根据自身专长协作处理复杂业务流程。
我们还可以大胆想象，未来开发者可以像调用云服务一样，按需租用或组合不同的智能体服务，甚至实现智能体之间的自动交易和结算。目前其实就有很多智能体平台，只不过智能体之间的连接协作甚少。
A2A 和 MCP 协议的区别
虽然 A2A 和 MCP 都算是协议（或者标准），但二者存在本质上的区别。
-
MCP 协议是 智能体和外部工具之间的标准，它规定了智能体如何安全、规范地调用外部的数据库、搜索引擎、代码执行等工具资源。你可以把 MCP 理解为 “智能体-工具” 的 HTTP 协议。
-
而 A2A 协议则是 智能体之间的通信协议。它更像是让不同的 AI 角色之间可以直接对话、协作和分工。
从安全角度看，MCP 和 A2A 处理的是不同层面的安全问题：
- MCP 的安全关注点：主要集中在单个智能体与工具之间的安全交互，主要防范的是工具滥用和提示词注入攻击。
- A2A 的安全关注点：更关注智能体网络中的身份认证、授权和信任链。A2A 需要解决“我怎么知道我在和谁通信”、“这个智能体有权限请求这项任务吗”、“如何防止恶意智能体窃取或篡改任务数据” 等问题。（这些问题也是 HTTP 协议需要考虑的）
显然，A2A 面临的安全挑战更加复杂，因为它处理的是跨网络、跨平台、多方协作的场景。
对于一个成熟的智能体系统，可能会同时运用 MCP 和 A2A：
- MCP 负责某个智能体内部调用工具完成任务；
- A2A 负责智能体之间协同完成任务；
目前 A2A 协议的落地并不成熟，暂时大家也不必关注如何进行 A2A 的开发，了解它是什么、有什么用就足够了。
感兴趣的同学，可以阅读 这篇文章 来了解 A2A 协议的实现细节和交互流程。
扩展思路
1. 给智能体添加循环检测和处理机制，防止智能体陷入无限循环。
可以参考 OpenManus 源码实现，示例代码如下：
private int duplicateThreshold = 2;
/**
* 处理陷入循环的状态
*/
protected void handleStuckState() {
String stuckPrompt = "观察到重复响应。考虑新策略，避免重复已尝试过的无效路径。";
this.nextStepPrompt = stuckPrompt + "\n" + (this.nextStepPrompt != null ? this.nextStepPrompt : "");
System.out.println("Agent detected stuck state. Added prompt: " + stuckPrompt);
}
/**
* 检查代理是否陷入循环
*
* @return 是否陷入循环
*/
protected boolean isStuck() {
List<Message> messages = this.memory.getMessages();
if (messages.size() < 2) {
return false;
}
Message lastMessage = messages.get(messages.size() - 1);
if (lastMessage.getContent() == null || lastMessage.getContent().isEmpty()) {
return false;
}
// 计算重复内容出现次数
int duplicateCount = 0;
for (int i = messages.size() - 2; i >= 0; i--) {
Message msg = messages.get(i);
if (msg.getRole() == Role.ASSISTANT &&
lastMessage.getContent().equals(msg.getContent())) {
duplicateCount++;
}
}
return duplicateCount >= this.duplicateThreshold;
}
// 每一步 step 执行完都要检查是否陷入循环
if (isStuck()) {
handleStuckState();
}
2. 智能体支持交互式执行，可以向用户询问信息或获取反馈，从而优化任务的完成效果。
实现思路可以参考 OpenManus，专门定义一个 AskHuman 工具，让 AI 自主决定什么时候需要寻求人类帮助：
当然也可以通过编写 Prompt 实现，比如 Prompt 中提到 “如果你认为需要向人类寻求帮助，输出结果中需要包含 [ASK_USER]{寻求帮助的具体问题}”，并且检查每一条 AI 给出的消息，如果包含了 ASK_USER 标记，就通过系统控制台和用户交互。
示例代码如下：
public boolean think() {
boolean shouldAct = super.think();
// 获取最新的助手消息
Message lastMessage = getMessageList().get(getMessageList().size() - 1);
if (lastMessage instanceof AssistantMessage) {
String content = lastMessage.getContent();
// 检查是否包含向用户询问的标记
if (content.contains("[ASK_USER]")) {
// 提取问题
String question = content.substring(content.indexOf("[ASK_USER]") + 10);
// 向用户输出问题
System.out.println("智能体需要你的帮助: " + question);
// 获取用户输入
Scanner scanner = new Scanner(System.in);
String userAnswer = scanner.nextLine();
// 添加用户回答到消息列表
UserMessage userResponse = new UserMessage("用户回答: " + userAnswer);
getMessageList().add(userResponse);
// 需要继续思考
return true;
}
}
return shouldAct;
}
3. 开发支持 MCP 协议的超级智能体
思路：其实利用 Spring AI，只需要把 MCP 服务中的工具提取出来变成工具列表，就可以复用已有的 ToolCallAgent 了
4. 自行开发一个特定领域的超级智能体，可以直接继承 ToolCallAgent 来实现；
5. 参考 OpenManus 的源码，实现更多功能，比如 “规划并执行” 模式的智能体工作流（参考 flow/planning.py）
重点实现
1. 完成 AI 超级智能体的开发
2. 理解 AI 超级智能体的实现原理和架构设计，要求能够口述讲清楚
3. 尝试自主阅读 OpenManus 的源码，思考如何进一步优化我们的程序，建议多利用 AI 工具辅助学习理解
火山引擎开发者社区是火山引擎打造的AI技术生态平台，聚焦Agent与大模型开发，提供豆包系列模型（图像/视频/视觉）、智能分析与会话工具，并配套评测集、动手实验室及行业案例库。社区通过技术沙龙、挑战赛等活动促进开发者成长，新用户可领50万Tokens权益，助力构建智能应用。
更多推荐
15
0- 0
已为社区贡献6条内容
所有评论(0)