1. 项目概述LangGraph 为何能成为构建智能体应用的新基石如果你最近在关注AI应用开发尤其是智能体Agent领域那么“LangGraph”这个名字一定不会陌生。它不是一个独立的大模型而是一个由LangChain团队推出的、专门用于构建有状态、多智能体工作流的Python框架。简单来说LangGraph让你能用代码“画”出智能体的思维流程图清晰地定义它们如何接收信息、调用工具、进行决策以及彼此之间如何协作。von-development/awesome-LangGraph这个项目正是围绕这个强大框架的生态资源集合它像一本精心编纂的“黄页”和“实战手册”为开发者提供了从入门到精通所需的一切核心概念解析、官方与社区的最佳实践、丰富的示例项目、实用的工具扩展以及最新的讨论动态。为什么我们需要关注LangGraph在传统的链式Chain调用中AI应用的流程是线性的、确定性的就像一条单行线。但现实世界的问题尤其是需要复杂推理、多轮对话或协同作业的任务更像一个错综复杂的交通网络充满了条件判断、循环和并行处理。LangGraph引入了“图”Graph的概念将应用状态和智能体行为节点化、边化使得构建具备记忆、分支、循环和并行执行能力的复杂AI应用变得前所未有的直观和可控。无论是构建一个能深度分析文档并自主撰写报告的智能助手还是一个能协调多个专家模型解决复杂问题的多智能体系统LangGraph都提供了坚实的工程基础。2. 核心架构与设计哲学拆解2.1 有状态图LangGraph的灵魂所在LangGraph最核心的抽象是“有状态图”Stateful Graph。这与我们熟知的DAG有向无环图工作流引擎有本质区别。在LangGraph中图不仅仅定义任务的执行顺序更重要的是它维护并管理着一个共享的“状态”State。这个状态是一个字典或Pydantic模型随着图节点的执行而不断演化。你可以把整个图想象成一个智能体的“工作记忆白板”。每个节点都是一个函数它读取当前白板上的内容状态进行计算、推理或调用工具然后将修改后的结果写回白板。边Edge则定义了基于当前状态下一步应该执行哪个节点。这种设计完美契合了智能体应用的需求智能体需要记住之前的对话历史、工具调用结果、中间推理步骤并根据这些信息决定下一步行动。例如一个简单的问答智能体状态可能包含{messages: [...], question: ..., answer: ...}。一个“生成回答”的节点会读取messages和question调用LLM然后将结果写入answer字段。而一条边可能规定如果answer字段不为空则图执行结束否则继续循环。2.2 节点Nodes与边Edges构建智能体的乐高积木在LangGraph中构建应用就是定义节点和连接它们的有条件边。节点本质上是任何可调用对象函数或类方法。一个节点通常负责一项具体任务比如LLM调用节点将状态中的提示词发送给大模型并解析返回结果。工具执行节点根据智能体的决策调用一个外部工具如搜索API、计算器、数据库查询。状态处理节点对状态进行格式化、过滤或聚合。节点的设计应遵循“单一职责”原则这极大地提升了代码的可测试性和可复用性。边决定了流程的控制流。LangGraph支持三种主要的边起始边Start Edge定义图的入口点。条件边Conditional Edge这是实现智能体“思考”的关键。它根据当前状态的值动态决定下一个要执行的节点。这模拟了人类“如果...那么...”的决策过程。普通边Normal Edge无条件地指向下一个节点。通过组合条件边和节点你可以轻松构建出“思考-行动-观察”循环ReAct模式、多路分支选择乃至复杂的多智能体投票机制。2.3 与LangChain的共生关系很多人会混淆LangGraph和LangChain。你可以这样理解LangChain是一个庞大的“工具箱”和“脚手架”它提供了连接LLM、工具、记忆存储等组件的标准化接口和大量现成实现。而LangGraph是LangChain生态中一个专注于“编排”和“工作流”的精密“控制台”。在实践中你几乎总是会同时使用两者。LangGraph负责高层次的流程控制和状态管理而具体的LLM调用通过LangChain的LCEL、工具定义、向量检索等功能则由LangChain的组件来承担。awesome-LangGraph中的许多示例都清晰地展示了这种协作模式。注意虽然LangGraph脱胎于LangChain但它是一个独立的包langgraph可以单独安装和使用。它的设计更加聚焦和抽象理论上可以编排任何Python函数而不仅仅是LangChain的链。3. 从零开始构建你的第一个LangGraph智能体理论说得再多不如动手实践。让我们构建一个最简单的、具备“思考-行动”循环的智能体它可以使用搜索引擎来回答时效性问题。3.1 环境搭建与依赖安装首先创建一个干净的Python环境并安装核心依赖。建议使用Python 3.10或更高版本。# 创建并激活虚拟环境可选但推荐 python -m venv langgraph-env source langgraph-env/bin/activate # Linux/macOS # 或 langgraph-env\Scripts\activate # Windows # 安装核心包 pip install langgraph langchain langchain-openai tavily-python这里我们安装了langgraph: 核心框架。langchainlangchain-openai: 用于方便地调用OpenAI的LLM。tavily-python: 一个简单好用的搜索API工具你需要去其官网注册获取免费API KEY。你也可以使用LangChain支持的其他搜索工具如Serper、DuckDuckGo。设置你的API密钥。通常通过环境变量管理既安全又方便export OPENAI_API_KEYyour-openai-key export TAVILY_API_KEYyour-tavily-key或者在Python脚本中直接设置import os os.environ[OPENAI_API_KEY] your-openai-key os.environ[TAVILY_API_KEY] your-tavily-key3.2 定义状态与工具在LangGraph中我们首先需要定义状态的“形状”。我们将使用Pydantic的BaseModel来获得类型提示和验证的好处。from typing import TypedDict, List, Annotated import operator from langchain_openai import ChatOpenAI from langchain_community.tools.tavily_search import TavilySearchResults from langgraph.graph import StateGraph, END # 1. 定义状态结构 class AgentState(TypedDict): # 消息历史这是与LLM交互的核心 messages: Annotated[List, operator.add] # 用户提出的原始问题 question: str # 智能体思考的中间步骤可选用于调试或复杂推理 scratchpad: List[str] # 从网络搜索得到的结果 search_results: str接下来定义智能体可以使用的工具。这里我们创建一个搜索工具。# 2. 实例化LLM和工具 llm ChatOpenAI(modelgpt-4o-mini, temperature0) search_tool TavilySearchResults(max_results3) # 限制返回3条结果 # 为了方便将工具包装成LangChain格式的工具列表 tools [search_tool] llm_with_tools llm.bind_tools(tools)3.3 创建图节点思考、行动、观察我们将构建一个包含三个核心节点的图agent思考下一步、search执行搜索行动、generate_answer生成最终答案。# 3. 定义“agent”思考节点 def agent_node(state: AgentState): 这个节点负责思考。它根据当前对话历史和问题决定下一步是调用工具还是直接回答。 print(f[Agent Node] 正在思考... 当前消息数{len(state[messages])}) # 准备给LLM的提示消息。我们让系统扮演一个有帮助的助手并可以使用搜索工具。 system_message 你是一个有帮助的助手。你可以使用搜索工具来获取最新信息。如果用户的问题需要实时信息或你不确定就使用搜索工具。如果你已经掌握了足够信息就直接用友好的语气回答。 # 构建消息列表系统指令 历史消息 最新用户问题 prompt_messages [ {role: system, content: system_message}, *state[messages], # 包含之前的对话 {role: user, content: state[question]} ] # 调用绑定了工具的LLM response llm_with_tools.invoke(prompt_messages) # 将LLM的响应添加到消息历史中 new_messages [response] return {messages: new_messages} # 4. 定义“search”工具执行节点 def search_node(state: AgentState): 这个节点执行搜索工具并将结果格式化后存入状态。 print([Search Node] 正在执行搜索...) # 从最新的LLM响应中提取工具调用信息 last_message state[messages][-1] tool_calls last_message.tool_calls if not tool_calls: return {search_results: 未找到工具调用指令。} # 假设只调用一个工具搜索 tool_call tool_calls[0] if tool_call[name] tavily_search_results_json: # 执行搜索 search_query tool_call[args][query] results search_tool.invoke({query: search_query}) # 将搜索结果格式化为易读的字符串 formatted_results \n---\n.join([ f来源{r.get(title, N/A)}\n链接{r.get(url, N/A)}\n摘要{r.get(content, N/A)[:200]}... for r in results ]) return {search_results: formatted_results, scratchpad: [f搜索了{search_query}]} return {search_results: 工具调用不匹配。} # 5. 定义“generate_answer”最终回答节点 def generate_answer_node(state: AgentState): 当智能体决定不再调用工具时由此节点综合所有信息生成最终回答。 print([Generate Answer Node] 正在生成最终答案...) # 准备给LLM的上下文历史对话 搜索结果 context f 用户的问题{state[question]} 搜索到的相关信息 {state.get(search_results, 无)} final_prompt [ {role: system, content: 你是一个有帮助的助手。请根据提供的上下文信息清晰、准确、友好地回答用户的问题。如果信息不足请诚实说明。}, {role: user, content: context} ] response llm.invoke(final_prompt) # 将最终答案作为一条新消息加入历史 final_message {role: assistant, content: response.content} return {messages: [final_message]}3.4 编排图流程与条件路由现在我们将节点组装起来并定义它们之间的流转逻辑。# 6. 创建图构建器 workflow StateGraph(AgentState) # 7. 添加节点 workflow.add_node(agent, agent_node) workflow.add_node(search, search_node) workflow.add_node(generate_answer, generate_answer_node) # 8. 设置入口点从“agent”思考开始 workflow.set_entry_point(agent) # 9. 定义条件路由函数决定下一步是继续搜索还是生成答案 def should_continue(state: AgentState) - str: 根据LLM的响应判断下一步。 如果LLM调用了工具就去‘search’节点。 如果LLM直接回复了就去‘generate_answer’节点。 last_message state[messages][-1] if last_message.tool_calls: return search # 有工具调用去执行搜索 else: return generate_answer # 没有工具调用去生成最终答案 # 10. 添加从“agent”出发的条件边 workflow.add_conditional_edges( agent, should_continue, { search: search, generate_answer: generate_answer } ) # 11. 添加从“search”返回“agent”的边执行完搜索后继续思考 workflow.add_edge(search, agent) # 12. 添加从“generate_answer”到结束的边 workflow.add_edge(generate_answer, END) # 13. 编译图 app workflow.compile()3.5 运行与测试现在让我们运行这个智能体问它一个需要最新信息的问题。# 14. 初始化状态并运行图 initial_state: AgentState { messages: [], # 初始对话历史为空 question: 2024年巴黎奥运会中国代表团获得了多少枚金牌, scratchpad: [], search_results: } # 运行图 final_state app.invoke(initial_state, config{recursion_limit: 10}) # 限制递归深度防止死循环 # 打印最终答案 print(\n *50) print(最终回答) for msg in final_state[messages]: if msg[role] assistant and content in msg: print(msg[content]) print(*50)运行上述代码你会在控制台看到类似以下的执行流程[Agent Node] 正在思考... 当前消息数0 [Search Node] 正在执行搜索... [Agent Node] 正在思考... 当前消息数2 [Generate Answer Node] 正在生成最终答案... 最终回答 根据最新的搜索结果在2024年巴黎奥运会上中国体育代表团表现出色共获得了40枚金牌...这个简单的例子展示了LangGraph的核心魅力你将智能体的决策逻辑是否调用工具清晰地编码在了should_continue函数和图的边中而不是隐藏在冗长的提示词或复杂的if-else代码里。整个流程一目了然易于调试和扩展。4. 进阶模式与最佳实践解析掌握了基础构建后我们来看看awesome-LangGraph项目中汇集的一些更强大的模式和最佳实践。4.1 多智能体协作模式这是LangGraph的杀手级应用。你可以创建多个具备不同专长的智能体如“研究员”、“写手”、“校对员”让它们通过共享的状态和预定义的交互规则进行协作。实现要点定义角色与专用工具为每个智能体创建独立的LLM配置和工具集。研究员可能拥有搜索和文献分析工具写手拥有文本风格化工具校对员拥有语法检查工具。设计状态结构状态中需要包含能区分不同智能体贡献的字段例如{draft: , research_notes: , editor_feedback: , current_agent: researcher}。实现路由逻辑通过条件边根据current_agent或任务完成情况如draft是否达到某个质量标准将控制权移交给下一个智能体。管理对话历史需要仔细设计消息历史避免不同智能体的消息互相干扰。一种常见做法是为每个智能体维护独立的历史或在消息中明确标注发送者。# 简化的多智能体状态示例 class MultiAgentState(TypedDict): task: str researcher_output: str writer_output: str reviewer_comments: str final_output: str step: str # 用于控制流程research, write, review, done4.2 持久化与检查点对于长时间运行或需要中断恢复的智能体工作流状态持久化至关重要。LangGraph内置了检查点Checkpoint机制。核心操作保存检查点在图执行到特定节点后自动将完整状态序列化保存到数据库如SQLite、PostgreSQL或内存中。从检查点恢复当需要继续运行时可以加载最后一个检查点的状态让图从中断处继续执行。版本与分支高级用法支持从历史检查点创建分支实现复杂的实验和回滚。这为构建可暂停、可恢复、可审计的长期对话助手或自动化流程提供了基础。4.3 流式输出与用户体验对于前端应用让用户实时看到智能体的“思考过程”能极大提升体验。LangGraph支持流式输出。实现方式节点级流式在节点函数中使用yield关键字逐步返回部分结果。例如在generate_answer_node中你可以让LLM以流式方式生成文本并逐步更新状态中的partial_answer字段。图级流式通过app.stream()方法调用图。它会返回一个异步生成器每次状态更新都会yield出来。前端可以通过SSEServer-Sent Events或WebSocket监听这些更新并实时渲染。# 流式调用示例 async for step in app.astream(initial_state, config{recursion_limit: 10}): node_name step[0] # 节点名 node_output step[1] # 节点输出 # 在这里处理流式更新例如发送到前端 if node_name agent and content in node_output.get(messages, [{}])[-1]: print(f思考中: {node_output[messages][-1][content][:50]}...)4.4 错误处理与韧性在生产环境中智能体可能遇到工具API失败、LLM输出格式错误、网络超时等问题。LangGraph提供了几种错误处理策略节点级Try-Catch在每个节点函数内部进行细致的异常捕获并返回一个表示错误的状态更新如{error: API调用失败, should_retry: True}。条件边路由到错误处理节点在should_continue这类路由函数中检查状态中是否有错误标志如果有则路由到一个专用的error_handler节点。该节点可以记录日志、尝试重试或向用户返回友好的错误信息。设置超时与重试对于工具调用节点可以使用tenacity等重试库进行装饰或利用LangChain工具内置的重试机制。5. 常见问题、调试技巧与性能优化在实际开发中你肯定会遇到各种挑战。以下是一些从社区和实战中总结的常见问题与解决方案。5.1 状态管理混乱问题状态字典变得过于庞大和复杂难以维护和调试不同节点意外修改了不该修改的字段。解决方案使用Pydantic BaseModel代替TypedDict这能提供严格的类型验证和清晰的架构定义。在编译图时LangGraph能利用Pydantic模型进行序列化和验证。状态划分将状态划分为几个逻辑子模块。例如UserContext、ConversationHistory、ToolResults。这可以通过嵌套的Pydantic模型实现。写时复制模式对于列表等可变对象在节点中返回更新后的新对象而不是直接修改原状态。Annotated[List, operator.add]注解实际上就是这种模式的简化实现它要求节点返回一个列表该列表会与状态中的原列表相加。5.2 图陷入无限循环问题智能体在“思考-行动”循环中出不来或者在不同节点间来回跳转。调试步骤设置递归限制在调用app.invoke()时始终传入config{recursion_limit: N}。这是最重要的安全阀。添加日志在每个节点的开始和结束处打印状态的关键字段。使用print(f”[Node: {node_name}] State keys: {state.keys()}”)。可视化你的图LangGraph提供了workflow.get_graph().draw_mermaid()功能可以生成Mermaid图表代码将其粘贴到Mermaid在线编辑器中能直观看到你的流程设计是否有逻辑闭环。审查条件边逻辑仔细检查should_continue或类似的路由函数。确保在所有可能的状态下都有明确的出口指向END或某个能结束循环的节点。一个常见的错误是当工具调用结果为空或不符合预期时路由逻辑没有处理好导致智能体反复尝试同一个操作。5.3 LLM调用成本与延迟过高问题复杂的图会导致多次调用LLM成本和响应时间激增。优化策略缓存对LLM调用和工具调用实施缓存。LangChain集成了多种缓存后端InMemory, Redis, SQLite。对于相同输入产生相同输出的节点如一些信息提取节点缓存效果显著。优化提示词确保发给LLM的提示词精简且有效。避免在每次调用时传递冗长的、不变的系统指令。可以考虑将其固化在LLM的初始化配置中。并行化如果图中有多个独立的节点例如同时调用两个不同的搜索API可以探索使用add_edge的并行模式或者利用asyncio在自定义节点中实现并发。但要注意这可能会让状态管理变得更复杂。使用更小/更快的模型对于简单的路由决策或工具调用解析可以考虑使用gpt-3.5-turbo或更小的开源模型将gpt-4等大模型仅用于需要深度推理和创造性的核心节点。5.4 工具调用解析失败问题LLM没有按照预期格式返回工具调用信息导致search_node等工具执行节点崩溃。加固方法使用LangChain的bind_tools如示例所示这能显著提高工具调用的格式合规性。结构化输出对于关键信息提取使用LLM的“结构化输出”功能如OpenAI的JSON Mode或LangChain的with_structured_output强制返回一个指定JSON schema的对象而不是自由文本。后处理与降级在工具执行节点中对LLM的输出进行健壮的解析。使用try-except块如果解析失败则返回一个错误状态并由错误处理节点决定是重试、换一种方式提问还是直接向用户请求澄清。5.5 如何测试LangGraph应用测试有状态、非确定性的智能体应用颇具挑战但并非不可能。单元测试节点函数每个节点都是一个纯函数给定输入状态返回输出状态。你可以单独测试它们用模拟的输入状态验证其输出是否符合预期。集成测试固定流程对于确定的子图可以用固定的输入状态调用并断言最终的输出状态。由于LLM的非确定性你可能需要mock LLM的返回使用LangChain的ChatOpenAImocking功能或类似pytest-mock的工具。端到端测试与评估对于整个智能体定义一组具有标准答案的测试用例。运行整个图并使用LLM作为“裁判”或基于规则的检查器来评估最终答案的质量、工具调用的正确性等。这更接近验收测试。构建基于LangGraph的智能体应用是一个将模糊的智能体概念工程化、产品化的过程。它迫使你清晰地定义状态、决策点和交互流程。awesome-LangGraph项目正是这样一个宝库它收集了社区在探索过程中沉淀下来的各种模式、技巧和实战代码。从简单的循环智能体到复杂的多智能体协作系统这个框架正在成为新一代AI应用开发的事实标准之一。我的体会是开始可能会觉得有些抽象但一旦理解了“状态流”这一核心概念你就会发现用它来建模复杂业务逻辑是如此的自然和强大。最后一个小建议在开发复杂图时养成先画流程图再写代码的习惯这能帮你理清思路事半功倍。
LangGraph框架：构建有状态多智能体工作流的Python实践指南
1. 项目概述LangGraph 为何能成为构建智能体应用的新基石如果你最近在关注AI应用开发尤其是智能体Agent领域那么“LangGraph”这个名字一定不会陌生。它不是一个独立的大模型而是一个由LangChain团队推出的、专门用于构建有状态、多智能体工作流的Python框架。简单来说LangGraph让你能用代码“画”出智能体的思维流程图清晰地定义它们如何接收信息、调用工具、进行决策以及彼此之间如何协作。von-development/awesome-LangGraph这个项目正是围绕这个强大框架的生态资源集合它像一本精心编纂的“黄页”和“实战手册”为开发者提供了从入门到精通所需的一切核心概念解析、官方与社区的最佳实践、丰富的示例项目、实用的工具扩展以及最新的讨论动态。为什么我们需要关注LangGraph在传统的链式Chain调用中AI应用的流程是线性的、确定性的就像一条单行线。但现实世界的问题尤其是需要复杂推理、多轮对话或协同作业的任务更像一个错综复杂的交通网络充满了条件判断、循环和并行处理。LangGraph引入了“图”Graph的概念将应用状态和智能体行为节点化、边化使得构建具备记忆、分支、循环和并行执行能力的复杂AI应用变得前所未有的直观和可控。无论是构建一个能深度分析文档并自主撰写报告的智能助手还是一个能协调多个专家模型解决复杂问题的多智能体系统LangGraph都提供了坚实的工程基础。2. 核心架构与设计哲学拆解2.1 有状态图LangGraph的灵魂所在LangGraph最核心的抽象是“有状态图”Stateful Graph。这与我们熟知的DAG有向无环图工作流引擎有本质区别。在LangGraph中图不仅仅定义任务的执行顺序更重要的是它维护并管理着一个共享的“状态”State。这个状态是一个字典或Pydantic模型随着图节点的执行而不断演化。你可以把整个图想象成一个智能体的“工作记忆白板”。每个节点都是一个函数它读取当前白板上的内容状态进行计算、推理或调用工具然后将修改后的结果写回白板。边Edge则定义了基于当前状态下一步应该执行哪个节点。这种设计完美契合了智能体应用的需求智能体需要记住之前的对话历史、工具调用结果、中间推理步骤并根据这些信息决定下一步行动。例如一个简单的问答智能体状态可能包含{messages: [...], question: ..., answer: ...}。一个“生成回答”的节点会读取messages和question调用LLM然后将结果写入answer字段。而一条边可能规定如果answer字段不为空则图执行结束否则继续循环。2.2 节点Nodes与边Edges构建智能体的乐高积木在LangGraph中构建应用就是定义节点和连接它们的有条件边。节点本质上是任何可调用对象函数或类方法。一个节点通常负责一项具体任务比如LLM调用节点将状态中的提示词发送给大模型并解析返回结果。工具执行节点根据智能体的决策调用一个外部工具如搜索API、计算器、数据库查询。状态处理节点对状态进行格式化、过滤或聚合。节点的设计应遵循“单一职责”原则这极大地提升了代码的可测试性和可复用性。边决定了流程的控制流。LangGraph支持三种主要的边起始边Start Edge定义图的入口点。条件边Conditional Edge这是实现智能体“思考”的关键。它根据当前状态的值动态决定下一个要执行的节点。这模拟了人类“如果...那么...”的决策过程。普通边Normal Edge无条件地指向下一个节点。通过组合条件边和节点你可以轻松构建出“思考-行动-观察”循环ReAct模式、多路分支选择乃至复杂的多智能体投票机制。2.3 与LangChain的共生关系很多人会混淆LangGraph和LangChain。你可以这样理解LangChain是一个庞大的“工具箱”和“脚手架”它提供了连接LLM、工具、记忆存储等组件的标准化接口和大量现成实现。而LangGraph是LangChain生态中一个专注于“编排”和“工作流”的精密“控制台”。在实践中你几乎总是会同时使用两者。LangGraph负责高层次的流程控制和状态管理而具体的LLM调用通过LangChain的LCEL、工具定义、向量检索等功能则由LangChain的组件来承担。awesome-LangGraph中的许多示例都清晰地展示了这种协作模式。注意虽然LangGraph脱胎于LangChain但它是一个独立的包langgraph可以单独安装和使用。它的设计更加聚焦和抽象理论上可以编排任何Python函数而不仅仅是LangChain的链。3. 从零开始构建你的第一个LangGraph智能体理论说得再多不如动手实践。让我们构建一个最简单的、具备“思考-行动”循环的智能体它可以使用搜索引擎来回答时效性问题。3.1 环境搭建与依赖安装首先创建一个干净的Python环境并安装核心依赖。建议使用Python 3.10或更高版本。# 创建并激活虚拟环境可选但推荐 python -m venv langgraph-env source langgraph-env/bin/activate # Linux/macOS # 或 langgraph-env\Scripts\activate # Windows # 安装核心包 pip install langgraph langchain langchain-openai tavily-python这里我们安装了langgraph: 核心框架。langchainlangchain-openai: 用于方便地调用OpenAI的LLM。tavily-python: 一个简单好用的搜索API工具你需要去其官网注册获取免费API KEY。你也可以使用LangChain支持的其他搜索工具如Serper、DuckDuckGo。设置你的API密钥。通常通过环境变量管理既安全又方便export OPENAI_API_KEYyour-openai-key export TAVILY_API_KEYyour-tavily-key或者在Python脚本中直接设置import os os.environ[OPENAI_API_KEY] your-openai-key os.environ[TAVILY_API_KEY] your-tavily-key3.2 定义状态与工具在LangGraph中我们首先需要定义状态的“形状”。我们将使用Pydantic的BaseModel来获得类型提示和验证的好处。from typing import TypedDict, List, Annotated import operator from langchain_openai import ChatOpenAI from langchain_community.tools.tavily_search import TavilySearchResults from langgraph.graph import StateGraph, END # 1. 定义状态结构 class AgentState(TypedDict): # 消息历史这是与LLM交互的核心 messages: Annotated[List, operator.add] # 用户提出的原始问题 question: str # 智能体思考的中间步骤可选用于调试或复杂推理 scratchpad: List[str] # 从网络搜索得到的结果 search_results: str接下来定义智能体可以使用的工具。这里我们创建一个搜索工具。# 2. 实例化LLM和工具 llm ChatOpenAI(modelgpt-4o-mini, temperature0) search_tool TavilySearchResults(max_results3) # 限制返回3条结果 # 为了方便将工具包装成LangChain格式的工具列表 tools [search_tool] llm_with_tools llm.bind_tools(tools)3.3 创建图节点思考、行动、观察我们将构建一个包含三个核心节点的图agent思考下一步、search执行搜索行动、generate_answer生成最终答案。# 3. 定义“agent”思考节点 def agent_node(state: AgentState): 这个节点负责思考。它根据当前对话历史和问题决定下一步是调用工具还是直接回答。 print(f[Agent Node] 正在思考... 当前消息数{len(state[messages])}) # 准备给LLM的提示消息。我们让系统扮演一个有帮助的助手并可以使用搜索工具。 system_message 你是一个有帮助的助手。你可以使用搜索工具来获取最新信息。如果用户的问题需要实时信息或你不确定就使用搜索工具。如果你已经掌握了足够信息就直接用友好的语气回答。 # 构建消息列表系统指令 历史消息 最新用户问题 prompt_messages [ {role: system, content: system_message}, *state[messages], # 包含之前的对话 {role: user, content: state[question]} ] # 调用绑定了工具的LLM response llm_with_tools.invoke(prompt_messages) # 将LLM的响应添加到消息历史中 new_messages [response] return {messages: new_messages} # 4. 定义“search”工具执行节点 def search_node(state: AgentState): 这个节点执行搜索工具并将结果格式化后存入状态。 print([Search Node] 正在执行搜索...) # 从最新的LLM响应中提取工具调用信息 last_message state[messages][-1] tool_calls last_message.tool_calls if not tool_calls: return {search_results: 未找到工具调用指令。} # 假设只调用一个工具搜索 tool_call tool_calls[0] if tool_call[name] tavily_search_results_json: # 执行搜索 search_query tool_call[args][query] results search_tool.invoke({query: search_query}) # 将搜索结果格式化为易读的字符串 formatted_results \n---\n.join([ f来源{r.get(title, N/A)}\n链接{r.get(url, N/A)}\n摘要{r.get(content, N/A)[:200]}... for r in results ]) return {search_results: formatted_results, scratchpad: [f搜索了{search_query}]} return {search_results: 工具调用不匹配。} # 5. 定义“generate_answer”最终回答节点 def generate_answer_node(state: AgentState): 当智能体决定不再调用工具时由此节点综合所有信息生成最终回答。 print([Generate Answer Node] 正在生成最终答案...) # 准备给LLM的上下文历史对话 搜索结果 context f 用户的问题{state[question]} 搜索到的相关信息 {state.get(search_results, 无)} final_prompt [ {role: system, content: 你是一个有帮助的助手。请根据提供的上下文信息清晰、准确、友好地回答用户的问题。如果信息不足请诚实说明。}, {role: user, content: context} ] response llm.invoke(final_prompt) # 将最终答案作为一条新消息加入历史 final_message {role: assistant, content: response.content} return {messages: [final_message]}3.4 编排图流程与条件路由现在我们将节点组装起来并定义它们之间的流转逻辑。# 6. 创建图构建器 workflow StateGraph(AgentState) # 7. 添加节点 workflow.add_node(agent, agent_node) workflow.add_node(search, search_node) workflow.add_node(generate_answer, generate_answer_node) # 8. 设置入口点从“agent”思考开始 workflow.set_entry_point(agent) # 9. 定义条件路由函数决定下一步是继续搜索还是生成答案 def should_continue(state: AgentState) - str: 根据LLM的响应判断下一步。 如果LLM调用了工具就去‘search’节点。 如果LLM直接回复了就去‘generate_answer’节点。 last_message state[messages][-1] if last_message.tool_calls: return search # 有工具调用去执行搜索 else: return generate_answer # 没有工具调用去生成最终答案 # 10. 添加从“agent”出发的条件边 workflow.add_conditional_edges( agent, should_continue, { search: search, generate_answer: generate_answer } ) # 11. 添加从“search”返回“agent”的边执行完搜索后继续思考 workflow.add_edge(search, agent) # 12. 添加从“generate_answer”到结束的边 workflow.add_edge(generate_answer, END) # 13. 编译图 app workflow.compile()3.5 运行与测试现在让我们运行这个智能体问它一个需要最新信息的问题。# 14. 初始化状态并运行图 initial_state: AgentState { messages: [], # 初始对话历史为空 question: 2024年巴黎奥运会中国代表团获得了多少枚金牌, scratchpad: [], search_results: } # 运行图 final_state app.invoke(initial_state, config{recursion_limit: 10}) # 限制递归深度防止死循环 # 打印最终答案 print(\n *50) print(最终回答) for msg in final_state[messages]: if msg[role] assistant and content in msg: print(msg[content]) print(*50)运行上述代码你会在控制台看到类似以下的执行流程[Agent Node] 正在思考... 当前消息数0 [Search Node] 正在执行搜索... [Agent Node] 正在思考... 当前消息数2 [Generate Answer Node] 正在生成最终答案... 最终回答 根据最新的搜索结果在2024年巴黎奥运会上中国体育代表团表现出色共获得了40枚金牌...这个简单的例子展示了LangGraph的核心魅力你将智能体的决策逻辑是否调用工具清晰地编码在了should_continue函数和图的边中而不是隐藏在冗长的提示词或复杂的if-else代码里。整个流程一目了然易于调试和扩展。4. 进阶模式与最佳实践解析掌握了基础构建后我们来看看awesome-LangGraph项目中汇集的一些更强大的模式和最佳实践。4.1 多智能体协作模式这是LangGraph的杀手级应用。你可以创建多个具备不同专长的智能体如“研究员”、“写手”、“校对员”让它们通过共享的状态和预定义的交互规则进行协作。实现要点定义角色与专用工具为每个智能体创建独立的LLM配置和工具集。研究员可能拥有搜索和文献分析工具写手拥有文本风格化工具校对员拥有语法检查工具。设计状态结构状态中需要包含能区分不同智能体贡献的字段例如{draft: , research_notes: , editor_feedback: , current_agent: researcher}。实现路由逻辑通过条件边根据current_agent或任务完成情况如draft是否达到某个质量标准将控制权移交给下一个智能体。管理对话历史需要仔细设计消息历史避免不同智能体的消息互相干扰。一种常见做法是为每个智能体维护独立的历史或在消息中明确标注发送者。# 简化的多智能体状态示例 class MultiAgentState(TypedDict): task: str researcher_output: str writer_output: str reviewer_comments: str final_output: str step: str # 用于控制流程research, write, review, done4.2 持久化与检查点对于长时间运行或需要中断恢复的智能体工作流状态持久化至关重要。LangGraph内置了检查点Checkpoint机制。核心操作保存检查点在图执行到特定节点后自动将完整状态序列化保存到数据库如SQLite、PostgreSQL或内存中。从检查点恢复当需要继续运行时可以加载最后一个检查点的状态让图从中断处继续执行。版本与分支高级用法支持从历史检查点创建分支实现复杂的实验和回滚。这为构建可暂停、可恢复、可审计的长期对话助手或自动化流程提供了基础。4.3 流式输出与用户体验对于前端应用让用户实时看到智能体的“思考过程”能极大提升体验。LangGraph支持流式输出。实现方式节点级流式在节点函数中使用yield关键字逐步返回部分结果。例如在generate_answer_node中你可以让LLM以流式方式生成文本并逐步更新状态中的partial_answer字段。图级流式通过app.stream()方法调用图。它会返回一个异步生成器每次状态更新都会yield出来。前端可以通过SSEServer-Sent Events或WebSocket监听这些更新并实时渲染。# 流式调用示例 async for step in app.astream(initial_state, config{recursion_limit: 10}): node_name step[0] # 节点名 node_output step[1] # 节点输出 # 在这里处理流式更新例如发送到前端 if node_name agent and content in node_output.get(messages, [{}])[-1]: print(f思考中: {node_output[messages][-1][content][:50]}...)4.4 错误处理与韧性在生产环境中智能体可能遇到工具API失败、LLM输出格式错误、网络超时等问题。LangGraph提供了几种错误处理策略节点级Try-Catch在每个节点函数内部进行细致的异常捕获并返回一个表示错误的状态更新如{error: API调用失败, should_retry: True}。条件边路由到错误处理节点在should_continue这类路由函数中检查状态中是否有错误标志如果有则路由到一个专用的error_handler节点。该节点可以记录日志、尝试重试或向用户返回友好的错误信息。设置超时与重试对于工具调用节点可以使用tenacity等重试库进行装饰或利用LangChain工具内置的重试机制。5. 常见问题、调试技巧与性能优化在实际开发中你肯定会遇到各种挑战。以下是一些从社区和实战中总结的常见问题与解决方案。5.1 状态管理混乱问题状态字典变得过于庞大和复杂难以维护和调试不同节点意外修改了不该修改的字段。解决方案使用Pydantic BaseModel代替TypedDict这能提供严格的类型验证和清晰的架构定义。在编译图时LangGraph能利用Pydantic模型进行序列化和验证。状态划分将状态划分为几个逻辑子模块。例如UserContext、ConversationHistory、ToolResults。这可以通过嵌套的Pydantic模型实现。写时复制模式对于列表等可变对象在节点中返回更新后的新对象而不是直接修改原状态。Annotated[List, operator.add]注解实际上就是这种模式的简化实现它要求节点返回一个列表该列表会与状态中的原列表相加。5.2 图陷入无限循环问题智能体在“思考-行动”循环中出不来或者在不同节点间来回跳转。调试步骤设置递归限制在调用app.invoke()时始终传入config{recursion_limit: N}。这是最重要的安全阀。添加日志在每个节点的开始和结束处打印状态的关键字段。使用print(f”[Node: {node_name}] State keys: {state.keys()}”)。可视化你的图LangGraph提供了workflow.get_graph().draw_mermaid()功能可以生成Mermaid图表代码将其粘贴到Mermaid在线编辑器中能直观看到你的流程设计是否有逻辑闭环。审查条件边逻辑仔细检查should_continue或类似的路由函数。确保在所有可能的状态下都有明确的出口指向END或某个能结束循环的节点。一个常见的错误是当工具调用结果为空或不符合预期时路由逻辑没有处理好导致智能体反复尝试同一个操作。5.3 LLM调用成本与延迟过高问题复杂的图会导致多次调用LLM成本和响应时间激增。优化策略缓存对LLM调用和工具调用实施缓存。LangChain集成了多种缓存后端InMemory, Redis, SQLite。对于相同输入产生相同输出的节点如一些信息提取节点缓存效果显著。优化提示词确保发给LLM的提示词精简且有效。避免在每次调用时传递冗长的、不变的系统指令。可以考虑将其固化在LLM的初始化配置中。并行化如果图中有多个独立的节点例如同时调用两个不同的搜索API可以探索使用add_edge的并行模式或者利用asyncio在自定义节点中实现并发。但要注意这可能会让状态管理变得更复杂。使用更小/更快的模型对于简单的路由决策或工具调用解析可以考虑使用gpt-3.5-turbo或更小的开源模型将gpt-4等大模型仅用于需要深度推理和创造性的核心节点。5.4 工具调用解析失败问题LLM没有按照预期格式返回工具调用信息导致search_node等工具执行节点崩溃。加固方法使用LangChain的bind_tools如示例所示这能显著提高工具调用的格式合规性。结构化输出对于关键信息提取使用LLM的“结构化输出”功能如OpenAI的JSON Mode或LangChain的with_structured_output强制返回一个指定JSON schema的对象而不是自由文本。后处理与降级在工具执行节点中对LLM的输出进行健壮的解析。使用try-except块如果解析失败则返回一个错误状态并由错误处理节点决定是重试、换一种方式提问还是直接向用户请求澄清。5.5 如何测试LangGraph应用测试有状态、非确定性的智能体应用颇具挑战但并非不可能。单元测试节点函数每个节点都是一个纯函数给定输入状态返回输出状态。你可以单独测试它们用模拟的输入状态验证其输出是否符合预期。集成测试固定流程对于确定的子图可以用固定的输入状态调用并断言最终的输出状态。由于LLM的非确定性你可能需要mock LLM的返回使用LangChain的ChatOpenAImocking功能或类似pytest-mock的工具。端到端测试与评估对于整个智能体定义一组具有标准答案的测试用例。运行整个图并使用LLM作为“裁判”或基于规则的检查器来评估最终答案的质量、工具调用的正确性等。这更接近验收测试。构建基于LangGraph的智能体应用是一个将模糊的智能体概念工程化、产品化的过程。它迫使你清晰地定义状态、决策点和交互流程。awesome-LangGraph项目正是这样一个宝库它收集了社区在探索过程中沉淀下来的各种模式、技巧和实战代码。从简单的循环智能体到复杂的多智能体协作系统这个框架正在成为新一代AI应用开发的事实标准之一。我的体会是开始可能会觉得有些抽象但一旦理解了“状态流”这一核心概念你就会发现用它来建模复杂业务逻辑是如此的自然和强大。最后一个小建议在开发复杂图时养成先画流程图再写代码的习惯这能帮你理清思路事半功倍。
