如何使用 LangGraph 构建具有状态记忆的自动化多智能体工作流

把 LangGraph 扔进生产环境之前，最核心的认知得是：它本质上是一个带状态的循环图，而不是简单的线性 Chain。很多朋友习惯了用 LangChain 这种 DAG（有向无环图）逻辑，导致在做多智能体协作时，一旦需要“反思-修改-再提交”的循环，代码就写成了嵌套地狱。

LangGraph 解决这个问题的关键在于 State 对象的定义。它像一个全局共享的数据库，每个 Node（节点）在执行完后，返回的字典会自动 merge 到当前状态中。

一个典型的“研究员+审核员”双智能体架构，配置逻辑是这样的：

1. 定义状态结构
不要在 State 里塞太多无关变量，只放必要的上下文和消息历史。

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    # add_messages 确保新消息是追加而不是覆盖
    messages: Annotated[list, add_messages] 
    next_step: str
    iteration_count: int

2. 编写条件路由逻辑
这是最容易踩坑的地方。不要在 Node 内部决定去哪里，而要通过 Conditional Edge 决定。

def should_continue(state: AgentState):
    # 只要审核员在消息中提到了 "REVISE"，就跳回研究员节点
    last_message = state["messages"][-1].content
    if "REVISE" in last_message and state["iteration_count"] < 3:
        return "researcher"
    return "end"

3. 构建图并配置 Checkpointer（状态持久化）
如果想让 AI 记住之前的对话，或者在程序崩溃后能从某一步恢复，必须配置 MemorySaver。这是实现“状态记忆”的物理基础。

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END

workflow = StateGraph(AgentState)
# 添加节点和边...
workflow.add_edge(START, "researcher")
workflow.add_conditional_edges("reviewer", should_continue)

# 关键：配置内存检查点
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

# 调用时必须传入 thread_id，否则记忆无法关联
config = {"configurable": {"thread_id": "user_123"}}
app.invoke({"messages": [HumanMessage(content="分析AI芯片趋势")]}, config)

几个实战避坑指南：

状态膨胀问题：如果 messages 列表太长，Token 消耗会飞涨且导致模型丢失重点。建议在 Node 中手动对 messages 进行裁剪，或者引入一个专门的 Summarizer 节点，定期把历史记录压缩成一份简报存入 State。

死循环陷阱：多智能体协作最怕 A 说不对 B 改，B 改完 A 又说不对。在 should_continue 逻辑中必须强行加入 iteration_count 计数器，超过 3-5 次直接强制结束并输出当前最佳结果，否则你的 API 账单会让你心碎。

调试技巧：不要直接运行 app.invoke。建议安装 grandalf 库或者直接用 LangGraph Studio 可视化运行，能清晰看到 Token 在哪个节点被卡住，以及状态在传递过程中哪个字段被意外覆盖了。