LangGraph入门：构建有状态的多Agent应用
在大型语言模型（LLM）应用开发中，单一Agent的能力边界正被不断突破。无论是代码生成、客服对话还是复杂数据分析，单Agent往往难以应对多步骤、多轮交互、状态依赖的复杂场景。LangGraph——一个基于LangChain生态的图结构编排框架——应运而生，它将Agent交互从线性流程升级为有向图，为构建有状态、可循环、多角色协作的应用提供了标准范式。
从Chain到Graph：为什么需要图结构？
传统LangChain应用多采用Chain或Sequential Chain模式，即“A → B → C”的线性流水线。这种模式在简单任务中有效，但遇到以下场景时力不从心：
- **条件分支**：根据中间结果决定下一步调用哪个Agent - **循环迭代**：当Agent输出质量不达标时，需要自我修正或重新规划 - **多Agent协作**：多个Agent之间需要来回传递信息、共享上下文 - **持久化状态**：长时间对话或任务需要维护跨步骤的全局状态
LangGraph的核心创新在于：将应用逻辑建模为一个**有向图（Directed Graph）**，节点（Node）代表计算单元（Agent、工具调用、函数），边（Edge）决定数据流动方向。更重要的是，每个图实例维护一个**状态对象（State）**，所有节点共享该状态，从而实现了真正的“有状态”编排。
快速上手：构建第一个LangGraph应用
1. 环境准备与基本概念
首先安装LangGraph（需要Python 3.10+）：
```bash pip install langgraph langchain-openai ```
LangGraph的核心组件包括：
- **StateGraph**：构建图的主类，需要定义一个状态Schema - **Node**：图中的节点，可以是任意可调用对象（函数、Agent、Runnable） - **Edge**：连接节点的边，支持条件边（Conditional Edge） - **State**：在节点间传递的共享对象，支持TypedDict或Pydantic模型
2. 一个简单的多步问答Agent
假设我们需要一个Agent：先分析用户问题，再决定是调用搜索工具还是直接回答。传统方法需要写if-else逻辑，而LangGraph可以优雅地实现：
```python from typing import Literal, TypedDict from langgraph.graph import StateGraph, END from langgraph.checkpoint import MemorySaver
class AgentState(TypedDict): messages: list next_action: str
def classify_node(state: AgentState): """分析用户意图，决定下一步"""
模拟LLM分类结果 if "搜索" in state["messages"][-1]["content"]: return {"next_action": "search"} return {"next_action": "answer"}
def search_node(state: AgentState): response = f"搜索结果为：{state['messages'][-1]['content']}" return {"messages": [{"role": "assistant", "content": response}]}
def answer_node(state: AgentState): return {"messages": [{"role": "assistant", "content": "直接回答：你好！"}]}
构建图 builder = StateGraph(AgentState) builder.add_node("classify", classify_node) builder.add_node("search", search_node) builder.add_node("answer", answer_node)
builder.set_entry_point("classify")
条件边：根据next_action决定走向 builder.add_conditional_edges( "classify", lambda state: state["next_action"], {"search": "search", "answer": "answer"} ) builder.add_edge("search", END) builder.add_edge("answer", END)
添加持久化检查点 memory = MemorySaver() graph = builder.compile(checkpointer=memory) ```
这个例子展示了条件分支的核心用法：`classify`节点输出`next_action`字段，条件边根据该字段路由到不同节点。`MemorySaver`实现了状态持久化，使得多次调用可以共享上下文。
实战案例：多Agent协作的代码审查系统
为了展示LangGraph的深度能力，我们构建一个代码审查系统：包含“代码分析Agent”、“安全审查Agent”和“最终报告Agent”，三个Agent需要循环协作直至达成一致。
系统设计
状态对象设计如下：
```python from pydantic import BaseModel, Field
class CodeReviewState(BaseModel): code: str = Field(description="待审查代码") analysis_result: str = "" security_issues: list = [] review_round: int = 0 is_complete: bool = False final_report: str = "" ```
节点实现
```python def code_analysis_node(state: CodeReviewState): """分析代码质量、复杂度"""
模拟分析结果 issues = ["函数过长", "缺少注释"] return {"analysis_result": "\n".join(issues), "review_round": state.review_round + 1}
def security_check_node(state: CodeReviewState): """检查安全漏洞"""
模拟安全检查 vulns = ["SQL注入风险"] return {"security_issues": vulns, "review_round": state.review_round + 1}
def report_node(state: CodeReviewState): """生成最终报告""" report = f"""代码审查报告（第{state.review_round}轮） 代码质量：{state.analysis_result} 安全风险：{', '.join(state.security_issues) if state.security_issues else '无'} """ return {"final_report": report, "is_complete": True}
def should_continue(state: CodeReviewState): """判断是否继续审查""" if state.review_round >= 3 or state.is_complete: return "report" if not state.security_issues: return "report" return "continue" ```
图编排
```python builder = StateGraph(CodeReviewState) builder.add_node("analyze", code_analysis_node) builder.add_node("security", security_check_node) builder.add_node("report", report_node)
builder.set_entry_point("analyze") builder.add_edge("analyze", "security")
条件循环：安全检查完决定是否继续
builder.add_conditional_edges(
"security",
should_continue,
{
"continue": "analyze", 循环回分析节点
"report": "report"
}
)
builder.add_edge("report", END)
graph = builder.compile() ```
这个案例的关键设计在于：
1. **循环状态管理**：通过`review_round`计数器控制循环次数，避免无限循环 2. **共享状态**：所有Agent共享同一个`CodeReviewState`对象，可以读取/修改任意字段 3. **条件终止**：`should_continue`函数基于当前状态决定是否进入下一轮
进阶技巧：状态管理与调试
检查点（Checkpointing）
LangGraph内置了强大的检查点机制，使得应用可以“暂停-恢复”：
```python
在任意节点设置检查点
config = {"configurable": {"thread_id": "review_001"}}
for event in graph.stream({"code": "print('hello')"}, config):
print(event)
可以在此处手动中断，后续调用graph.resume()继续
```
这在需要人工介入的Agent场景（如审批流程）中极为有用。
状态Schema最佳实践
- 使用Pydantic模型而非TypedDict，以获得类型校验和默认值 - 为关键字段添加`Field`描述，便于文档生成和调试 - 避免在状态中存储大量二进制数据，建议使用外部存储（如Redis）的引用
与主流框架的对比
| 特性 | LangGraph | AutoGen | CrewAI | |------|-----------|---------|--------| | 状态管理 | 原生支持，全局共享 | 依赖ConversableAgent | 依赖Process | | 循环支持 | 原生图结构 | 需自定义回调 | 有限支持 | | 检查点 | 内置MemorySaver | 无 | 无 | | 学习曲线 | 中等 | 低 | 低 |
LangGraph在需要精细控制状态流、支持复杂循环的场景中优势明显，特别适合构建生产级的长时间运行Agent。
结语
LangGraph将图计算思想引入LLM应���开发，解决了传统线性编排在状态管理、循环控制、多Agent协作方面的痛点。从简单的条件分支到复杂的多轮审查系统，它提供了一套统一且可扩展的范式。随着LangChain生态的成熟，LangGraph正在成为构建企业级多Agent应用的标准选择。建议读者从简单图开始，逐步引入条件边、循环和检查点，最终掌握这个强大的编排工具。