1. 先搞清楚 LangGraph 到底解决了什么核心问题如果你正在接触 LangChain并且发现用 Chain 和 Agent 处理稍微复杂一点的业务流程时代码会变得又长又乱状态流转全靠自己手动维护那么 LangGraph 就是你一直在找的那个东西。它不是一个全新的框架而是 LangChain 生态里专门用来编排复杂、有状态、多步骤工作流的库。简单来说LangGraph 把智能体Agent或者任何多步骤的 LLM 应用抽象成一个有向图Graph。图中的每个节点Node是一个执行单元比如调用一次 LLM、执行一个工具、查询数据库边Edge定义了节点之间的流转条件。这样一来你就不用写一堆if-else和全局变量来管理“上一步做了什么”、“下一步该去哪”而是用声明式的方式画出一个工作流蓝图让框架帮你驱动执行。它最核心的价值就两点状态管理内置了一个State对象在整个图执行过程中流转所有节点都能读写它。你不用再操心怎么把上一步的输出塞给下一步的输入。循环与条件分支这是它比普通 Chain 强大的地方。你可以轻松实现“如果 LLM 返回的结果不满足要求就循环回去重新处理”或者“根据工具执行结果决定走 A 分支还是 B 分支”。这让构建真正自主的、能应对复杂场景的智能体成为可能。所以这篇文章不是泛泛而谈概念而是会直接带你从环境搭建开始一步步构建一个能实际运行的智能体并拆解其中每个关键参数和踩坑点。无论你是想用 LangGraph 做客服对话、数据分析流水线还是复杂决策系统这个从零到一的实战流程都是通用的。2. 环境准备与核心概念对齐在开始写代码之前先把环境和核心概念对齐能避免后面很多“为什么跑不起来”的问题。2.1 基础环境搭建LangGraph 是 Python 库建议使用 Python 3.8 及以上版本。首先创建一个干净的虚拟环境这是管理依赖的好习惯。# 创建并激活虚拟环境以 conda 为例 conda create -n langgraph-demo python3.10 conda activate langgraph-demo # 安装核心库 pip install langgraph langchain-openai这里注意几个关键点langgraph是核心框架。langchain-openai是 LangChain 官方维护的 OpenAI 集成包比旧的openai包兼容性更好。我们用它来调用大模型。你很可能还需要其他工具包比如langchain-community包含很多社区工具和加载器、tavily-python网络搜索工具等可以根据项目需要后续安装。一开始保持环境简洁。接下来是配置 API Key。你需要一个 OpenAI 的 API Key或其他 LangGraph 支持模型的 Key。永远不要把 Key 硬编码在代码里提交到版本库。# 在终端中设置环境变量Linux/macOS export OPENAI_API_KEYyour-api-key-here # 在 Windows PowerShell 中 $env:OPENAI_API_KEYyour-api-key-here更推荐的做法是使用.env文件配合python-dotenv管理。2.2 理解 LangGraph 三要素在动手前脑子里要对下面三个核心概念有清晰的认识后面写代码就是把这些概念实例化State状态这是一个贯穿整个工作流的共享数据存储类型通常是TypedDict或Pydantic BaseModel。它定义了工作流中需要传递的所有数据字段比如messages对话历史intermediate_steps工具调用结果question用户问题等。关键每个节点Node的函数接收一个State对象修改它然后返回更新后的State。框架负责传递。Node节点工作流中的一个步骤可以是一个普通函数。这个函数必须接收一个State对象作为参数并返回一个更新后的State对象或包含State的字典。节点里可以干任何事调用 LLM、执行工具、计算、查询数据库等。Edge边决定工作流下一步该执行哪个节点。分为两种条件边Conditional Edge根据State中的某个值动态决定下一个节点。这是实现循环和分支的关键。普通边Normal Edge无条件地指向下一个节点。LangGraph 预定义了END和START两个特殊节点分别代表结束和开始。一个常见的误区以为 LangGraph 是 LangChain 的替代品。其实不是。LangChain 是一个庞大的生态提供了模型 I/O、检索、工具、链等基础组件。LangGraph 是 LangChain 生态中专门负责编排和有状态执行的那一部分。你可以只用 LangGraph 的图编排能力配合其他模型 SDK但通常结合 LangChain 提供的丰富工具和链来构建节点效率最高。3. 实战构建一个具备网络搜索能力的问答智能体我们现在来构建一个经典的智能体它能理解用户问题自主决定是否需要联网搜索来获取最新信息然后结合搜索到的信息给出最终回答。这个流程完美契合了 LangGraph 的“条件分支”和“多步骤”特性。3.1 定义状态与工具首先定义智能体工作流中需要流转的状态。我们使用TypedDict。from typing import TypedDict, List, Annotated import operator from langchain_core.messages import BaseMessage # 定义状态结构 class AgentState(TypedDict): # 对话消息历史这是与LLM交互的核心 messages: Annotated[List[BaseMessage], operator.add] # 用户提出的原始问题 question: str # 是否需要执行搜索的标志位 should_search: boolAnnotated[List[BaseMessage], operator.add]这个写法是 LangGraph 的魔法。它告诉框架对于messages这个字段当多个节点返回的新messages列表时不要替换而是用operator.add即列表的操作将它们合并起来。这确保了对话历史能不断累积。接下来准备工具。这里我们使用一个模拟的搜索工具真实项目中可以换成 Tavily、SerperAPI 或自定义工具。from langchain.tools import tool tool def search_web(query: str) - str: 当需要获取最新、实时或未知信息时使用此工具进行网络搜索。 # 这里是模拟返回。真实情况请接入搜索API。 print(f[搜索工具被调用] 查询词: {query}) mock_results { 今天北京的天气: 北京今天晴转多云气温15-25摄氏度南风2-3级。, LangGraph是什么: LangGraph 是 LangChain 项目的一部分用于构建有状态、多智能体工作流的库。, 特斯拉最新股价: 根据模拟数据特斯拉TSLA最新股价为 $175.32上涨 2.1%。 } return mock_results.get(query, f未找到关于 {query} 的明确信息。)3.2 构建工作流图节点与边这是最核心的部分。我们将工作流拆解成几个节点并连接它们。from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage, AIMessage # 初始化大模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 1. 路由节点判断是否需要搜索 def router_node(state: AgentState) - AgentState: 根据对话历史和问题判断是否需要调用搜索工具。 system_prompt 你是一个助手。你需要判断用户的问题是否需要联网搜索最新信息才能回答。 例如问“今天天气”、“最新新闻”、“实时股价”等需要搜索。 而问“你是谁”、“怎么编程”等通用知识或无需实时信息的问题则不需要搜索。 只回答“需要”或“不需要”。 # 构建判断用的消息 judge_messages [ SystemMessage(contentsystem_prompt), HumanMessage(contentf用户问题{state[question]}) ] response llm.invoke(judge_messages) decision response.content.strip() # 更新状态 state[should_search] decision 需要 # 也可以把判断过程记录到消息历史这里我们选择不记录保持主对话历史干净 return state # 2. 搜索节点执行搜索并记录结果 def search_node(state: AgentState) - AgentState: 执行搜索并将搜索结果以系统消息格式加入历史。 if not state[should_search]: return state search_query state[question] search_result search_web.invoke(search_query) # 将搜索结果作为一条“系统”或“工具”消息加入历史供后续节点使用 search_result_msg SystemMessage(contentf[网络搜索结果]{search_result}) state[messages].append(search_result_msg) return state # 3. 回答节点生成最终答案 def answer_node(state: AgentState) - AgentState: 综合对话历史和可能有的搜索结果生成最终答案。 # 最终回答时我们基于全部 messages 历史来生成 response llm.invoke(state[messages]) # 将AI的回答加入消息历史 state[messages].append(response) return state # 开始构建图 builder StateGraph(AgentState) # 添加节点 builder.add_node(router, router_node) builder.add_node(search, search_node) builder.add_node(answer, answer_node) # 设置入口点 builder.set_entry_point(router) # 添加边router节点之后根据 should_search 决定分支 def decide_next_node(state: AgentState): # 根据 router 节点设置的状态标志决定 if state[should_search]: return search # 需要搜索去 search 节点 else: return answer # 不需要搜索直接去 answer 节点 builder.add_conditional_edges( router, # 源节点 decide_next_node, # 条件判断函数 # 条件函数返回的字符串必须对应已添加的节点名或 END { search: search, answer: answer } ) # 添加普通边search 节点之后必然去 answer 节点 builder.add_edge(search, answer) # answer 节点之后工作流结束 builder.add_edge(answer, END) # 编译图得到可执行对象 graph builder.compile()3.3 运行与调试现在我们可以运行这个智能体了。# 初始化一个状态 initial_state: AgentState { messages: [SystemMessage(content你是一个有用的助手。)], # 初始系统提示 question: 今天北京的天气怎么样, should_search: False # 初始值不重要会被 router 节点覆盖 } # 执行图 final_state graph.invoke(initial_state) # 查看最终的消息历史 for msg in final_state[messages]: print(f{msg.type}: {msg.content})运行后你会看到类似这样的输出system: 你是一个有用的助手。 [搜索工具被调用] 查询词: 今天北京的天气怎么样 system: [网络搜索结果]北京今天晴转多云气温15-25摄氏度南风2-3级。 ai: 根据最新的网络搜索信息北京今天指您提问的当天的天气是晴转多云气温在15到25摄氏度之间风力为南风2-3级。天气不错适合外出活动。关键调试点检查状态流转在router_node里打印decision变量确认条件判断是否符合预期。检查工具调用确保search_web工具被正确调用并且返回的格式是字符串。检查消息历史最终state[‘messages’]列表里消息的顺序和内容是否正确。错误的顺序可能导致 LLM 理解混乱。4. 进阶实现长期记忆与复杂循环上面的例子是一个简单的二分支流程。但智能体常常需要更复杂的交互比如多轮工具调用直到满足条件为止。这就需要用上 LangGraph 的循环能力。我们改造一下智能体让它能处理“持续搜索直到信息足够”的场景。4.1 改造状态与工具我们引入一个search_count计数器和一个is_satisfied满意度标志。class AdvancedAgentState(TypedDict): messages: Annotated[List[BaseMessage], operator.add] question: str # 存储最近一次工具调用的结果 last_tool_result: str # 记录搜索次数 search_count: int # AI 判断信息是否已足够 is_satisfied: bool # 一个更“笨”的搜索工具每次只返回一点信息 tool def incremental_search(query: str) - str: 模拟一个每次只返回部分信息的搜索。 print(f[增量搜索] 查询: {query}) # 模拟第一次和第二次搜索返回不同信息 if not hasattr(incremental_search, “call_count“): incremental_search.call_count 0 incremental_search.call_count 1 if incremental_search.call_count 1: return “信息片段A北京今日白天晴。“ elif incremental_search.call_count 2: return “信息片段B最高气温25度。“ else: return “信息片段C无明显补充信息。“4.2 构建带循环的工作流这个工作流的核心思想是LLM - 判断是否满意 - 不满意 - 再搜索 - 再判断形成一个循环。from langgraph.graph import StateGraph, END from langchain_core.messages import ToolMessage def llm_node(state: AdvancedAgentState) - AdvancedAgentState: LLM节点分析当前信息并决定下一步行动。 # 构建给LLM的提示要求它判断信息是否足够并可能提出新的搜索指令 prompt f你正在回答这个问题{state[‘question‘]} 目前你掌握的信息如下来自之前的搜索 {state.get(‘last_tool_result‘, ‘暂无信息‘)} 请判断当前信息是否足够你给出一个完整、准确的回答 如果足够请直接生成最终答案。 如果不够请明确指出你还缺少哪部分信息。你的输出将驱动下一次搜索。 human_msg HumanMessage(contentprompt) state[“messages”].append(human_msg) response llm.invoke(state[“messages”]) state[“messages”].append(response) # 记录LLM的思考 # 简单判断如果回答中包含“足够”或答案看起来完整则标记满意 # 更复杂的实现可以让LLM输出一个结构化字段如 need_more: bool) answer_text response.content if “足够” in answer_text or len(answer_text) 50: # 简单启发式规则 state[“is_satisfied”] True else: state[“is_satisfied”] False # 可以把LLM指出的缺失信息存下来作为下一次搜索的query state[“question”] f“{state[‘question‘]} - 补充{answer_text}” return state def tool_node(state: AdvancedAgentState) - AdvancedAgentState: 工具节点执行搜索。 result incremental_search.invoke(state[“question”]) state[“last_tool_result”] result state[“search_count”] 1 # 将工具执行结果以 ToolMessage 格式加入历史这是标准做法 tool_msg ToolMessage(contentresult, tool_call_id“mock_id”) state[“messages”].append(tool_msg) return state def final_answer_node(state: AdvancedAgentState) - AdvancedAgentState: 最终回答节点循环结束后调用。 # 基于完整的对话历史生成最终答案 final_prompt “请基于所有对话历史给出最终答案。” state[“messages”].append(HumanMessage(contentfinal_prompt)) final_response llm.invoke(state[“messages”]) state[“messages”].append(final_response) return state # 构建图 builder StateGraph(AdvancedAgentState) builder.add_node(“llm”, llm_node) builder.add_node(“tool”, tool_node) builder.add_node(“final”, final_answer_node) builder.set_entry_point(“llm”) # 关键定义条件边实现 LLM - (工具 or 最终答案) 的循环 def decide_after_llm(state: AdvancedAgentState): if state[“is_satisfied”]: return “final” # 满意去最终回答 else: if state[“search_count”] 3: # 防止无限循环设置上限 return “tool” # 不满意且未超限去搜索 else: return “final” # 搜索次数太多强制结束 builder.add_conditional_edges( “llm”, decide_after_llm, {“tool”: “tool”, “final”: “final”} ) # 工具执行完后无条件返回 LLM 节点进行下一轮判断 builder.add_edge(“tool”, “llm”) # 最终回答后结束 builder.add_edge(“final”, END) advanced_graph builder.compile()运行这个图你会看到智能体在llm和tool节点间循环直到is_satisfied为 True 或搜索达到3次上限然后跳转到final节点给出答案。这就是 LangGraph 处理多步、循环任务的核心模式。5. 生产环境部署与关键踩坑点当你把 Demo 跑通准备投入实际项目时下面这些点需要特别注意。5.1 状态设计的陷阱状态字段的更新语义这是最容易出错的地方。回顾我们用的Annotated[List[BaseMessage], operator.add]。如果你定义了一个字段是str类型那么后一个节点的返回值会覆盖前一个节点的值。你需要想清楚每个字段是“累积”还是“替换”。状态不可变在节点函数内部不要直接修改传入的state字典虽然Python里可能行得通。最佳实践是创建一份更新后的字典返回。LangGraph 内部会处理合并。状态序列化如果你需要持久化工作流状态比如暂停后恢复确保State中所有字段都是可序列化的如基本类型、列表、字典。避免放入数据库连接、文件句柄等不可序列化对象。5.2 错误处理与持久化节点错误单个节点执行出错如工具调用超时、LLM API 异常会导致整个图执行失败。在生产环境中你需要考虑在节点函数内部使用try...except进行局部错误处理并返回一个代表错误的状态。利用 LangGraph 的interrupt机制在特定节点后暂停图执行便于人工干预或异步回调。使用checkpointer持久化状态。这是 LangGraph 的高级特性可以将图的状态保存到数据库如Redis、PostgreSQL实现故障恢复和长时运行任务。from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(“:memory:”) # 示例用内存生产用实际DB graph builder.compile(checkpointermemory) # 调用时传入 configurable 参数来支持断点续跑 config {“configurable”: {“thread_id”: “user_123_session_1”}} graph.invoke(initial_state, configconfig)5.3 性能与成本考量控制循环次数像上面的例子一样必须设置循环上限search_count 3防止因LLM判断失误或工具失效导致无限循环消耗大量 Token 和 API 费用。优化提示词节点中给 LLM 的提示词要精准。在router_node或decide_after_llm中的判断提示词直接决定了循环逻辑的效率和准确性。让 LLM 输出结构化 JSON如{“reasoning”: “…”, “next_step”: “search”}比解析自然语言更可靠。并发与流式对于高并发场景LangGraph 本身是框架并发能力取决于你的部署方式如 FastAPI 后端。考虑使用异步节点函数。对于需要实时响应的场景可以探索 LangGraph 的流式输出支持逐步返回结果。5.4 调试与监控可视化LangGraph 提供了graph.get_graph().draw_mermaid_png()方法需要安装pygraphviz可以将你的工作流图生成图片直观查看节点和边的关系对于复杂流程排查问题非常有用。日志记录在每个节点的关键步骤如调用 LLM 前、工具执行后添加详细的日志记录输入输出和状态变化。这对于追踪生产环境中的诡异问题至关重要。跟踪与评估集成 LangSmithLangChain 官方的跟踪平台或类似的 LLM 应用监控工具。它可以记录每次图执行的所有步骤、耗时、Token 使用和中间状态是进行性能分析和效果评估的利器。最后也是最实在的建议不要一开始就设计极其复杂的图。从一个能跑通的最小可行图开始逐步添加节点和分支。每加一个功能就完整测试一遍状态流转。LangGraph 的强大在于其清晰的抽象但复杂度的管理责任在开发者自己身上。先把单条路径跑稳再考虑并发、持久化和错误恢复这些生产级特性。