4. 多Agent规划与执行
任务目标
本案例构建一个基于LangGraph的多Agent协作系统：一个规划Agent接收复杂任务描述，将其分解为有序的子任务，并根据每个子任务的性质分配给不同的执行Agent（代码Agent与写作Agent）。系统最终汇总各执行Agent的结果，形成完整输出。
环境依赖
需要安装以下Python库：
pip install langgraph langchain-core langchain-并在环境变量中配置_。本案例使用-mini模型，若使用其他模型请相应调整。
核心实现
系统包含三个节点：planner负责任务分解与分配，executor根据分配结果路由到对应执行Agent，aggregator汇总结果。执行Agent通过工具函数实现，规划Agent通过结构化输出定义子任务格式。
import json
from typing import TypedDict, List
from langgraph.graph import StateGraph, END
from langchain_ import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
# ---------- 状态定义 ----------
class AgentState(TypedDict):
task: str
plan: List[dict] # [{id, description, assignee}]
results: dict # {agent_name: output}
final_output: str
# ---------- 执行Agent（工具函数） ----------
def code_agent(task_desc: str) -> str:
"""模拟代码执行Agent：返回代码片段"""
return f"# 代码方案\n```python\ndef solve():\n # 针对: {task_desc}\n pass\n```"
def writing_agent(task_desc: str) -> str:
"""模拟写作Agent：返回说明文档"""
return f"## 说明\n针对「{task_desc}」的详细步骤与注意事项。"
# ---------- 节点函数 ----------
def planner(state: AgentState) -> dict:
llm = ChatOpenAI(model="-mini", temperature=0)
prompt = f"""将以下任务分解为最多3个子任务，每个子任务必须明确指派给code_agent或writing_agent。
任务：{state['task']}
输出JSON数组，格式：
[{{"description": "子任务描述", "assignee": "code_agent 或 writing_agent"}}]"""
response = llm.invoke([HumanMessage(content=prompt)])
# 解析LLM返回的JSON
content = response.content.strip()
if content.startswith("```json"):
content = content[7:-3]
plan = json.loads(content)
return {"plan": plan}
def executor(state: AgentState) -> dict:
results = {}
for item in state["plan"]:
desc = item["description"]
agent_name = item["assignee"]
if agent_name == "code_agent":
results[agent_name] = code_agent(desc)
elif agent_name == "writing_agent":
results[agent_name] = writing_agent(desc)
else:
results[agent_name] = f"未知Agent类型: {agent_name}"
return {"results": results}
def aggregator(state: AgentState) -> dict:
parts = []
for item in state["plan"]:
agent = item["assignee"]
desc = item["description"]
output = state["results"].get(agent, "无输出")
parts.append(f"### 子任务: {desc}\n执行者: {agent}\n\n{output}")
final = f"# 任务完成报告\n\n原始任务: {state['task']}\n\n" + "\n\n".join(parts)
return {"final_output": final}
# ---------- 构建图 ----------
def build_graph():
g = StateGraph(AgentState)
g.add_node("planner", planner)
g.add_node("executor", executor)
g.add_node("aggregator", aggregator)
g.set_entry_point("planner")
g.add_edge("planner", "executor")
g.add_edge("executor", "aggregator")
g.add_edge("aggregator", END)
return g.compile()
# ---------- 运行入口 ----------
if __name__ == "__main__":
app = build_graph()
# 测试任务
test_task = "开发一个Python函数计算斐波那契数列，并编写使用说明文档"
result = app.invoke({"task": test_task, "plan": [], "results": {}, "final_output": ""})
print(result["final_output"])运行与验证
运行方式：直接执行脚本：
python multi_agent_planner.py验证方法：观察输出是否满足以下条件：
- 输出包含
任务完成报告标题及原始任务描述； - 规划结果将任务拆分为至少2个子任务，且分别指派给
code_agent和writing_agent； - 每个子任务下方有对应Agent生成的代码或文档内容；
- 若LLM返回格式错误，程序会抛出JSON解析异常——此时可重试或调整提示词。
预期输出结构示例（实际内容取决于LLM规划结果）：
# 任务完成报告
原始任务: 开发一个Python函数计算斐波那契数列，并编写使用说明文档
### 子任务: 实现斐波那契函数
执行者: code_agent
# 代码方案
```python
def solve():
# 针对: 实现斐波那契函数
pass
```
### 子任务: 编写文档
执行者: writing_agent
## 说明
针对「编写文档」的详细步骤与注意事项。关键说明
- 规划Agent使用LLM进行任务分解，因此每次运行产生的子任务划分可能不同，但必须遵循JSON格式约束；
- 执行Agent在本案例中为模拟实现，实际工程中可替换为真正的代码生成模型调用或文档编写API；
- 若需增加Agent类型，只需在
executor节点中扩展分支，并在规划提示词中声明新的可选值。