篷汎山1. 项目概述
在人工智能领域，多Agent协作系统正变得越来越重要。OpenManus项目展示了一个典型的"先规划再执行"的多Agent协作框架，其核心创新点在于将规划(Planning)和执行(Execution)两个阶段明确分离，并通过状态驱动的方式实现智能工作流。
这个架构解决了多Agent协作中的几个关键痛点：
- 任务分配：如何将复杂任务合理分配给不同特长的Agent
- 执行顺序：确定任务步骤的先后关系
- 状态管理：跟踪每个步骤的执行进度
- 错误处理：当某个步骤失败时如何恢复
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 核心架构设计
2.1 整体流程
OpenManus的Planning架构包含三个主要阶段：
- 计划生成阶段：由LLM根据用户请求和可用Agent能力，动态生成执行计划
- 状态驱动执行阶段：按照计划步骤顺序执行，每个步骤由最适合的Agent处理
- 执行总结阶段：在所有步骤完成后，生成最终结果报告
这种设计将"决策"和"执行"分离，使得系统更加灵活和可扩展。
2.2 主要组件
系统包含三个核心模块：
- PlanningTool：计划管理器，负责计划的CRUD操作和状态跟踪
- PlanningFlow：流程协调器，控制整个执行流程
- BaseFlow：抽象基类，定义所有Flow类型的基本接口
3. 详细实现解析
3.1 PlanningTool设计
PlanningTool是整个架构的基础设施，提供完整的计划管理功能：
python复制class PlanningTool:
def __init__(self):
self.plans = {} # 存储所有计划
async def execute(self, command: str, **kwargs):
if command == "create":
return await self._create_plan(**kwargs)
elif command == "update":
return await self._update_plan(**kwargs)
# 其他命令处理...
关键数据结构：
python复制{
"plan_id": "unique_id",
"title": "计划标题",
"steps": ["[AGENT1] 步骤1", "[AGENT2] 步骤2"],
"step_statuses": ["completed", "in_progress"],
"step_notes": ["已完成", "处理中"]
}
状态流转设计：
code复制not_started → in_progress → completed
↓
blocked
3.2 PlanningFlow实现
PlanningFlow是系统的"大脑"，协调整个执行流程：
python复制class PlanningFlow(BaseFlow):
def __init__(self, agents: Dict[str, BaseAgent]):
self.agents = agents
self.planning_tool = PlanningTool()
self.current_step_index = None
async def execute(self, input_text: str) -> str:
# 阶段1: 创建计划
await self._create_initial_plan(input_text)
# 阶段2: 执行步骤
while True:
step_index, step_info = await self._get_current_step_info()
if step_index is None:
break
executor = self._get_executor(step_info)
await self._execute_step(executor, step_info)
# 阶段3: 生成总结
return await self._finalize_plan()
3.3 Agent路由机制
系统通过正则表达式提取步骤中的Agent标记，实现灵活路由：
python复制def _get_executor(self, step_info: dict) -> BaseAgent:
# 提取步骤中的Agent标记
step_text = step_info["text"]
match = re.search(r"\[([A-Z_]+)\]", step_text)
if match:
agent_key = match.group(1).lower()
if agent_key in self.agents:
return self.agents[agent_key]
# 默认返回第一个可用Agent
return next(iter(self.agents.values()))
4. 关键设计决策分析
4.1 动态计划生成
与传统硬编码计划不同，OpenManus的计划完全由LLM动态生成：
python复制async def _create_initial_plan(self, request: str):
# 构建系统提示
system_msg = f"""
你是计划助手，可以创建清晰、可执行的计划。
可用Agent: {[a.description for a in self.agents.values()]}
创建步骤时请使用格式'[AGENT_NAME] 步骤描述'
"""
# 调用LLM生成计划
response = await self.llm.ask_tool(
messages=[{"role": "user", "content": request}],
tools=[self.planning_tool.to_param()]
)
# 处理LLM响应
if response.tool_calls:
await self.planning_tool.execute(**response.tool_calls[0].arguments)
这种设计使得系统能够：
- 根据任务复杂度自动调整计划粒度
- 灵活适应不同类型的任务
- 充分利用LLM的语义理解能力
4.2 状态驱动执行
执行阶段通过检查计划状态决定下一步操作：
python复制async def _get_current_step_info(self):
plan = await self.planning_tool.get(self.active_plan_id)
for i, status in enumerate(plan["step_statuses"]):
if status in ["not_started", "in_progress"]:
# 更新状态为in_progress
await self.planning_tool.mark_step(
plan_id=self.active_plan_id,
step_index=i,
status="in_progress"
)
return i, plan["steps"][i]
return None, None # 没有未完成步骤
这种设计避免了维护复杂的执行队列，通过状态自然推导执行顺序。
5. 扩展性与灵活性
5.1 添加新Agent
要添加新Agent类型非常简单：
- 实现新的Agent类
- 将其添加到agents字典
- LLM在生成计划时会自动考虑新Agent的能力
python复制agents = {
"manus": ManusAgent(),
"data_analysis": DataAnalysisAgent(),
"new_agent": NewAgent() # 新添加的Agent
}
5.2 自定义路由逻辑
可以通过重写_get_executor方法实现自定义路由：
python复制def _get_executor(self, step_info):
step_text = step_info["text"]
# 自定义路由逻辑
if "数据分析" in step_text:
return self.agents["data_analysis"]
elif "生成报告" in step_text:
return self.agents["manus"]
return super()._get_executor(step_info)
6. 实际应用示例
6.1 数据分析任务
用户输入："帮我分析sales.csv并生成报告"
系统执行流程：
-
LLM生成计划：
- [DATA_ANALYSIS] 读取并分析sales.csv
- [MANUS] 生成可视化图表
- [MANUS] 输出分析报告
-
按顺序执行每个步骤
-
生成最终总结报告
6.2 复杂任务处理
用户输入："帮我分析销售数据，找出异常点，预测下季度趋势，并生成PPT报告"
LLM会自动生成更详细的计划：
- [DATA_ANALYSIS] 数据清洗和预处理
- [DATA_ANALYSIS] 异常检测分析
- [DATA_ANALYSIS] 趋势预测建模
- [MANUS] 生成分析图表
- [MANUS] 编写分析报告
- [MANUS] 制作PPT演示文稿
7. 性能优化建议
7.1 并行执行优化
对于没有依赖关系的步骤，可以修改执行器实现并行处理：
python复制async def _execute_parallel(self):
plan = await self.planning_tool.get(self.active_plan_id)
# 找出所有可并行执行的步骤
parallel_steps = self._find_parallel_steps(plan)
# 并行执行
results = await asyncio.gather(
*[self._execute_step(step) for step in parallel_steps]
)
return results
7.2 计划缓存
对常见任务类型可以缓存生成的计划：
python复制class CachedPlanningTool(PlanningTool):
def __init__(self):
super().__init__()
self.plan_cache = {}
async def create_plan(self, request: str):
# 生成缓存键
cache_key = self._generate_cache_key(request)
if cache_key in self.plan_cache:
return self.plan_cache[cache_key]
# 调用父类方法生成新计划
plan = await super().create_plan(request)
self.plan_cache[cache_key] = plan
return plan
8. 错误处理与容错
8.1 步骤失败处理
当某个步骤执行失败时，系统提供多种恢复策略：
- 自动重试：对暂时性错误自动重试
- 跳过步骤：标记为失败后继续执行后续步骤
- 人工干预：暂停流程等待人工处理
python复制async def _execute_step(self, executor, step_info):
retry_count = 0
max_retries = 3
while retry_count < max_retries:
try:
result = await executor.run(step_info)
await self._mark_step_completed()
return result
except TemporaryError as e:
retry_count += 1
await asyncio.sleep(1)
except CriticalError as e:
await self._mark_step_failed()
raise
await self._mark_step_failed()
return None
8.2 计划生成回退
当LLM无法生成有效计划时，使用默认计划：
python复制async def _create_initial_plan(self, request):
try:
# 尝试LLM生成计划
return await self._create_plan_with_llm(request)
except Exception as e:
logger.warning(f"LLM计划生成失败，使用默认计划: {e}")
return await self.planning_tool.create(
title=f"默认计划: {request[:50]}",
steps=["分析请求", "执行任务", "验证结果"]
)
9. 监控与日志
完善的监控体系对系统运维至关重要：
python复制class Monitoring:
def __init__(self):
self.metrics = {
"plan_created": Counter(),
"step_completed": Counter(),
"errors": Counter()
}
def log_plan_created(self, plan_id):
self.metrics["plan_created"].inc()
def log_step_completed(self, step_index):
self.metrics["step_completed"].inc()
def log_error(self, error_type):
self.metrics["errors"].inc(error_type)
可以监控的关键指标：
- 计划生成成功率
- 步骤执行平均时间
- 错误类型分布
- Agent负载情况
10. 安全考虑
10.1 输入验证
对所有外部输入进行严格验证：
python复制def validate_step(step: str) -> bool:
# 检查步骤格式
if not isinstance(step, str):
return False
# 检查长度限制
if len(step) > 500:
return False
# 检查是否有非法字符
if re.search(r"[<>{}]", step):
return False
return True
10.2 Agent权限控制
实现基于角色的访问控制：
python复制class AgentRBAC:
def __init__(self):
self.roles = {
"data_analysis": ["read_data", "analyze"],
"manus": ["generate_report", "create_chart"]
}
def check_permission(self, agent_name: str, action: str) -> bool:
return action in self.roles.get(agent_name, [])
11. 测试策略
11.1 单元测试
对核心组件进行严格测试：
python复制def test_planning_tool():
tool = PlanningTool()
# 测试计划创建
plan_id = tool.create(title="测试", steps=["步骤1"])
assert plan_id is not None
# 测试状态更新
tool.mark_step(plan_id, 0, "completed")
assert tool.get(plan_id).step_statuses[0] == "completed"
11.2 集成测试
测试整个工作流：
python复制async def test_full_workflow():
agents = {"manus": MockAgent(), "data": MockAgent()}
flow = PlanningFlow(agents)
# 执行测试任务
result = await flow.execute("测试任务")
# 验证结果
assert "完成" in result
assert flow.planning_tool.get(flow.active_plan_id).all_steps_completed
12. 部署建议
12.1 容器化部署
使用Docker打包应用：
dockerfile复制FROM python:3.9
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["python", "run_flow.py"]
12.2 水平扩展
对高负载场景，可以水平扩展PlanningFlow实例：
bash复制# 启动多个实例
docker-compose up --scale planning_flow=3
13. 架构演进方向
未来可能的改进方向：
- 动态Agent注册：支持运行时添加/移除Agent
- 计划优化：引入强化学习优化计划生成
- 跨流程协作：多个PlanningFlow实例协同工作
- 可视化监控：提供图形化的执行监控界面
14. 经验总结
在实际实现这类系统时，有几个关键经验：
- 保持组件职责单一：这大大降低了系统复杂度
- 设计时考虑扩展性：预留足够的扩展点
- 完善的错误处理：多Agent系统出错是常态
- 详尽的日志记录：对调试和监控至关重要
- 渐进式复杂度：从简单版本开始，逐步增加功能
15. 常见问题解决
15.1 LLM生成计划质量不高
解决方案：
- 优化系统提示词
- 提供更详细的Agent描述
- 添加示例计划
15.2 步骤执行卡住
排查步骤：
- 检查Agent是否响应
- 查看步骤状态是否正确更新
- 检查网络连接
- 查看日志中的错误信息
15.3 性能瓶颈
优化方向：
- 引入步骤并行执行
- 缓存常用计划
- 优化Agent实现
这套架构在实际项目中展现了良好的灵活性和扩展性，特别适合需要多步骤协作的复杂AI任务。通过将规划与执行分离，系统能够充分利用LLM的规划能力，同时保持执行部分的高效和可靠。