LangGraph实战：用图编排框架构建可控的AI智能体工作流

最近在尝试把一些零散的 AI 工具串联起来，做成一个能自动处理复杂任务的“智能体”时，我遇到了一个典型问题：流程跑通了，但一旦任务步骤变多，或者需要多个“智能体”协作，代码就迅速变成了一团乱麻。状态管理、步骤跳转、错误处理、循环逻辑……这些原本在传统软件开发里很清晰的概念，在构建 AI 驱动的应用时，却常常需要自己从头“发明轮子”。

这让我开始关注一个叫LangGraph的框架。它不像 LangChain 那样提供海量的工具链，而是专注于一件事：用图（Graph）的方式来定义和管理 AI 的工作流。听起来有点抽象，但它的核心价值非常直接——把一次性的、脆弱的脚本，变成可预测、可调试、可长期运行的“企业级”智能体系统。

很多人一听到“图”和“架构”就觉得头大，认为这是只有大厂架构师才需要关心的东西。这其实是个误解。LangGraph 解决的不是“高并发”或“分布式”那种宏观架构问题，它解决的是每个开发者在构建 AI 应用时都会遇到的微观工作流失控问题。比如，一个客服机器人，到底是该直接回答，还是去查知识库，或者转人工？这个决策逻辑和状态流转，用 if-else 硬写会非常痛苦，而 LangGraph 提供了一种声明式的、可视化的方式来定义它。

所以，这篇文章不会堆砌晦涩的论文术语，而是从一个实践者的角度，带你理解 LangGraph 到底改变了什么。我们会从“为什么需要它”开始，一步步拆解它的核心概念，并通过一个从单智能体到多智能体协作的实战案例，让你看到如何用它把想法变成稳定、可维护的代码。你会发现，它的学习曲线远比想象中平缓，而带来的清晰度和可控性提升是立竿见影的。

1. 先搞清楚：LangGraph 解决的不是“智能”问题，而是“流程”问题

在深入代码之前，我们必须先建立一个关键认知：LangGraph 本身不生产“智能”。它不提供大语言模型（LLM），也不直接处理自然语言理解。它的核心定位是一个“编排框架”。

你可以把它想象成一个超级增强版的if-else和while循环，专门为 AI 应用设计。在传统的编程中，流程控制是确定性的：条件 A 成立，就执行 B。但在 AI 应用中，很多“条件”是 LLM 根据上下文“思考”后产生的非确定性结果。比如，“用户这个问题我能否回答？是否需要更多信息？”。

1.1 传统 AI 脚本的典型困境：状态散落与逻辑胶水

在没有专门框架之前，我们如何构建一个多步骤的 AI 应用？通常的写法是这样的：

# 一种常见的、脆弱的实现方式 def handle_user_query(query, conversation_history): # 步骤1：判断意图 intent = llm_classify_intent(query) if intent == "qa": # 步骤2：检索知识 docs = retrieve_knowledge(query) # 步骤3：生成回答 answer = llm_generate_answer(query, docs) return answer elif intent == "chitchat": # 步骤2：闲聊处理 answer = llm_chitchat(query, conversation_history) return answer elif intent == "need_more_info": # 步骤2：追问澄清 clarifying_question = llm_generate_clarification(query) # 步骤3：等待用户回复... 状态怎么保存？ # 我们需要把当前状态（如：等待 clarification）和中间结果存到某处 save_state(user_id, {"waiting_for": "clarification", "original_query": query}) return clarifying_question # ... 更多的 elif

这段代码的问题显而易见：

1. 状态管理混乱：当流程需要暂停等待用户输入（如追问）时，当前的“进度”和中间数据（original_query）必须手动保存到数据库或缓存中，与业务逻辑高度耦合。
2. 逻辑嵌套深：每个分支可能又衍生出新的分支，代码可读性和可维护性急剧下降。
3. 难以调试和可视化：你无法一眼看出整个应用有哪些节点、哪些路径。当流程复杂后，理清逻辑依赖关系变得异常困难。
4. 缺乏通用模式：错误重试、循环控制、并行执行、人工审核（Human-in-the-loop）等常见需求，都需要重新发明一套实现。

LangGraph 的出现，正是为了系统性地解决这些“流程胶水”问题。它让你能用“画图”的方式定义应用，图中的节点是功能单元（调用 LLM、执行工具、条件判断），边是控制流。

1.2 LangGraph 的核心抽象：图、状态、边

理解 LangGraph，只需要抓住三个核心概念：

1. State（状态）：这是一个贯穿整个工作流的、共享的上下文对象。它通常是一个字典（或 Pydantic 模型），包含了所有节点需要读取和写入的数据。比如{"messages": [...], "user_query": "...", "retrieved_docs": [...], "next_step": "..."}。LangGraph 负责在节点间传递和持久化这个状态。
2. Node（节点）：一个执行单元，通常是一个函数。它接收当前的State，执行一些操作（如调用 LLM、运行工具），然后返回一个对 State 的更新。例如，一个“检索”节点，会读取State["query"]，调用检索器，然后将结果写入State["documents"]。
3. Edge（边）：决定工作流下一步该走到哪个节点。边分为两种：
   • 条件边（Conditional Edge）：根据State中的某个值（通常是 LLM 或函数判断的结果）来决定下一个节点。这是实现分支逻辑的关键。
   • 普通边：无条件地指向下一个节点。

通过组合这些元素，你可以定义出循环（节点A -> 条件边 -> 根据条件返回节点A或结束）、分支（节点 -> 条件边 -> 节点B 或 节点C）和并行（高级特性）等复杂流程。

关键理解：LangGraph 将你的应用逻辑从“代码的行文顺序”中解放出来，变成了“对一张图的定义”。这张图就是你的应用架构图，它清晰、可视、易于修改和推理。

2. 从零开始：用 LangGraph 构建你的第一个智能体工作流

理论说得再多，不如亲手搭建一个。让我们从一个最简单的例子开始：一个能进行多轮对话，并在必要时主动查询天气的聊天助手。

这个智能体的逻辑是：

1. 接收用户输入。
2. 判断用户是否在问天气（意图识别）。
3. 如果是，调用天气查询工具，并整合信息回复。
4. 如果不是，直接让 LLM 生成回复。
5. 等待下一轮用户输入。

2.1 环境准备与状态定义

首先，安装必要的包。这里我们使用 OpenAI 的模型（你也可以替换为 Ollama 本地模型）。

pip install langgraph langchain-openai langchain-community

接下来，定义工作流的状态。我们使用TypedDict来获得更好的类型提示。

from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END # 1. 定义状态结构 class AgentState(TypedDict): # 完整的对话历史 messages: Annotated[List[str], operator.add] # 用户的最新输入 latest_user_input: str # LLM 判断的意图 intent: str # 天气查询的结果（如果有） weather_info: str

Annotated[List[str], operator.add]是 LangGraph 的一个精妙设计。它表示messages字段是一个列表，当多个节点对它进行修改时，默认使用operator.add（即列表的extend操作）来合并更新，而不是覆盖。这非常适合记录对话历史。

2.2 创建节点：功能单元的实现

节点就是普通的 Python 函数，它接收State并返回一个包含更新字段的字典。

from langchain_openai import ChatOpenAI import os os.environ["OPENAI_API_KEY"] = "你的密钥" llm = ChatOpenAI(model="gpt-3.5-turbo") # 节点1：接收用户输入 def receive_input(state: AgentState): # 在实际应用中，这里可能从API接收数据 user_input = input("用户: ") # 模拟用户输入 return {"latest_user_input": user_input, "messages": [f"用户: {user_input}"]} # 节点2：判断意图 def classify_intent(state: AgentState): prompt = f""" 请判断用户最新消息的意图。消息是：{state['latest_user_input']} 可选意图：['闲聊', '查询天气', '其他']。 只返回意图关键词。 """ response = llm.invoke(prompt) intent = response.content.strip() return {"intent": intent} # 节点3：查询天气（模拟工具调用） def call_weather_tool(state: AgentState): # 这里模拟一个工具调用，真实情况可能调用API city = "北京" # 简单起见，假设查询北京 weather = "晴，25摄氏度" return {"weather_info": f"{city}的天气是：{weather}"} # 节点4：生成最终回复 def generate_response(state: AgentState): if state['intent'] == '查询天气': # 结合天气信息生成回复 prompt = f"用户问：{state['latest_user_input']}。已知天气信息：{state['weather_info']}。请生成友好回复。" else: # 普通闲聊回复 prompt = f"请根据对话历史回复用户最新消息：{state['latest_user_input']}。对话历史：{state['messages'][-5:]}" response = llm.invoke(prompt) ai_message = f"助手: {response.content}" return {"messages": [ai_message]}

2.3 组装成图：定义工作流逻辑

这是 LangGraph 最核心的部分，我们将节点和边连接起来，形成完整的工作流。

# 初始化图构建器 workflow = StateGraph(AgentState) # 2. 添加节点 workflow.add_node("接收输入", receive_input) workflow.add_node("意图识别", classify_intent) workflow.add_node("查询天气", call_weather_tool) workflow.add_node("生成回复", generate_response) # 3. 设置入口点 workflow.set_entry_point("接收输入") # 4. 添加边（定义流程逻辑） workflow.add_edge("接收输入", "意图识别") # 接收输入后一定进行意图识别 # 从“意图识别”到下一个节点，需要条件判断 def route_by_intent(state: AgentState): # 根据意图，决定下一个节点 if state['intent'] == '查询天气': return "查询天气" else: # 闲聊或其他意图，直接去生成回复 return "生成回复" workflow.add_conditional_edges( "意图识别", # 源节点 route_by_intent, # 路由判断函数 { "查询天气": "查询天气", # 如果函数返回“查询天气”，则跳转到“查询天气”节点 "生成回复": "生成回复", # 如果函数返回“生成回复”，则跳转到“生成回复”节点 } ) # 5. 添加普通边 workflow.add_edge("查询天气", "生成回复") # 查询完天气后，去生成回复 workflow.add_edge("生成回复", END) # 生成回复后，工作流结束 # 6. 编译图 app = workflow.compile()

现在，一张清晰的“图”就定义好了：接收输入 -> 意图识别 -> (查询天气 -> 生成回复 | 生成回复) -> END。

2.4 运行与可视化

你可以像调用函数一样运行这个工作流，初始状态是一个空字典。

# 运行工作流 initial_state = {"messages": [], "latest_user_input": "", "intent": "", "weather_info": ""} final_state = app.invoke(initial_state) print("最终对话历史:", final_state["messages"])

LangGraph 的一个强大功能是可视化。你可以将图导出为 PNG 图片，直观地看到整个流程。

from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果环境不支持，可以输出Mermaid文本到支持的工具查看 print(app.get_graph().draw_mermaid())

这张图就是你应用的活文档，任何新成员都能快速理解业务逻辑的全貌。

3. 进阶实战：构建多智能体协作系统

单智能体工作流只是开始。LangGraph 真正的威力在于优雅地编排多个智能体（或称为“角色”）协同工作，解决更复杂的任务。例如，一个内容创作系统，可能包含“策划”、“撰稿”、“校对”三个智能体。

让我们设计一个“多智能体审阅系统”：

• Writer（写手）：负责根据主题生成初稿。
• Critic（批评家）：负责评审初稿，提出修改意见。
• Editor（编辑）：负责综合初稿和意见，输出最终稿。 流程是：Writer -> Critic -> Editor。Critic 可以要求 Writer 重写，形成循环。

3.1 定义多智能体状态与节点

状态需要容纳多个智能体的输出。

from typing import TypedDict, List, Annotated, Literal import operator class MultiAgentState(TypedDict): topic: str # 写作主题 draft: str # 当前草稿 critique: str # 批评意见 final_output: str # 最终输出 # 一个标志位，控制流程走向 needs_revision: Annotated[Literal["yes", "no"], operator.add] # 记录所有步骤的日志 process_log: Annotated[List[str], operator.add] # 初始化LLM (可以使用不同配置的LLM代表不同角色) llm = ChatOpenAI(model="gpt-4") def writer_node(state: MultiAgentState): """写手节点：生成初稿""" prompt = f"你是一位专业写手。请根据以下主题撰写一篇短文初稿：{state['topic']}" response = llm.invoke(prompt) log_entry = f"Writer 生成了初稿。" return {"draft": response.content, "process_log": [log_entry]} def critic_node(state: MultiAgentState): """批评家节点：评审初稿""" prompt = f"""你是一位严厉的批评家。请评审以下文章初稿，指出其在逻辑、事实、文笔上的具体问题，并提出修改建议。 初稿：{state['draft']} 请直接输出批评意见。""" response = llm.invoke(prompt) # 简单判断：如果批评意见超过一定长度或包含“重写”等关键词，则要求修改 needs_revision = "yes" if "重写" in response.content or len(response.content) > 100 else "no" log_entry = f"Critic 给出了评审意见，要求重写：{needs_revision}。" return {"critique": response.content, "needs_revision": [needs_revision], "process_log": [log_entry]} def editor_node(state: MultiAgentState): """编辑节点：综合草稿和意见，产出终稿""" prompt = f"""你是一位资深编辑。请综合以下初稿和批评意见，输出一份修改后的最终稿。 初稿：{state['draft']} 批评意见：{state['critique']} 最终稿：""" response = llm.invoke(prompt) log_entry = f"Editor 产出了最终稿。" return {"final_output": response.content, "process_log": [log_entry]}

3.2 实现带循环的图逻辑

这里的关键在于，Critic 节点后需要一个条件边，根据needs_revision的值决定是返回 Writer 重写，还是继续到 Editor。

from langgraph.graph import StateGraph, END # 构建图 workflow = StateGraph(MultiAgentState) # 添加节点 workflow.add_node("Writer", writer_node) workflow.add_node("Critic", critic_node) workflow.add_node("Editor", editor_node) # 设置流程 workflow.set_entry_point("Writer") workflow.add_edge("Writer", "Critic") # 定义条件路由函数 def decide_after_critique(state: MultiAgentState): # 取最近一次 needs_revision 的值 if state.get('needs_revision') and state['needs_revision'][-1] == "yes": return "rewrite" # 需要重写 else: return "finalize" # 可以定稿 # 添加条件边 workflow.add_conditional_edges( "Critic", decide_after_critique, { "rewrite": "Writer", # 返回 Writer 节点，形成循环 "finalize": "Editor", } ) workflow.add_edge("Editor", END) # 编译 app = workflow.compile()

3.3 运行与观察循环

现在运行这个多智能体系统。我们设置一个初始主题。

initial_state = { "topic": "人工智能对未来工作的影响", "draft": "", "critique": "", "final_output": "", "needs_revision": [], "process_log": [] } # 运行，并设置一个最大迭代次数防止无限循环 for i in range(5): # 最多循环5次 result = app.invoke(initial_state) print(f"\n=== 第{i+1}轮运行 ===") print(f"流程日志: {result['process_log']}") print(f"是否需要重写: {result.get('needs_revision', [])}") if result.get('final_output'): print(f"最终输出: {result['final_output'][:200]}...") # 预览前200字符 break # 将本轮结果作为下一轮的初始状态（模拟持续运行） initial_state = result else: print("达到最大循环次数，流程可能陷入循环。")

通过输出日志，你可以清晰地看到Writer -> Critic -> (Writer) -> Critic -> Editor的完整协作过程。这种循环和条件跳转逻辑，如果用传统代码编写，会非常复杂且难以维护，而在 LangGraph 中，它只是一张清晰的图。

4. 从原型到生产：LangGraph 的工程化考量

将 LangGraph 工作流从实验脚本变为可投入生产环境的服务，还需要考虑以下几个关键方面。这些是区分“玩具”和“工具”的核心。

4.1 持久化与检查点：让工作流“暂停”与“恢复”

生产环境中的工作流往往需要运行很长时间（如等待用户回复、等待外部 API 回调），或者需要应对服务重启。LangGraph 的Checkpoint机制解决了这个问题。

它允许你将工作流的完整状态（包括执行到了哪个节点）保存到数据库（如 MySQL、PostgreSQL）。当需要恢复时，可以从上一个检查点继续执行。

from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 # 1. 创建 SQLite 存储（生产环境可用 PostgreSQL 等） conn = sqlite3.connect("checkpoints.db") checkpointer = SqliteSaver(conn) # 2. 在编译图时传入 checkpointer app = workflow.compile(checkpointer=checkpointer) # 3. 运行时会自动创建检查点 config = {"configurable": {"thread_id": "user_123_session_1"}} initial_state = {"topic": "..."} # 第一次调用，创建流程 result1 = app.invoke(initial_state, config=config) print(f"运行ID: {result1['metadata']['run_id']}") # 模拟流程暂停... (例如，等待用户输入) # 4. 根据 thread_id 和 run_id 恢复流程 # 我们可以获取当前状态，或继续调用下一个节点 # app.get_state(config) 可以获取最新状态 # 再次调用 app.invoke() 会从上一个检查点继续

这对于构建需要多轮交互的对话机器人或长耗时批处理任务至关重要。

4.2 稳定性增强：错误处理、超时与重试

任何依赖外部服务（LLM API、工具 API）的节点都可能失败。LangGraph 允许你为节点配置错误处理策略。

from langgraph.graph import StateGraph from langgraph.types import Command, interrupt def unreliable_external_api_node(state): import random if random.random() < 0.3: # 模拟30%失败率 raise ConnectionError("API调用失败！") return {"result": "success"} def handle_api_error(state, error): """错误处理函数""" # 可以记录日志、发送警报、修改状态以尝试其他方案等 print(f"节点执行失败，错误: {error}") # 返回一个指令，例如重试、跳转到其他节点或结束 # return Command(retry=5) # 重试5次（需框架支持） # 或者，跳转到降级处理节点 return {"fallback_result": "used cached data", "next_node": "fallback_node"} # 在定义节点时，可以关联错误处理器（具体语法可能随版本变化，此为概念示意） # workflow.add_node_with_fallback("api_node", unreliable_external_api_node, handle_api_error)

此外，对于 LLM 调用节点，务必设置合理的超时（timeout）和重试逻辑，可以使用 LangChain 的with_retry或with_fallbacks来包装 LLM 调用。

4.3 可观测性与调试：让黑盒变透明

复杂的图工作流调试起来很困难。LangGraph 提供了强大的跟踪（Tracing）和日志功能。

• LangSmith 集成：这是最佳实践。LangSmith 可以可视化记录每一次工作流执行的全过程，包括每个节点的输入输出、耗时、Token 使用量等。这对于性能分析、错误排查和成本监控不可或缺。
• 自定义日志：如我们在状态中定义的process_log字段，在每个节点记录关键动作，是另一种有效的调试手段。
• 图可视化：如前所述，app.get_graph().draw_mermaid()生成的图是理解和沟通架构的最佳工具。

4.4 性能与扩展性思考

• 节点粒度：不要把所有逻辑塞进一个节点。将功能拆分为细粒度的节点（如“意图识别”、“参数提取”、“工具调用”、“结果解析”），有利于复用、测试和并行化。
• 并发执行：LangGraph 支持定义并行节点（通过add_node和add_edge的特殊组合），当多个节点间没有数据依赖时，可以同时运行以提高效率。
• 外部化状态：对于非常大的状态（如长文档），可以考虑不将其完全放在工作流状态中，而是只存储一个引用 ID（如数据库记录 ID），在节点内部按需加载。这可以避免状态对象过大带来的序列化开销。

5. 核心价值与适用边界：什么时候该用 LangGraph？

经过上面的实践，我们可以更清晰地总结 LangGraph 的价值和它最适合的场景。

5.1 LangGraph 带来的核心改变

1. 声明式编排：你将“业务逻辑是什么”与“逻辑如何执行”分离开。你定义图，框架负责调度、状态管理和持久化。这带来了极高的代码可读性和可维护性。
2. 复杂流程的直观建模：循环、条件分支、多角色协作等模式，用图来表示比用过程式代码直观得多。它迫使你思考应用的状态机模型，这是一种更严谨的设计方式。
3. 内置的工程化支持：检查点、可视化、与 LangSmith 的深度集成，这些开箱即用的功能，让你能快速搭建出具备生产就绪特性的应用，而不是从零开始造轮子。
4. 与 LangChain 生态无缝衔接：你可以轻松使用 LangChain 提供的数百个组件（工具、检索器、链）作为图中的节点，享受两个生态的优势。

5.2 最适合 LangGraph 的场景

• 多步骤、有状态的对话系统：客服机器人、游戏 NPC、教学助手等，需要根据历史对话决定下一步行动。
• 多智能体协作系统：如我们演示的写作审阅流程，或模拟辩论、竞拍、协同决策等场景。
• 需要人工干预的流程：内容审核、合同审批等，流程可以在某个节点暂停，等待人工审核（Human-in-the-loop）后再继续。
• 复杂的批处理或数据处理管道：其中某些步骤需要条件判断或循环，例如数据清洗、验证、增强的流水线。

5.3 可能不适用或需要简化的场景

• 极其简单的线性链：如果你的应用只是输入 -> LLM -> 输出，没有分支和状态，直接调用 LLM 或使用简单的 LangChainLCEL可能更轻量。
• 对延迟极其敏感的实时应用：图的调度本身有微小开销。对于要求毫秒级响应的场景，需要仔细评估和性能测试。
• 完全无状态的请求/响应：每次请求都是独立的，不需要记住之前任何交互。

5.4 给实践者的最终建议

1. 从小图开始：不要试图一开始就设计一个包含几十个节点的庞大系统。从一个核心的、3-5个节点的流程开始，验证想法。
2. 状态设计是关键：花时间设计好你的State结构。想清楚哪些数据是全局共享的，哪些是节点局部使用的。良好的状态设计是清晰工作流的基础。
3. 拥抱可视化：养成画图的习惯。在编码前，先用白板或绘图工具画出工作流草图。这能帮你理清逻辑，也便于团队沟通。
4. 优先考虑可观测性：在早期就集成 LangSmith 或建立自己的日志规范。当流程出错时，清晰的执行轨迹能帮你快速定位问题节点。
5. 理解它只是一个编排框架：LangGraph 不解决如何调用 API、如何解析 PDF、如何优化提示词等问题。它解决的是如何将这些能力有序、可靠地组织起来。你的核心竞争力，仍然在于对业务逻辑的深刻理解和对 AI 能力的恰当运用。

LangGraph 的出现，标志着 AI 应用开发从“脚本阶段”向“工程阶段”的演进。它提供的不是某个炫酷的新模型能力，而是一套让复杂 AI 工作流变得可控、可维护、可扩展的工程学工具。当你下次再面对那些交织着判断、循环与协作的 AI 应用需求时，不妨先画一张图，或许你会发现，最复杂的部分，已经有人为你提供了优雅的解决方案。