尧图网站建设 
在实际 AI 应用开发中构建一个能自主决策、调用工具并管理复杂流程的智能体Agent是许多开发者的目标。LangChain 提供了基础的 Agent 构建模块但当任务流程变得复杂、需要状态管理或涉及多步骤循环时仅靠 LangChain 的 Agent 接口会显得力不从心。这时LangGraph 作为一个基于有向图Graph的工作流编排框架其价值就凸显出来了。它允许你将 Agent 的逻辑清晰地定义为节点和边从而构建出具备长期记忆、条件分支和循环的动态工作流。本文面向已经了解 LangChain 基础概念希望将智能体能力从简单的“一问一答”升级到复杂、可编排流程的开发者。我们将通过一个完整的实战案例带你理解 LangGraph 的核心概念并一步步构建一个具备动态决策能力的智能体工作流。你将学会如何定义状态、创建节点、编排流程并最终运行一个可以处理多轮交互、根据结果调整策略的智能体系统。1. 理解 LangGraph从链到图的智能体进化在深入代码之前必须厘清 LangGraph 与 LangChain 的关系及其解决的问题。LangChain 是一个用于构建由语言模型驱动的应用程序的框架其Agent和AgentExecutor是早期实现智能体逻辑的核心。它们通常遵循“思考-行动-观察”的循环但这个循环是隐式且相对固定的状态管理较为简单。而 LangGraph 是 LangChain 生态系统中的一个库它引入了“图”这一抽象。在图论中图由节点Node和边Edge组成。在 LangGraph 的语境下节点Node通常是一个可调用的函数或工具负责执行一项具体任务例如调用语言模型、执行搜索、更新数据库。边Edge定义了节点之间的流转条件。它决定了在当前节点执行完毕后下一步应该前往哪个节点。边可以是固定的也可以是基于当前状态的动态条件。状态State一个贯穿整个工作流的共享数据结构。每个节点读取并修改这个状态状态的变化驱动着工作流的走向。这种设计带来了几个关键优势显式流程控制工作流的每一步都清晰可见不再是黑盒循环便于调试和理解。复杂逻辑编排轻松实现条件分支if-else、循环while、并行等复杂逻辑。持久化状态长期记忆图的状态可以被持久化到数据库从而实现跨会话的记忆这是构建具有“长期记忆”智能体的基础。人类介入点可以在特定节点设置“中断”等待人工审核或输入这对于关键业务流程至关重要。简单来说如果你需要构建的智能体只是回答一个问题并调用一次工具LangChain Agent 可能就够了。但如果你需要构建一个能处理多轮对话、根据中间结果选择不同路径、甚至需要记住之前交互内容的智能体LangGraph 是更强大的工具。它不是一个替代品而是一个在复杂场景下的增强和演进。2. 环境准备与核心依赖配置开始构建前我们需要一个干净的 Python 环境。建议使用 Python 3.9 或更高版本。2.1 创建虚拟环境与安装依赖首先创建一个新的项目目录并初始化虚拟环境。mkdir langgraph-agent-demo cd langgraph-agent-demo python -m venv venv激活虚拟环境Windows:venv\Scripts\activatemacOS/Linux:source venv/bin/activate接下来安装核心依赖。我们将使用 LangChain 和 LangGraph 的最新稳定版并选择 OpenAI 作为语言模型后端。同时为了示例需要我们安装一个用于网页内容提取的工具langchain-community。pip install langgraph langchain langchain-openai langchain-community注意langchain是一个元包通常安装它和特定版本的集成包如langchain-openai是标准做法。langgraph是独立包专门用于图工作流。2.2 配置 API 密钥本示例使用 OpenAI 的 GPT 模型。你需要在环境变量中设置你的 API 密钥。# 在命令行中临时设置仅当前会话有效 export OPENAI_API_KEYyour-api-key-here或者在 Python 代码中直接设置不推荐用于生产环境import os os.environ[OPENAI_API_KEY] your-api-key-here2.3 验证安装与基础导入创建一个名为demo_setup.py的脚本验证环境是否就绪。# demo_setup.py from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END print(LangChain OpenAI 客户端导入成功。) print(LangGraph StateGraph 导入成功。) # 简单测试 LLM 连接可选会消耗 token try: llm ChatOpenAI(modelgpt-3.5-turbo) # 不实际调用仅检查初始化 print(fLLM 模型初始化成功: {llm.model_name}) except Exception as e: print(f初始化失败: {e})运行该脚本确保没有报错python demo_setup.py3. 构建第一个 LangGraph 智能体天气查询工作流我们将构建一个简单的智能体它接收用户关于天气的查询决定是否需要查询具体城市然后模拟调用一个天气工具并返回结果。这个例子包含了状态定义、节点、边和条件路由。3.1 定义工作流状态状态是 LangGraph 的核心它是一个 TypedDict定义了工作流中所有需要传递和修改的数据。我们使用typing中的TypedDict和Annotated以及langgraph.graph的add_messages来实现消息的累加。# weather_agent.py from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage import operator # 1. 定义状态结构 class AgentState(TypedDict): # 用户输入的问题 input: str # 从输入中解析出的城市可能为空 city: str # 累积的对话消息历史用于给LLM提供上下文 messages: Annotated[list, operator.add] # 工作流的最终输出 output: str这里的关键是messages字段。Annotated[list, operator.add]是一个 LangGraph 的注解它告诉框架在节点间传递状态时对于messages列表采用“追加”add的方式合并而不是覆盖。这完美契合了对话历史累积的需求。3.2 创建节点函数节点是普通的 Python 函数它接收当前状态执行操作并返回一个包含更新后状态字段的字典。我们先创建第一个节点解析节点。它的职责是分析用户输入判断是否包含城市信息。# weather_agent.py (续) llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) def parse_input(state: AgentState) - dict: 解析用户输入提取城市信息。 user_input state[input] messages [HumanMessage(contentf请从以下用户问题中提取城市名。如果问题不涉及具体城市或无法提取请回答‘无’。问题{user_input})] response llm.invoke(messages) city response.content.strip() # 更新状态 return {city: city, messages: state[messages] [HumanMessage(contentuser_input), response]}接着创建第二个节点查询天气节点。它模拟调用一个天气 API。# weather_agent.py (续) def query_weather(state: AgentState) - dict: 模拟查询天气。 city state[city] # 这里模拟一个固定的API响应真实场景会调用如OpenWeatherMap的API weather_info f{city}的天气晴温度 22°C湿度 65%。 ai_msg AIMessage(contentweather_info) return {output: weather_info, messages: state[messages] [ai_msg]}然后创建第三个节点请求澄清节点。当城市信息缺失时请求用户澄清。# weather_agent.py (续) def ask_for_city(state: AgentState) - dict: 向用户请求城市信息。 clarification 您想查询哪个城市的天气呢 ai_msg AIMessage(contentclarification) # 注意在真实交互中这里可能需要暂停工作流等待用户输入。 # 本例中我们简化处理直接输出提示。 return {output: clarification, messages: state[messages] [ai_msg]}3.3 定义条件路由边我们需要一个路由函数根据parse_input节点的结果决定下一步是去查询天气还是请求澄清。# weather_agent.py (续) def route_after_parse(state: AgentState) - Literal[query_weather, ask_for_city]: 根据解析出的城市信息决定路由。 city state.get(city, ) # 如果城市信息有效且不是“无”则去查询天气 if city and city ! 无 and len(city) 20: # 简单长度过滤防止LLM胡言乱语 return query_weather else: return ask_for_city3.4 组装工作流图现在我们将节点和边组合成一个完整的工作流。# weather_agent.py (续) # 1. 创建图构建器 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(parse_input, parse_input) workflow.add_node(query_weather, query_weather) workflow.add_node(ask_for_city, ask_for_city) # 3. 设置入口点 workflow.set_entry_point(parse_input) # 4. 添加条件边 workflow.add_conditional_edges( parse_input, route_after_parse, # 路由判断函数 { query_weather: query_weather, ask_for_city: ask_for_city } ) # 5. 添加普通边从查询天气或请求澄清到结束 workflow.add_edge(query_weather, END) workflow.add_edge(ask_for_city, END) # 6. 编译图 app workflow.compile()3.5 运行与验证工作流编译后的app就是一个可执行的工作流。我们创建两个测试用例来验证其行为。# weather_agent.py (续) if __name__ __main__: # 测试用例1包含城市信息 print( 测试1查询‘北京天气怎么样’ ) initial_state1 {input: 北京天气怎么样, city: , messages: [], output: } result1 app.invoke(initial_state1) print(f最终输出: {result1[output]}) print(f最终城市: {result1[city]}) print(---\n) # 测试用例2不包含城市信息 print( 测试2查询‘今天会下雨吗’ ) initial_state2 {input: 今天会下雨吗, city: , messages: [], output: } result2 app.invoke(initial_state2) print(f最终输出: {result2[output]}) print(f最终城市: {result2[city]}) # 可选查看完整的状态演变 # import pprint # pprint.pprint(result2)保存并运行这个文件python weather_agent.py你应该看到类似以下的输出 测试1查询‘北京天气怎么样’ 最终输出: 北京的天气晴温度 22°C湿度 65%。 最终城市: 北京 --- 测试2查询‘今天会下雨吗’ 最终输出: 您想查询哪个城市的天气呢 最终城市: 无这个简单的例子展示了 LangGraph 的核心流程定义状态、创建节点、通过条件边控制流程。parse_input节点后的条件路由实现了智能体的首次决策。4. 进阶实战构建具备工具调用与循环的 Research Agent现在我们来构建一个更复杂的智能体它模拟一个研究助手能够根据用户主题自动决定是否需要联网搜索并整合信息生成报告。这个例子将引入工具调用、循环直到满足条件才结束以及更复杂的状态管理。4.1 定义进阶状态与工具我们首先定义一个更丰富的状态并创建一个模拟的网页搜索工具。# research_agent.py from typing import TypedDict, Annotated, Literal, List from langgraph.graph import StateGraph, END, START from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_core.tools import tool import operator import json # 1. 定义进阶状态 class ResearchState(TypedDict): # 用户原始查询 query: str # 累积的消息历史 messages: Annotated[List, operator.add] # 已收集的研究资料 research_materials: List[str] # 当前轮次防止无限循环 iteration: int # 最终报告 report: str # 标记是否需要继续搜索 should_continue: bool # 2. 创建一个模拟的搜索工具 tool def web_search_tool(query: str) - str: 模拟网页搜索工具。根据查询返回模拟的搜索结果。 在实际项目中这里应替换为真实的 SerperAPI、Tavily 或 Google Search API 调用。 # 这是一个模拟响应库 mock_responses { LangGraph 是什么: LangGraph 是 LangChain 的一个库用于构建基于有向图的工作流特别适合多步骤、有状态的智能体应用。它提供了状态管理和条件路由。, LangGraph 长期记忆: LangGraph 通过将图的状态持久化到数据库如 PostgreSQL、Redis来实现长期记忆使得智能体可以跨会话记住上下文。, 智能体开发框架对比: 主流智能体框架包括 LangChain Agent、LangGraph、AutoGen、CrewAI。LangGraph 在复杂工作流编排和状态管理上优势明显。, 天气: 这是一个通用词汇没有具体信息。 } # 返回匹配的模拟结果若无匹配则返回默认信息 for key, value in mock_responses.items(): if key.lower() in query.lower(): return f模拟搜索到信息{value} return f模拟搜索到信息关于‘{query}’未找到高度相关的特定资料。4.2 创建决策与执行节点这个工作流将包含三个核心节点规划节点分析当前查询和已有材料决定下一步是搜索还是生成报告。搜索节点调用搜索工具收集信息。报告节点整合所有材料生成最终报告。# research_agent.py (续) llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) tools [web_search_tool] llm_with_tools llm.bind_tools(tools) def planner_node(state: ResearchState) - dict: 规划节点决定下一步行动。 messages state[messages] # 构建给LLM的提示包含历史和研究材料 prompt f 你是一个研究助手。当前用户查询是{state[query]} 目前已收集的研究材料有{state[research_materials]} 当前是第 {state[iteration]} 轮分析。 请决定下一步行动 1. 如果已有材料足以回答用户查询或者迭代次数超过3次则选择 generate_report。 2. 如果还需要更多信息则选择 search_web。 只返回 generate_report 或 search_web 中的一个词。 decision_msg HumanMessage(contentprompt) response llm.invoke([decision_msg]) decision response.content.strip().lower() should_continue decision search_web and state[iteration] 3 update { messages: state[messages] [decision_msg, response], should_continue: should_continue, iteration: state[iteration] 1 } # 将决策也存入状态供后续边使用 update[_next_step] decision return update def search_node(state: ResearchState) - dict: 搜索节点调用工具获取信息。 # 基于查询和历史让LLM决定搜索关键词 prompt f根据以下查询和历史生成一个简洁的网页搜索关键词。查询{state[query]} search_query_msg HumanMessage(contentprompt) response llm.invoke([search_query_msg]) search_keyword response.content.strip() # 调用工具 tool_result web_search_tool.invoke(search_keyword) # 更新材料列表 new_materials state[research_materials] [tool_result] # 构造工具调用和结果消息以便记录到历史 tool_call_id fcall_{state[iteration]} ai_msg_with_tool AIMessage(content, tool_calls[{ name: web_search_tool, args: {query: search_keyword}, id: tool_call_id }]) tool_msg ToolMessage(contenttool_result, tool_call_idtool_call_id) return { research_materials: new_materials, messages: state[messages] [search_query_msg, response, ai_msg_with_tool, tool_msg] } def report_node(state: ResearchState) - dict: 报告节点生成最终答案。 prompt f 基于以下用户查询和收集到的研究材料生成一份简洁、准确的研究报告。 用户查询{state[query]} 研究材料 {chr(10).join(state[research_materials])} 报告 report_msg HumanMessage(contentprompt) response llm.invoke([report_msg]) return { report: response.content, messages: state[messages] [report_msg, response], should_continue: False # 生成报告后停止循环 }4.3 组装带循环的工作流这个工作流的关键在于引入循环规划 - (搜索 - 规划) 循环直到规划节点决定生成报告。# research_agent.py (续) # 1. 创建图 workflow StateGraph(ResearchState) # 2. 添加节点 workflow.add_node(planner, planner_node) workflow.add_node(search, search_node) workflow.add_node(generate_report, report_node) # 3. 设置入口点 workflow.set_entry_point(planner) # 4. 定义从规划节点出发的条件边 def decide_after_planning(state: ResearchState) - Literal[search, generate_report, __end__]: 根据规划节点的决策和循环条件路由。 # 从 planner_node 存入的状态中获取决策 next_step state.get(_next_step, ) should_continue state.get(should_continue, False) if not should_continue or next_step generate_report: return generate_report elif next_step search_web and should_continue: return search else: # 安全回退结束流程 return __end__ workflow.add_conditional_edges( planner, decide_after_planning, { search: search, generate_report: generate_report, __end__: END } ) # 5. 添加从搜索节点回到规划节点的边形成循环 workflow.add_edge(search, planner) # 6. 添加从报告节点到结束的边 workflow.add_edge(generate_report, END) # 7. 编译 app workflow.compile()4.4 运行与观察循环行为现在我们可以运行这个研究助手观察它如何循环决策。# research_agent.py (续) if __name__ __main__: print( 启动研究助手 ) initial_state { query: 请帮我研究一下 LangGraph 和它的长期记忆功能, messages: [], research_materials: [], iteration: 0, report: , should_continue: True } # 为了观察流程我们可以使用流的模式或者多次调用。 # 这里我们直接调用并打印最终状态。 final_state app.invoke(initial_state) print(f\n用户查询: {final_state[query]}) print(f\n最终迭代次数: {final_state[iteration]}) print(f\n收集到的材料:) for i, mat in enumerate(final_state[research_materials]): print(f {i1}. {mat[:100]}...) # 截断显示 print(f\n 生成的研究报告 ) print(final_state[report]) print(*50)运行此脚本python research_agent.py输出将展示智能体如何根据初始查询可能先搜索“LangGraph 是什么”将结果加入材料库然后规划节点判断是否需要更多信息例如关于“长期记忆”从而发起第二次搜索最后在材料足够或达到迭代上限时生成一份整合报告。这个过程完美演示了 LangGraph 如何管理带有循环和状态依赖的多步骤工作流。5. 关键配置、调试与生产环境考量构建出可运行的工作流只是第一步。要让其在开发和生产环境中稳定可靠还需要关注以下方面。5.1 状态持久化与长期记忆LangGraph 的State可以被持久化这是实现“长期记忆”的关键。你需要一个Checkpointer来保存和加载图的状态。from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import StateGraph, START, END import sqlite3 # 创建或连接到 SQLite 数据库 conn sqlite3.connect(checkpoints.db) checkpointer SqliteSaver(conn) # 在编译图时传入 checkpointer app workflow.compile(checkpointercheckpointer) # 调用时需要指定一个线程IDthread_id来标识会话 config {configurable: {thread_id: user_session_123}} initial_state {...} # 第一次调用状态会被保存 result1 app.invoke(initial_state, configconfig) # 一段时间后可以基于同一个 thread_id 继续执行状态会被恢复 new_state {input: 基于我们刚才的对话...} result2 app.invoke(new_state, configconfig) # 此时 result2 包含了之前的 messages 历史生产环境中你可能需要使用 PostgreSQL、Redis 等更强大的后端LangGraph 提供了相应的Saver实现。5.2 错误处理与超时控制节点函数可能因网络、API限制或逻辑错误而失败。在生产工作流中必须加入错误处理。from tenacity import retry, stop_after_attempt, wait_exponential import asyncio retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def reliable_llm_call(messages): 一个带重试机制的LLM调用函数。 try: response llm.invoke(messages) return response except Exception as e: print(fLLM调用失败: {e}) raise # 触发重试 def robust_node(state: State): try: # ... 业务逻辑使用 reliable_llm_call result reliable_llm_call(state[messages]) return {output: result.content} except Exception as e: # 记录错误并返回一个错误状态让路由函数可以处理 return {error: str(e), should_stop: True}在图定义中你可以添加一个专门的error_handler节点或者让路由函数检查状态中的error字段并路由到降级处理或人工审核节点。5.3 可视化与调试LangGraph 支持将工作流图可视化这对于理解和沟通复杂流程至关重要。# 将图导出为 PNG 图片 from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果环境不支持可以输出 Mermaid 文本在支持 Mermaid 的 Markdown 编辑器中查看 print(app.get_graph().draw_mermaid())在调试时可以启用更详细的日志或者使用app.invoke(..., debugTrue)来查看每一步的状态变化。5.4 性能与成本优化缓存对 LLM 调用和工具调用如搜索结果实施缓存避免重复计算。可以使用langchain.cache。流式输出对于报告生成等长文本节点使用 LLM 的流式响应 (stream) 来提升用户体验。限制循环次数如我们例子中的iteration字段必须设置硬性上限防止因逻辑错误导致无限循环和 API 费用激增。异步执行如果节点间没有强依赖可以考虑使用add_node的异步版本和asyncio来并行执行提升整体吞吐量。6. 常见问题排查清单在开发 LangGraph 智能体时你可能会遇到以下典型问题。问题现象可能原因检查方式处理建议app.invoke报错KeyError状态State的 TypedDict 定义与节点返回的字典键不匹配。1. 检查TypedDict的所有字段名。2. 检查每个节点函数return的字典键是否都是TypedDict中定义的键。确保节点返回的字典是状态字段的子集或全集。使用set(state.keys())打印状态键进行比对。条件边add_conditional_edges不生效总是走默认或报错1. 路由函数返回值不在映射字典的键中。2. 路由函数依赖的状态字段在调用时未正确更新。1. 在路由函数内打印其返回值。2. 检查提供该字段的节点是否已正确执行并更新状态。确保路由函数返回的字面量字符串与add_conditional_edges中映射字典的键完全一致。使用print调试状态流转。消息历史messages没有被累加状态中messages字段未使用Annotated[list, operator.add]注解。检查状态类定义。必须使用Annotated[list, operator.add]或Annotated[list, add_messages]来声明messages字段框架才知道如何合并列表。工作流陷入无限循环循环条件如should_continue始终为True或条件边逻辑有误。1. 在规划节点打印决策逻辑和循环条件。2. 检查状态中用于控制循环的字段如iteration是否被正确更新。务必设置循环上限如iteration。在路由函数中严格判断终止条件。使用app.invoke的debug模式跟踪状态。工具调用结果未存入历史工具调用后未正确构造AIMessage和ToolMessage并追加到messages。检查调用工具的节点是否返回了包含更新后messages列表的状态。遵循 LangChain 消息协议先返回带有tool_calls的AIMessage再返回对应的ToolMessage。两者都需加入messages列表。持久化状态恢复后上下文丢失1.thread_id不一致。2. 使用的Checkpointer配置错误或数据库连接问题。1. 确认每次调用时config中的thread_id相同。2. 检查数据库表是否创建数据是否写入。确保会话的thread_id稳定。对于生产环境使用更可靠的Checkpointer后端并监控其连接状态。LLM 调用超时或响应慢网络问题、API 限流或提示词过于复杂。1. 增加超时设置。2. 简化提示词。3. 查看 LLM 供应商的控制台监控。在 LLM 客户端设置超时参数。对提示词进行优化和裁剪。实现重试和退避机制如使用tenacity库。7. 生产环境最佳实践将 LangGraph 智能体部署到生产环境除了上述错误处理和性能优化还需遵循以下实践配置外部化不要将 API 密钥、模型参数、数据库连接字符串等硬编码在代码中。使用环境变量或配置中心如python-dotenv读取.env文件。日志与监控在每个节点函数的入口和出口添加结构化日志记录输入、输出、耗时和错误。集成像 Prometheus 和 Grafana 这样的监控系统跟踪工作流执行次数、成功率、延迟等指标。版本化管理图定义工作流图StateGraph的定义是核心业务逻辑。应将其作为代码的一部分进行版本控制Git。考虑将复杂的图拆分成模块化组件便于管理和测试。实现降级策略对于关键节点如 LLM 调用、外部 API 调用设计降级方案。例如当搜索工具失败时可以转而从本地知识库中检索或直接返回一个友好的提示信息而不是让整个工作流崩溃。人工审核环节对于涉及内容安全、重要决策或高成本操作的节点可以设计边路由到一个“人工审核”节点暂停工作流并通知审核人员待人工确认后再继续执行。这可以通过 LangGraph 的interrupt机制实现。全面的测试单元测试单独测试每个节点函数。集成测试测试整个工作流在不同输入下的行为。负载测试模拟高并发场景检查状态持久化后端和 LLM API 的承受能力。清晰的文档使用app.get_graph().draw_mermaid()生成的图表作为技术文档的一部分帮助团队成员理解智能体的决策流程。为每个节点和状态字段编写清晰的注释。LangGraph 将智能体开发从线性的链式思维提升到了图式的工作流思维为构建复杂、可靠、可维护的 AI 应用提供了强大的底层支持。从简单的条件路由到带有记忆和循环的复杂智能体其核心始终在于对状态和流程的精确控制。开始你的项目时建议从一个小而具体的工作流入手逐步迭代增加节点和复杂性并始终将可观测性和错误处理放在首位。 LangGraph实战：构建具备长期记忆与动态决策的智能体工作流
在实际 AI 应用开发中构建一个能自主决策、调用工具并管理复杂流程的智能体Agent是许多开发者的目标。LangChain 提供了基础的 Agent 构建模块但当任务流程变得复杂、需要状态管理或涉及多步骤循环时仅靠 LangChain 的 Agent 接口会显得力不从心。这时LangGraph 作为一个基于有向图Graph的工作流编排框架其价值就凸显出来了。它允许你将 Agent 的逻辑清晰地定义为节点和边从而构建出具备长期记忆、条件分支和循环的动态工作流。本文面向已经了解 LangChain 基础概念希望将智能体能力从简单的“一问一答”升级到复杂、可编排流程的开发者。我们将通过一个完整的实战案例带你理解 LangGraph 的核心概念并一步步构建一个具备动态决策能力的智能体工作流。你将学会如何定义状态、创建节点、编排流程并最终运行一个可以处理多轮交互、根据结果调整策略的智能体系统。1. 理解 LangGraph从链到图的智能体进化在深入代码之前必须厘清 LangGraph 与 LangChain 的关系及其解决的问题。LangChain 是一个用于构建由语言模型驱动的应用程序的框架其Agent和AgentExecutor是早期实现智能体逻辑的核心。它们通常遵循“思考-行动-观察”的循环但这个循环是隐式且相对固定的状态管理较为简单。而 LangGraph 是 LangChain 生态系统中的一个库它引入了“图”这一抽象。在图论中图由节点Node和边Edge组成。在 LangGraph 的语境下节点Node通常是一个可调用的函数或工具负责执行一项具体任务例如调用语言模型、执行搜索、更新数据库。边Edge定义了节点之间的流转条件。它决定了在当前节点执行完毕后下一步应该前往哪个节点。边可以是固定的也可以是基于当前状态的动态条件。状态State一个贯穿整个工作流的共享数据结构。每个节点读取并修改这个状态状态的变化驱动着工作流的走向。这种设计带来了几个关键优势显式流程控制工作流的每一步都清晰可见不再是黑盒循环便于调试和理解。复杂逻辑编排轻松实现条件分支if-else、循环while、并行等复杂逻辑。持久化状态长期记忆图的状态可以被持久化到数据库从而实现跨会话的记忆这是构建具有“长期记忆”智能体的基础。人类介入点可以在特定节点设置“中断”等待人工审核或输入这对于关键业务流程至关重要。简单来说如果你需要构建的智能体只是回答一个问题并调用一次工具LangChain Agent 可能就够了。但如果你需要构建一个能处理多轮对话、根据中间结果选择不同路径、甚至需要记住之前交互内容的智能体LangGraph 是更强大的工具。它不是一个替代品而是一个在复杂场景下的增强和演进。2. 环境准备与核心依赖配置开始构建前我们需要一个干净的 Python 环境。建议使用 Python 3.9 或更高版本。2.1 创建虚拟环境与安装依赖首先创建一个新的项目目录并初始化虚拟环境。mkdir langgraph-agent-demo cd langgraph-agent-demo python -m venv venv激活虚拟环境Windows:venv\Scripts\activatemacOS/Linux:source venv/bin/activate接下来安装核心依赖。我们将使用 LangChain 和 LangGraph 的最新稳定版并选择 OpenAI 作为语言模型后端。同时为了示例需要我们安装一个用于网页内容提取的工具langchain-community。pip install langgraph langchain langchain-openai langchain-community注意langchain是一个元包通常安装它和特定版本的集成包如langchain-openai是标准做法。langgraph是独立包专门用于图工作流。2.2 配置 API 密钥本示例使用 OpenAI 的 GPT 模型。你需要在环境变量中设置你的 API 密钥。# 在命令行中临时设置仅当前会话有效 export OPENAI_API_KEYyour-api-key-here或者在 Python 代码中直接设置不推荐用于生产环境import os os.environ[OPENAI_API_KEY] your-api-key-here2.3 验证安装与基础导入创建一个名为demo_setup.py的脚本验证环境是否就绪。# demo_setup.py from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END print(LangChain OpenAI 客户端导入成功。) print(LangGraph StateGraph 导入成功。) # 简单测试 LLM 连接可选会消耗 token try: llm ChatOpenAI(modelgpt-3.5-turbo) # 不实际调用仅检查初始化 print(fLLM 模型初始化成功: {llm.model_name}) except Exception as e: print(f初始化失败: {e})运行该脚本确保没有报错python demo_setup.py3. 构建第一个 LangGraph 智能体天气查询工作流我们将构建一个简单的智能体它接收用户关于天气的查询决定是否需要查询具体城市然后模拟调用一个天气工具并返回结果。这个例子包含了状态定义、节点、边和条件路由。3.1 定义工作流状态状态是 LangGraph 的核心它是一个 TypedDict定义了工作流中所有需要传递和修改的数据。我们使用typing中的TypedDict和Annotated以及langgraph.graph的add_messages来实现消息的累加。# weather_agent.py from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage import operator # 1. 定义状态结构 class AgentState(TypedDict): # 用户输入的问题 input: str # 从输入中解析出的城市可能为空 city: str # 累积的对话消息历史用于给LLM提供上下文 messages: Annotated[list, operator.add] # 工作流的最终输出 output: str这里的关键是messages字段。Annotated[list, operator.add]是一个 LangGraph 的注解它告诉框架在节点间传递状态时对于messages列表采用“追加”add的方式合并而不是覆盖。这完美契合了对话历史累积的需求。3.2 创建节点函数节点是普通的 Python 函数它接收当前状态执行操作并返回一个包含更新后状态字段的字典。我们先创建第一个节点解析节点。它的职责是分析用户输入判断是否包含城市信息。# weather_agent.py (续) llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) def parse_input(state: AgentState) - dict: 解析用户输入提取城市信息。 user_input state[input] messages [HumanMessage(contentf请从以下用户问题中提取城市名。如果问题不涉及具体城市或无法提取请回答‘无’。问题{user_input})] response llm.invoke(messages) city response.content.strip() # 更新状态 return {city: city, messages: state[messages] [HumanMessage(contentuser_input), response]}接着创建第二个节点查询天气节点。它模拟调用一个天气 API。# weather_agent.py (续) def query_weather(state: AgentState) - dict: 模拟查询天气。 city state[city] # 这里模拟一个固定的API响应真实场景会调用如OpenWeatherMap的API weather_info f{city}的天气晴温度 22°C湿度 65%。 ai_msg AIMessage(contentweather_info) return {output: weather_info, messages: state[messages] [ai_msg]}然后创建第三个节点请求澄清节点。当城市信息缺失时请求用户澄清。# weather_agent.py (续) def ask_for_city(state: AgentState) - dict: 向用户请求城市信息。 clarification 您想查询哪个城市的天气呢 ai_msg AIMessage(contentclarification) # 注意在真实交互中这里可能需要暂停工作流等待用户输入。 # 本例中我们简化处理直接输出提示。 return {output: clarification, messages: state[messages] [ai_msg]}3.3 定义条件路由边我们需要一个路由函数根据parse_input节点的结果决定下一步是去查询天气还是请求澄清。# weather_agent.py (续) def route_after_parse(state: AgentState) - Literal[query_weather, ask_for_city]: 根据解析出的城市信息决定路由。 city state.get(city, ) # 如果城市信息有效且不是“无”则去查询天气 if city and city ! 无 and len(city) 20: # 简单长度过滤防止LLM胡言乱语 return query_weather else: return ask_for_city3.4 组装工作流图现在我们将节点和边组合成一个完整的工作流。# weather_agent.py (续) # 1. 创建图构建器 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(parse_input, parse_input) workflow.add_node(query_weather, query_weather) workflow.add_node(ask_for_city, ask_for_city) # 3. 设置入口点 workflow.set_entry_point(parse_input) # 4. 添加条件边 workflow.add_conditional_edges( parse_input, route_after_parse, # 路由判断函数 { query_weather: query_weather, ask_for_city: ask_for_city } ) # 5. 添加普通边从查询天气或请求澄清到结束 workflow.add_edge(query_weather, END) workflow.add_edge(ask_for_city, END) # 6. 编译图 app workflow.compile()3.5 运行与验证工作流编译后的app就是一个可执行的工作流。我们创建两个测试用例来验证其行为。# weather_agent.py (续) if __name__ __main__: # 测试用例1包含城市信息 print( 测试1查询‘北京天气怎么样’ ) initial_state1 {input: 北京天气怎么样, city: , messages: [], output: } result1 app.invoke(initial_state1) print(f最终输出: {result1[output]}) print(f最终城市: {result1[city]}) print(---\n) # 测试用例2不包含城市信息 print( 测试2查询‘今天会下雨吗’ ) initial_state2 {input: 今天会下雨吗, city: , messages: [], output: } result2 app.invoke(initial_state2) print(f最终输出: {result2[output]}) print(f最终城市: {result2[city]}) # 可选查看完整的状态演变 # import pprint # pprint.pprint(result2)保存并运行这个文件python weather_agent.py你应该看到类似以下的输出 测试1查询‘北京天气怎么样’ 最终输出: 北京的天气晴温度 22°C湿度 65%。 最终城市: 北京 --- 测试2查询‘今天会下雨吗’ 最终输出: 您想查询哪个城市的天气呢 最终城市: 无这个简单的例子展示了 LangGraph 的核心流程定义状态、创建节点、通过条件边控制流程。parse_input节点后的条件路由实现了智能体的首次决策。4. 进阶实战构建具备工具调用与循环的 Research Agent现在我们来构建一个更复杂的智能体它模拟一个研究助手能够根据用户主题自动决定是否需要联网搜索并整合信息生成报告。这个例子将引入工具调用、循环直到满足条件才结束以及更复杂的状态管理。4.1 定义进阶状态与工具我们首先定义一个更丰富的状态并创建一个模拟的网页搜索工具。# research_agent.py from typing import TypedDict, Annotated, Literal, List from langgraph.graph import StateGraph, END, START from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_core.tools import tool import operator import json # 1. 定义进阶状态 class ResearchState(TypedDict): # 用户原始查询 query: str # 累积的消息历史 messages: Annotated[List, operator.add] # 已收集的研究资料 research_materials: List[str] # 当前轮次防止无限循环 iteration: int # 最终报告 report: str # 标记是否需要继续搜索 should_continue: bool # 2. 创建一个模拟的搜索工具 tool def web_search_tool(query: str) - str: 模拟网页搜索工具。根据查询返回模拟的搜索结果。 在实际项目中这里应替换为真实的 SerperAPI、Tavily 或 Google Search API 调用。 # 这是一个模拟响应库 mock_responses { LangGraph 是什么: LangGraph 是 LangChain 的一个库用于构建基于有向图的工作流特别适合多步骤、有状态的智能体应用。它提供了状态管理和条件路由。, LangGraph 长期记忆: LangGraph 通过将图的状态持久化到数据库如 PostgreSQL、Redis来实现长期记忆使得智能体可以跨会话记住上下文。, 智能体开发框架对比: 主流智能体框架包括 LangChain Agent、LangGraph、AutoGen、CrewAI。LangGraph 在复杂工作流编排和状态管理上优势明显。, 天气: 这是一个通用词汇没有具体信息。 } # 返回匹配的模拟结果若无匹配则返回默认信息 for key, value in mock_responses.items(): if key.lower() in query.lower(): return f模拟搜索到信息{value} return f模拟搜索到信息关于‘{query}’未找到高度相关的特定资料。4.2 创建决策与执行节点这个工作流将包含三个核心节点规划节点分析当前查询和已有材料决定下一步是搜索还是生成报告。搜索节点调用搜索工具收集信息。报告节点整合所有材料生成最终报告。# research_agent.py (续) llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) tools [web_search_tool] llm_with_tools llm.bind_tools(tools) def planner_node(state: ResearchState) - dict: 规划节点决定下一步行动。 messages state[messages] # 构建给LLM的提示包含历史和研究材料 prompt f 你是一个研究助手。当前用户查询是{state[query]} 目前已收集的研究材料有{state[research_materials]} 当前是第 {state[iteration]} 轮分析。 请决定下一步行动 1. 如果已有材料足以回答用户查询或者迭代次数超过3次则选择 generate_report。 2. 如果还需要更多信息则选择 search_web。 只返回 generate_report 或 search_web 中的一个词。 decision_msg HumanMessage(contentprompt) response llm.invoke([decision_msg]) decision response.content.strip().lower() should_continue decision search_web and state[iteration] 3 update { messages: state[messages] [decision_msg, response], should_continue: should_continue, iteration: state[iteration] 1 } # 将决策也存入状态供后续边使用 update[_next_step] decision return update def search_node(state: ResearchState) - dict: 搜索节点调用工具获取信息。 # 基于查询和历史让LLM决定搜索关键词 prompt f根据以下查询和历史生成一个简洁的网页搜索关键词。查询{state[query]} search_query_msg HumanMessage(contentprompt) response llm.invoke([search_query_msg]) search_keyword response.content.strip() # 调用工具 tool_result web_search_tool.invoke(search_keyword) # 更新材料列表 new_materials state[research_materials] [tool_result] # 构造工具调用和结果消息以便记录到历史 tool_call_id fcall_{state[iteration]} ai_msg_with_tool AIMessage(content, tool_calls[{ name: web_search_tool, args: {query: search_keyword}, id: tool_call_id }]) tool_msg ToolMessage(contenttool_result, tool_call_idtool_call_id) return { research_materials: new_materials, messages: state[messages] [search_query_msg, response, ai_msg_with_tool, tool_msg] } def report_node(state: ResearchState) - dict: 报告节点生成最终答案。 prompt f 基于以下用户查询和收集到的研究材料生成一份简洁、准确的研究报告。 用户查询{state[query]} 研究材料 {chr(10).join(state[research_materials])} 报告 report_msg HumanMessage(contentprompt) response llm.invoke([report_msg]) return { report: response.content, messages: state[messages] [report_msg, response], should_continue: False # 生成报告后停止循环 }4.3 组装带循环的工作流这个工作流的关键在于引入循环规划 - (搜索 - 规划) 循环直到规划节点决定生成报告。# research_agent.py (续) # 1. 创建图 workflow StateGraph(ResearchState) # 2. 添加节点 workflow.add_node(planner, planner_node) workflow.add_node(search, search_node) workflow.add_node(generate_report, report_node) # 3. 设置入口点 workflow.set_entry_point(planner) # 4. 定义从规划节点出发的条件边 def decide_after_planning(state: ResearchState) - Literal[search, generate_report, __end__]: 根据规划节点的决策和循环条件路由。 # 从 planner_node 存入的状态中获取决策 next_step state.get(_next_step, ) should_continue state.get(should_continue, False) if not should_continue or next_step generate_report: return generate_report elif next_step search_web and should_continue: return search else: # 安全回退结束流程 return __end__ workflow.add_conditional_edges( planner, decide_after_planning, { search: search, generate_report: generate_report, __end__: END } ) # 5. 添加从搜索节点回到规划节点的边形成循环 workflow.add_edge(search, planner) # 6. 添加从报告节点到结束的边 workflow.add_edge(generate_report, END) # 7. 编译 app workflow.compile()4.4 运行与观察循环行为现在我们可以运行这个研究助手观察它如何循环决策。# research_agent.py (续) if __name__ __main__: print( 启动研究助手 ) initial_state { query: 请帮我研究一下 LangGraph 和它的长期记忆功能, messages: [], research_materials: [], iteration: 0, report: , should_continue: True } # 为了观察流程我们可以使用流的模式或者多次调用。 # 这里我们直接调用并打印最终状态。 final_state app.invoke(initial_state) print(f\n用户查询: {final_state[query]}) print(f\n最终迭代次数: {final_state[iteration]}) print(f\n收集到的材料:) for i, mat in enumerate(final_state[research_materials]): print(f {i1}. {mat[:100]}...) # 截断显示 print(f\n 生成的研究报告 ) print(final_state[report]) print(*50)运行此脚本python research_agent.py输出将展示智能体如何根据初始查询可能先搜索“LangGraph 是什么”将结果加入材料库然后规划节点判断是否需要更多信息例如关于“长期记忆”从而发起第二次搜索最后在材料足够或达到迭代上限时生成一份整合报告。这个过程完美演示了 LangGraph 如何管理带有循环和状态依赖的多步骤工作流。5. 关键配置、调试与生产环境考量构建出可运行的工作流只是第一步。要让其在开发和生产环境中稳定可靠还需要关注以下方面。5.1 状态持久化与长期记忆LangGraph 的State可以被持久化这是实现“长期记忆”的关键。你需要一个Checkpointer来保存和加载图的状态。from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import StateGraph, START, END import sqlite3 # 创建或连接到 SQLite 数据库 conn sqlite3.connect(checkpoints.db) checkpointer SqliteSaver(conn) # 在编译图时传入 checkpointer app workflow.compile(checkpointercheckpointer) # 调用时需要指定一个线程IDthread_id来标识会话 config {configurable: {thread_id: user_session_123}} initial_state {...} # 第一次调用状态会被保存 result1 app.invoke(initial_state, configconfig) # 一段时间后可以基于同一个 thread_id 继续执行状态会被恢复 new_state {input: 基于我们刚才的对话...} result2 app.invoke(new_state, configconfig) # 此时 result2 包含了之前的 messages 历史生产环境中你可能需要使用 PostgreSQL、Redis 等更强大的后端LangGraph 提供了相应的Saver实现。5.2 错误处理与超时控制节点函数可能因网络、API限制或逻辑错误而失败。在生产工作流中必须加入错误处理。from tenacity import retry, stop_after_attempt, wait_exponential import asyncio retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def reliable_llm_call(messages): 一个带重试机制的LLM调用函数。 try: response llm.invoke(messages) return response except Exception as e: print(fLLM调用失败: {e}) raise # 触发重试 def robust_node(state: State): try: # ... 业务逻辑使用 reliable_llm_call result reliable_llm_call(state[messages]) return {output: result.content} except Exception as e: # 记录错误并返回一个错误状态让路由函数可以处理 return {error: str(e), should_stop: True}在图定义中你可以添加一个专门的error_handler节点或者让路由函数检查状态中的error字段并路由到降级处理或人工审核节点。5.3 可视化与调试LangGraph 支持将工作流图可视化这对于理解和沟通复杂流程至关重要。# 将图导出为 PNG 图片 from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果环境不支持可以输出 Mermaid 文本在支持 Mermaid 的 Markdown 编辑器中查看 print(app.get_graph().draw_mermaid())在调试时可以启用更详细的日志或者使用app.invoke(..., debugTrue)来查看每一步的状态变化。5.4 性能与成本优化缓存对 LLM 调用和工具调用如搜索结果实施缓存避免重复计算。可以使用langchain.cache。流式输出对于报告生成等长文本节点使用 LLM 的流式响应 (stream) 来提升用户体验。限制循环次数如我们例子中的iteration字段必须设置硬性上限防止因逻辑错误导致无限循环和 API 费用激增。异步执行如果节点间没有强依赖可以考虑使用add_node的异步版本和asyncio来并行执行提升整体吞吐量。6. 常见问题排查清单在开发 LangGraph 智能体时你可能会遇到以下典型问题。问题现象可能原因检查方式处理建议app.invoke报错KeyError状态State的 TypedDict 定义与节点返回的字典键不匹配。1. 检查TypedDict的所有字段名。2. 检查每个节点函数return的字典键是否都是TypedDict中定义的键。确保节点返回的字典是状态字段的子集或全集。使用set(state.keys())打印状态键进行比对。条件边add_conditional_edges不生效总是走默认或报错1. 路由函数返回值不在映射字典的键中。2. 路由函数依赖的状态字段在调用时未正确更新。1. 在路由函数内打印其返回值。2. 检查提供该字段的节点是否已正确执行并更新状态。确保路由函数返回的字面量字符串与add_conditional_edges中映射字典的键完全一致。使用print调试状态流转。消息历史messages没有被累加状态中messages字段未使用Annotated[list, operator.add]注解。检查状态类定义。必须使用Annotated[list, operator.add]或Annotated[list, add_messages]来声明messages字段框架才知道如何合并列表。工作流陷入无限循环循环条件如should_continue始终为True或条件边逻辑有误。1. 在规划节点打印决策逻辑和循环条件。2. 检查状态中用于控制循环的字段如iteration是否被正确更新。务必设置循环上限如iteration。在路由函数中严格判断终止条件。使用app.invoke的debug模式跟踪状态。工具调用结果未存入历史工具调用后未正确构造AIMessage和ToolMessage并追加到messages。检查调用工具的节点是否返回了包含更新后messages列表的状态。遵循 LangChain 消息协议先返回带有tool_calls的AIMessage再返回对应的ToolMessage。两者都需加入messages列表。持久化状态恢复后上下文丢失1.thread_id不一致。2. 使用的Checkpointer配置错误或数据库连接问题。1. 确认每次调用时config中的thread_id相同。2. 检查数据库表是否创建数据是否写入。确保会话的thread_id稳定。对于生产环境使用更可靠的Checkpointer后端并监控其连接状态。LLM 调用超时或响应慢网络问题、API 限流或提示词过于复杂。1. 增加超时设置。2. 简化提示词。3. 查看 LLM 供应商的控制台监控。在 LLM 客户端设置超时参数。对提示词进行优化和裁剪。实现重试和退避机制如使用tenacity库。7. 生产环境最佳实践将 LangGraph 智能体部署到生产环境除了上述错误处理和性能优化还需遵循以下实践配置外部化不要将 API 密钥、模型参数、数据库连接字符串等硬编码在代码中。使用环境变量或配置中心如python-dotenv读取.env文件。日志与监控在每个节点函数的入口和出口添加结构化日志记录输入、输出、耗时和错误。集成像 Prometheus 和 Grafana 这样的监控系统跟踪工作流执行次数、成功率、延迟等指标。版本化管理图定义工作流图StateGraph的定义是核心业务逻辑。应将其作为代码的一部分进行版本控制Git。考虑将复杂的图拆分成模块化组件便于管理和测试。实现降级策略对于关键节点如 LLM 调用、外部 API 调用设计降级方案。例如当搜索工具失败时可以转而从本地知识库中检索或直接返回一个友好的提示信息而不是让整个工作流崩溃。人工审核环节对于涉及内容安全、重要决策或高成本操作的节点可以设计边路由到一个“人工审核”节点暂停工作流并通知审核人员待人工确认后再继续执行。这可以通过 LangGraph 的interrupt机制实现。全面的测试单元测试单独测试每个节点函数。集成测试测试整个工作流在不同输入下的行为。负载测试模拟高并发场景检查状态持久化后端和 LLM API 的承受能力。清晰的文档使用app.get_graph().draw_mermaid()生成的图表作为技术文档的一部分帮助团队成员理解智能体的决策流程。为每个节点和状态字段编写清晰的注释。LangGraph 将智能体开发从线性的链式思维提升到了图式的工作流思维为构建复杂、可靠、可维护的 AI 应用提供了强大的底层支持。从简单的条件路由到带有记忆和循环的复杂智能体其核心始终在于对状态和流程的精确控制。开始你的项目时建议从一个小而具体的工作流入手逐步迭代增加节点和复杂性并始终将可观测性和错误处理放在首位。