LangGraph 是一个用于构建有状态、多智能体 LLM 应用的框架。它实现状态机与有向图进行编排,支持复杂工作流、持久状态管理、人在回路以及时间旅行调试。其关键创新是将智能体协调从顺序链转变为带持久状态、条件分支和生产级调试能力的循环图。
适用场景
适合使用 LangGraph 的情况:需要多智能体协调;需要复杂状态管理;需要人在回路工作流(审批门、评审);需要调试/可观测性(时间旅行、重放);需要基于输出进行条件分支;构建生产级智能体系统;需要跨会话状态持久化。
不适合使用 LangGraph 的情况:简单单智能体任务;无需状态持久化;原型/实验阶段(可使用简单链);团队缺乏图/状态机专业知识;无状态请求-响应模式。
快速开始
以下代码展示了基本用法:
from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator
1. 定义状态模式
class AgentState(TypedDict): messages: Annotated[list, operator.add] current_step: str
2. 创建图
workflow = StateGraph(AgentState)
3. 添加节点(智能体)
def researcher(state): return {"messages": ["Research complete"], "current_step": "research"}
def writer(state): return {"messages": ["Article written"], "current_step": "writing"}
workflow.add_node("researcher", researcher) workflow.add_node("writer", writer)
4. 添加边(转换)
workflow.add_edge("researcher", "writer") workflow.add_edge("writer", END)
5. 设置入口点并编译
workflow.set_entry_point("researcher") app = workflow.compile()
6. 执行
result = app.invoke({"messages": [], "current_step": "start"}) print(result)
核心概念
StateGraph:表示智能体有向图的基本构建块,带有共享状态。关键属性包括:节点(转换状态的智能体函数)、边(节点之间的转换,静态或条件)、状态(节点之间传递的共享数据结构)、入口点(执行的起始节点)、END(表示完成的终止节点)。
节点:接收当前状态并返回状态更新的函数。节点行为:接收完整状态作为输入;返回部分状态(仅需更新的字段);可以是同步或异步函数;可以调用 LLM、调用 API、运行计算。
边:定义节点之间的转换。静态边为直接转换,例如 workflow.add_edge("researcher", "writer")。条件边通过路由函数决定下一个节点,例如:
def should_continue(state: AgentState) -> str: last_message = state["messages"][-1] if "APPROVED" in last_message: return END elif "NEEDS_REVISION" in last_message: return "writer" else: return "reviewer"
workflow.add_conditional_edges( "reviewer", should_continue, {END: END, "writer": "writer", "reviewer": "reviewer"} )
图构建完整示例
以下示例展示研究、写作、评审的循环工作流:
from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator from langchain_anthropic import ChatAnthropic
class ResearchState(TypedDict): topic: str research_notes: Annotated[list, operator.add] draft: str revision_count: int approved: bool
llm = ChatAnthropic(model="claude-sonnet-4")
def research_node(state: ResearchState) -> dict: topic = state["topic"] prompt = f"Research key points about: {topic}" response = llm.invoke(prompt) return {"research_notes": [response.content]}
def write_node(state: ResearchState) -> dict: notes = "\n".join(state["research_notes"]) prompt = f"Write article based on:\n{notes}" response = llm.invoke(prompt) return {"draft": response.content, "revision_count": state.get("revision_count", 0)}
def review_node(state: ResearchState) -> dict: draft = state["draft"] prompt = f"Review this draft. Reply APPROVED or NEEDS_REVISION:\n{draft}" response = llm.invoke(prompt) approved = "APPROVED" in response.content return { "approved": approved, "revision_count": state["revision_count"] + 1, "research_notes": [f"Review feedback: {response.content}"] }
def should_continue_writing(state: ResearchState) -> str: if state["approved"]: return END elif state["revision_count"] >= 3: return END else: return "writer"
workflow = StateGraph(ResearchState) workflow.add_node("researcher", research_node) workflow.add_node("writer", write_node) workflow.add_node("reviewer", review_node) workflow.add_edge("researcher", "writer") workflow.add_edge("writer", "reviewer") workflow.add_conditional_edges("reviewer", should_continue_writing, {END: END, "writer": "writer"}) workflow.set_entry_point("researcher") app = workflow.compile() result = app.invoke({"topic": "AI Safety", "research_notes": [], "draft": "", "revision_count": 0, "approved": False}) print(f"Final draft: {result['draft']}") print(f"Revisions: {result['revision_count']}")
状态管理
使用 TypedDict 定义状态模式,支持简单字段、带 reducer 的列表、自定义 reducer 的字典以及可选字段。状态 reducer 控制状态更新如何合并,内置 reducer 包括:Annotated[list, operator.add](追加到列表)、Annotated[set, operator.or_](集合的并集)、Annotated[int, operator.add](整数求和)。也可自定义 reducer,例如深度合并字典。节点返回部分状态,仅包含需要更新的字段,其他字段保持不变。
条件路由
基本路由函数根据状态路由到不同节点,例如根据质量分数路由到 publish、review 或 revise。动态路由可使用 LLM 决定下一步,例如让 LLM 返回 researcher、coder、writer 或 FINISH。多条件路由可基于多个条件(如错误、批准、迭代次数)进行优先级路由。
多智能体模式
监督者模式:一个监督者协调多个专业智能体。示例中定义了 ResearchAgent、CodingAgent 和 SupervisorAgent,监督者决定下一个智能体,各智能体执行后循环回监督者。
分层多智能体:嵌套监督者模式,包含子团队。示例中创建后端团队和前端团队子图,顶层协调器将团队子图作为节点,并支持并行执行。
Swarm 模式(2025):动态智能体交接与协作决策。示例中智能体可根据需要动态交接给其他智能体,例如研究智能体判断是否需要实现,从而交接给编码智能体或写作智能体。
限制与注意事项
资料中未提供关于收费、价格、平台支持、具体性能效果或安装命令的详细信息。使用前需注意:不适合简单单智能体任务、无需状态持久化、原型实验阶段、团队缺乏图/状态机专业知识或无状态请求-响应模式。