Langgraph agent framework

何时使用

需要把 AI 工作流变成显式、可调试、可恢复的图时使用 LangGraph。典型场景：

• 单智能体带工具调用（ReAct 模式：思考→调工具→再思考的循环）。
• 多步骤/多智能体共享状态，需要 reducer 控制状态如何合并。
• 按条件路由到不同分支（分类器→不同处理路径）。
• 多轮对话或长任务，需要 checkpointer 持久化与断点续跑。
• 敏感操作前插入人工审批（human-in-the-loop）。
• Map-Reduce 式并行（多主题并发研究后汇总）。

不该用的边界：

• 纯单次 LLM 调用、无状态提示链——直接用 LangChain/SDK 即可，引入图反而过度设计。
• TypeScript 为主的项目——LangGraph 主力是 Python（TS 版本仍较早期）。
• 团队不熟悉图/状态概念且任务简单时，调试成本不划算。

前置：Python 3.9+、langgraph 包、可用的 LLM API（OpenAI/Anthropic 等）、基本异步与图概念。

步骤

构建一个图的固定套路（七步）：

1. 定义状态：用 TypedDict，给需要累加的字段加 Annotated[type, reducer]。
2. 定义工具/节点：节点是函数 state -> dict，只返回要更新的字段（部分更新）。
3. 绑定工具到 LLM：llm.bind_tools(tools)。
4. 定义路由函数：返回下一个节点名或 END。
5. 建图加边：add_node / add_edge / add_conditional_edges，循环用回边实现。
6. 编译：graph.compile()；需要记忆/人审时传 checkpointer= 和 interrupt_before=。
7. 运行：app.invoke(input, config)；带 thread_id 实现对话连续性。

关键约束：

• reducer 决定状态如何合并——add_messages 追加消息（不覆盖），operator.add 累加列表，自定义函数可合并字典；不加 reducer 则字段被覆盖。
• 循环必须有终止条件：路由函数要能返回 END，否则无限循环。
• 生产环境务必配 checkpointer（SQLite 开发、PostgreSQL 生产）。

指令

• 设计状态先想清楚每个字段「累加还是覆盖」，据此选 reducer。
• 节点返回部分更新，不要返回整个 state。
• 工具执行用预置 ToolNode(tools)，不要手写循环。
• 人审用 interrupt_before=["节点名"]；恢复时先 app.update_state(config, {...}) 再 app.invoke(None, config)。
• 并行用 Send("节点名", 子状态) 从条件边扇出，汇总节点用累加 reducer 收集结果。

示例

最小 ReAct 智能体（状态→工具→LLM→路由→建图→运行）：

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # 追加而非覆盖

@tool
def calculator(expression: str) -> str:
    """计算数学表达式。"""
    return str(eval(expression))

tools = [calculator]
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

def agent(state: AgentState) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}

def should_continue(state: AgentState) -> str:
    return "tools" if state["messages"][-1].tool_calls else END

graph = StateGraph(AgentState)
graph.add_node("agent", agent)
graph.add_node("tools", ToolNode(tools))
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, ["tools", END])
graph.add_edge("tools", "agent")  # 回边：工具结果交回 agent
app = graph.compile()

result = app.invoke({"messages": [("user", "25 * 4 等于多少？")]})

持久化 + 多轮记忆（带 thread_id）：

from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string("agent_state.db")  # 生产用 PostgresSaver
app = graph.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "user-123"}}
app.invoke({"messages": [("user", "我叫 Alice")]}, config=config)
app.invoke({"messages": [("user", "我叫什么？")]}, config=config)  # 记得是 Alice

state = app.get_state(config)            # 取当前对话状态
for cp in app.get_state_history(config): # 遍历所有检查点
    print(cp.config, cp.values)

人审（执行前暂停审批）：

app = graph.compile(checkpointer=memory, interrupt_before=["execute"])
config = {"configurable": {"thread_id": "approval-flow"}}
app.invoke({"messages": [("user", "发送报告")]}, config)  # 跑到 execute 前暂停

state = app.get_state(config)
print(state.values["pending_action"])     # 人工审阅待执行动作
app.update_state(config, {"approved": True})
app.invoke(None, config)                   # 批准后恢复执行

并行 Map-Reduce（用 Send 扇出）：

from langgraph.constants import Send

def fanout_topics(state) -> list[Send]:
    return [Send("research", {"topic": t}) for t in state["topics"]]

graph.add_conditional_edges(START, fanout_topics, ["research"])
graph.add_edge("research", "summarize")  # 各分支汇入 summarize

注意事项

• 无限循环：带回边的图（agent↔tools）必须有能返回 END 的路由条件。
• 状态覆盖陷阱：忘记加 reducer 会让后写的更新覆盖前一个节点的结果；多智能体共享状态时尤其注意。
• 调试难：图行为可能不直观，建议接 LangSmith/Langfuse 做可观测与追踪。
• 内存 vs 落盘：SqliteSaver.from_conn_string(":memory:") 进程退出即丢；持久化要给文件路径或用 PostgreSQL。
• 恢复语义：人审恢复时 invoke(None, config)，传 None 表示「不追加新输入、从断点继续」。
• 只在任务确实匹配上述场景时使用；输出需经环境内验证、测试与专家评审，缺关键输入/权限/安全边界时先停下来确认。

互见

• crewai：需要基于角色的多智能体协作时。
• langfuse：需要 LLM 可观测性/追踪时。
• structured-output：需要结构化 JSON 输出时。
• agent-evaluation：需要评测智能体性能时。
• 生产组合：LangGraph 设计图 → 结构化输出规范工具响应 → Langfuse 监控。

---

采编自 sickn33/antigravity-awesome-skills（MIT），原条目上游为 vibeship-spawner-skills（Apache 2.0）。