LangGraph 实战用图状态机构建多步骤 Agent 工作流的完整指南一、深度引言与场景痛点刚接触 Agent 开发的时候几乎所有教程都在教你这样写把用户输入喂给 LLMLLM 决定调用哪个工具拿到工具结果后再喂给 LLM最后输出答案。这条链清晰、可控、容易调试。直到你遇到这样的需求用户说帮我查一下公司上个月的销售数据如果同比下降超过 10%就生成一份分析报告发给所有部门负责人。用链式思维写的话你会得到一段近乎不可维护的代码if-else 嵌套层层叠叠每个分支里再套 try-except失败重试的逻辑散落在各处。更要命的是你没法表达先并行查询多个数据源然后汇总结果做判断这种拓扑结构。链式调用的本质是线性的而真实世界的 Agent 工作流是图状的——有分支、有汇聚、有循环、有条件跳转。这就是 LangGraph 要解决的问题。它把 Agent 工作流建模成一个有向图节点是处理单元调用 LLM、执行工具、做判断边定义了数据如何在节点间流转。图的拓扑结构天然支持条件路由、并行执行和循环迭代这些在链式模型中都需要大量样板代码才能模拟。二、底层机制与原理深度剖析LangGraph 的核心是 StateGraph。理解它只需要搞懂三个概念状态State、节点Node、边Edge。状态是一个可序列化的字典贯穿整张图的执行生命周期每个节点都能读写它。节点是纯函数签名为(state) - state_update接收当前状态、返回状态的部分更新。边分两种普通边直接连接两个节点条件边根据状态中的某个字段动态决定跳转到哪个节点。一个智能数据分析 Agent 可以按这样的工作流组织parse_intent节点负责理解用户意图然后条件边route根据意图类型分发查数据走查询分支直接要报告走报告分支闲聊走回复分支。查询分支中validate条件边检查数据有效性失败的进入retry_fetch循环——这是链式调用很难优雅表达的结构。check_anomaly条件边判断分析结果有异常才生成报告正常则生成摘要。有向拓扑能把这类业务逻辑表达得清清楚楚。关键点在于StateGraph 里的边不只是下一步做什么它还隐式定义了依赖关系。当多条边汇聚到同一个节点时LangGraph 会先并行执行所有上游节点如果它们之间没有依赖再把结果合并后传给汇合节点。这种并行性不需要手动用 asyncio.gather框架自动处理。三、生产级代码实现import asyncio import logging from typing import Annotated, Any, Literal from langgraph.graph import StateGraph, END from langgraph.graph.state import CompiledStateGraph from langgraph.checkpoint.memory import MemorySaver from pydantic import BaseModel, Field from typing_extensions import TypedDict logger logging.getLogger(__name__) class WorkflowState(TypedDict): 工作流共享状态 user_input: str intent: str query_result: dict[str, Any] | None retry_count: int analysis_result: str final_response: str error: str | None class IntentResult(BaseModel): intent: Literal[data_query, report, chitchat] Field( description用户意图分类 ) confidence: float Field(ge0.0, le1.0) async def parse_intent(state: WorkflowState) - dict[str, Any]: 节点解析用户意图 try: # 实际场景中调用 LLM 做意图分类 text state[user_input].lower() if 查 in text or 数据 in text: intent data_query elif 报告 in text or 分析 in text: intent report else: intent chitchat return {intent: intent, error: None} except Exception as e: logger.exception(Intent parsing failed) return {intent: chitchat, error: str(e)} def route_by_intent(state: WorkflowState) - Literal[fetch_data, generate_report, chat_reply]: 条件边根据意图路由到不同处理分支 routes { data_query: fetch_data, report: generate_report, chitchat: chat_reply, } return routes.get(state[intent], chat_reply) async def fetch_data(state: WorkflowState) - dict[str, Any]: 节点查询数据源 try: # 模拟异步数据查询实际可能是 HTTP/DB 调用 await asyncio.sleep(0.5) data {sales: [100, 85, 120], trend: down} return {query_result: data, error: None} except asyncio.TimeoutError: logger.warning(Data fetch timeout) return {error: timeout} def validate_data(state: WorkflowState) - Literal[analyze, retry_fetch]: 条件边校验查询结果有效性 if state[query_result] is not None and len(state[query_result]) 0: return analyze return retry_fetch async def retry_fetch(state: WorkflowState) - dict[str, Any]: 节点重试查询带退避和上限 retry_count state.get(retry_count, 0) 1 if retry_count 3: return {retry_count: retry_count, error: max_retries_exceeded} backoff min(2 ** retry_count, 30) await asyncio.sleep(backoff) # 重试逻辑... return {retry_count: retry_count, query_result: None} def check_retry_limit(state: WorkflowState) - Literal[fetch_data, fallback_handler]: 条件边检查重试次数 return fallback_handler if state[retry_count] 3 else fetch_data async def analyze_data(state: WorkflowState) - dict[str, Any]: 节点分析查询结果 data state[query_result] or {} sales data.get(sales, []) avg sum(sales) / len(sales) if sales else 0 prev_avg 120 # 模拟上月平均值 decline (prev_avg - avg) / prev_avg if prev_avg else 0 return {analysis_result: f月均销售额 {avg:.1f}同比下降 {decline:.0%}} def check_anomaly(state: WorkflowState) - Literal[generate_report, generate_summary]: 条件边检测数据异常 if 下降 in state.get(analysis_result, ) and 10% in state.get(analysis_result, ): return generate_report return generate_summary def build_agent_graph() - CompiledStateGraph: 构建 LangGraph 工作流 builder StateGraph(WorkflowState) # 注册节点 builder.add_node(parse_intent, parse_intent) builder.add_node(fetch_data, fetch_data) builder.add_node(retry_fetch, retry_fetch) builder.add_node(analyze, analyze_data) builder.add_node(generate_report, lambda s: {final_response: f报告{s.get(analysis_result, )}}) builder.add_node(generate_summary, lambda s: {final_response: f摘要{s.get(analysis_result, )}数据正常}) builder.add_node(chat_reply, lambda s: {final_response: 你好请问需要什么帮助}) builder.add_node(fallback_handler, lambda s: {final_response: 数据查询失败请稍后重试}) # 注册边 builder.set_entry_point(parse_intent) builder.add_conditional_edges(parse_intent, route_by_intent, { fetch_data: fetch_data, generate_report: generate_report, chat_reply: chat_reply, }) builder.add_conditional_edges(fetch_data, validate_data, { analyze: analyze, retry_fetch: retry_fetch, }) builder.add_conditional_edges(retry_fetch, check_retry_limit, { fetch_data: fetch_data, fallback_handler: fallback_handler, }) builder.add_conditional_edges(analyze, check_anomaly, { generate_report: generate_report, generate_summary: generate_summary, }) builder.add_edge(generate_report, END) builder.add_edge(generate_summary, END) builder.add_edge(chat_reply, END) builder.add_edge(fallback_handler, END) # 使用内存 checkpointer 支持状态持久化 memory MemorySaver() return builder.compile(checkpointermemory) async def main() - None: graph build_agent_graph() test_cases [ 查一下上个月的销售数据, 帮我生成一份周报, 今天天气怎么样, ] for i, user_input in enumerate(test_cases): config {configurable: {thread_id: fsession_{i}}} try: result await graph.ainvoke( {user_input: user_input, retry_count: 0}, configconfig, ) print(f[{user_input}] → {result.get(final_response, no response)}) except Exception as e: logger.exception(Graph execution failed for: %s, user_input) if __name__ __main__: logging.basicConfig(levellogging.INFO) asyncio.run(main())代码里几个设计值得注意。一是MemorySaver作为 checkpointer它把每次图执行的中间状态自动序列化存储——这意味着即使进程重启恢复 thread_id 就能从中断处继续执行不需要自己写状态恢复逻辑。二是条件边函数是纯逻辑判断不涉及 I/O执行速度极快不会成为图的性能瓶颈。三是每个节点都包裹了 try-except节点内的异常不会让整张图崩溃而是通过error字段传递给后续节点做决策。四是retry_fetch用了指数退避策略避免在外部服务恢复瞬间发起洪峰请求。四、边界分析与架构权衡LangGraph 解决了工作流编排的表达力问题但它的代价是调试复杂度。线性链式调用出问题时你顺着代码一行行看就行图结构出问题时你需要同时看节点代码和拓扑结构定位根因的难度陡增。建议在开发阶段开启 LangGraph 的 tracing 功能把每次执行的节点序列记录到 LangSmith 或本地文件。和 LangChain 的 AgentExecutor 相比AgentExecutor 封装了 ReAct 循环的细节开箱即用但定制性差。你要在 AgentExecutor 里加一个查询失败 3 次后换一个搜索引擎的逻辑得深入修改内部实现。LangGraph 从一开始就把控制权交给开发者每一步跳转都是你自己定义的。和自研编排引擎相比LangGraph 省去了维护图执行引擎的成本并行调度、状态序列化、错误传播但对简单线性场景来说它是过度设计。另一个边界是状态序列化。如果 WorkflowState 里放了不可序列化的对象如数据库连接、文件句柄checkpointer 会报错。最佳实践是状态里只放纯数据和引用S3 key、DB row id节点内部自行创建和销毁资源。五、总结LangGraph 用图状态机的方式解决了 Agent 工作流编排的核心问题条件分支、循环重试、并行执行和状态持久化。它的抽象非常简洁——StateGraph Node Edge——但表达能力极强从简单的分支路由到复杂的多层嵌套都能处理。生产落地时重点做好三件事节点异常不扩散、条件边保持纯净、状态只存可序列化数据。对于需要维护跨步骤上下文的多步骤 AgentLangGraph 是目前最成熟的图编排方案。
