2.9 KiB
2.9 KiB
Notes: LangGraph 多智能体研究
核心概念
LangGraph 基础
- StateGraph: 有向无环图(DAG),节点是 Agent/函数,边是流转逻辑
- State: 贯穿整个图流动的状态对象
- Node: 执行单元(可以是 Agent、函数、条件判断)
- Edge: 连接节点的边,支持条件边(conditional edges)
Supervisor + Workers 模式参考
1. LangChain 官方 Supervisor 示例
from langgraph.prebuilt import create_react_agent
from langgraph.graph import StateGraph, END
# 定义 Workers
research_agent = create_react_agent(llm, tools=[search])
coder_agent = create_react_agent(llm, tools=[write_file])
# 定义 Supervisor 节点
def supervisor_node(state):
# LLM 决定下一步调用哪个 Agent
response = llm.with_structured_output(SupervisorOutput).invoke(
[SystemMessage(content=SUPERVISOR_PROMPT)] + state["messages"]
)
return {"next": response.next_agent}
# 构建图
graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("research", research_agent)
graph.add_node("code", coder_agent)
2. 状态定义
from typing import TypedDict, Annotated
import operator
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
task: str
plan: list
results: dict
iteration: int
next: str # 控制下一步流向
3. 条件边实现
def should_continue(state):
if state["iteration"] >= MAX_ITERATIONS:
return "end"
if state.get("task_complete"):
return "end"
return "continue"
graph.add_conditional_edges(
"review",
should_continue,
{
"continue": "supervisor",
"end": END
}
)
设计决策
架构优势
- 清晰的分层: Supervisor 负责任务规划,Workers 负责执行
- 可扩展: 容易添加新的 Worker 类型
- 可控: 迭代次数全局配置
- 灵活: 支持条件分支和循环
需要解决的问题
- Supervisor 如何做规划: 需要设计 prompt 让 LLM 生成任务列表
- 任务队列: 需要支持并行分发多个 Worker
- 共享上下文: 需要设计数据结构在 Agent 间共享状态
- Review 机制: 需要定义检查标准和重试逻辑
关键 Prompt 设计
Supervisor System Prompt
你是一个任务规划专家(Supervisor)。用户的任务是:{task}
请按以下步骤执行:
1. 分析任务需求和约束
2. 将任务分解为可执行的子任务
3. 为每个子任务选择合适的执行 Agent:
- research: 信息搜索和调研
- coder: 代码编写和修改
- review: 结果检查和评审
4. 确定执行顺序和依赖关系
当前任务进度:{progress}
共享上下文:{context}
请输出你的决策,格式如下:
- 需要执行的子任务列表
- 每个任务的执行 Agent
- 任务执行顺序
- 是否需要汇总结果