最近在尝试构建一个能自主处理复杂任务的AI智能体时你是否也遇到过这样的困境Agent的逻辑流难以编排状态管理混乱记忆能力时有时无调试起来更是让人头疼。传统的LangChain Agent虽然强大但在处理多步骤、有状态的复杂工作流时往往显得力不从心。这正是LangGraph要解决的问题。本文将为你系统拆解如何结合LangGraph、Langchain和Agent从零构建一个具备长期记忆和复杂推理能力的AI应用。无论你是想入门AI应用开发还是希望将现有项目升级为更智能的Agent系统这套保姆级教程都能提供从概念到部署的完整路径。1. 背景与核心概念为什么需要LangGraph在深入代码之前我们首先要理解这几个核心组件分别解决了什么问题以及它们如何协同工作。1.1 AI Agent与LangChain智能体的基石AI Agent智能体不仅仅是一个聊天机器人。它是一个能够感知环境、进行决策并执行动作以实现特定目标的软件实体。一个典型的Agent包含几个关键部分工具ToolsAgent可以调用的外部能力如搜索网络、执行计算、读写文件。记忆Memory存储对话历史、中间结果和知识使Agent具备上下文感知能力。规划Planning将复杂目标拆解为可执行的步骤序列。执行Execution调用工具并处理结果。LangChain是一个用于开发由大语言模型驱动的应用程序的框架。它提供了构建Agent所需的基础模块标准化接口统一不同大模型如OpenAI、Anthropic、本地模型的调用方式。丰富的工具链内置了大量实用工具搜索引擎、计算器、API调用等也支持自定义。基础记忆系统提供了对话缓存、向量存储等记忆后端。然而当任务流程包含条件分支、循环或需要维护复杂状态时仅用LangChain原生的AgentExecutor会变得非常笨拙。1.2 LangGraph为智能体引入“工作流引擎”这就是LangGraph的用武之地。你可以把它理解为专门为LangChain Agent设计的“工作流引擎”或“状态机框架”。它基于流行的网络分析库NetworkX构建核心思想是将Agent的执行过程建模为一个有向图Graph。节点Nodes代表一个执行单元可以是一个工具调用、一个条件判断或者一次LLM调用。边Edges定义了节点之间的流转路径。边可以是固定的也可以根据上一个节点的输出结果动态决定条件边。状态State一个贯穿整个图执行过程的共享字典。每个节点都可以读取和修改状态这是实现长期记忆和复杂信息传递的关键。LangGraph与LangChain Agent的核心区别LangChain Agent通常是线性的“思考-行动-观察”循环难以处理非线性的任务流。LangGraph允许你显式地定义包含分支、循环、并行和子流程的复杂工作流对执行过程拥有更精细的控制。1.3 核心价值可预测的状态管理与循环LangGraph带来的最大好处是状态管理的清晰化和循环与分支的原生支持。例如构建一个“研究助手”Agent流程可能是1) 理解问题2) 搜索网络3) 判断信息是否足够分支如果不足循环回步骤2如果足够则4) 撰写报告。用LangGraph可以非常直观地建模这个流程并确保“已收集信息”这个状态在整个循环中得以保持和更新。2. 环境准备与版本说明在开始实战之前我们需要搭建一个稳定的开发环境。本文将使用Python作为开发语言。2.1 基础环境配置建议使用Python 3.10或更高版本以获得最佳的兼容性。我们使用venv创建虚拟环境来隔离项目依赖。# 1. 创建项目目录并进入 mkdir langgraph-agent-tutorial cd langgraph-agent-tutorial # 2. 创建Python虚拟环境 python -m venv venv # 3. 激活虚拟环境 # 在Windows上 venv\Scripts\activate # 在macOS/Linux上 source venv/bin/activate # 4. 升级pip pip install --upgrade pip2.2 依赖安装创建requirements.txt文件并填入以下核心依赖。注意LangGraph和LangChain版本迭代较快以下版本为当前推荐组合。# requirements.txt langchain0.1.0 langchain-community0.0.10 langgraph0.0.26 langchain-openai0.0.5 # 用于OpenAI模型 openai1.6.1 # OpenAI官方SDK tiktoken0.5.2 # 用于Token计数 python-dotenv1.0.0 # 用于管理环境变量安装依赖pip install -r requirements.txt重要提示如果你计划使用本地大模型如通过Ollama部署的Llama 2、Mistral等则需要安装langchain-ollama等对应社区包并注释掉langchain-openai。本文示例将主要使用OpenAI GPT模型进行演示因为其稳定性和可用性较高。2.3 配置API密钥为了调用OpenAI的模型你需要一个API密钥。建议使用.env文件管理敏感信息避免将其硬编码在代码中。在项目根目录创建.env文件。在文件中添加你的OpenAI API密钥OPENAI_API_KEYsk-your-actual-openai-api-key-here在代码中通过python-dotenv加载from dotenv import load_dotenv load_dotenv() # 这会从.env文件加载环境变量到os.environ # 现在可以通过 os.getenv(OPENAI_API_KEY) 获取密钥3. 核心概念拆解图、状态与节点理解LangGraph的三个核心抽象是编写代码的基础。3.1 状态State数据的共享总线在LangGraph中状态是一个类似字典的对象它在图的所有节点之间传递和共享。通常我们使用TypedDict来定义状态的模式这有助于类型提示和代码清晰度。from typing import TypedDict, List, Annotated import operator class AgentState(TypedDict): # 用户的原始问题 input: str # Agent的“思考”过程或中间答案 thoughts: Annotated[List[str], operator.add] # 关键这是一个追加式列表 # 从工具获得的结果 tool_outputs: Annotated[List[str], operator.add] # 最终给用户的回答 final_output: str关键点Annotated[List[str], operator.add]。这个注解是LangGraph的魔法所在。它告诉框架当多个节点修改thoughts字段时不是覆盖而是将它们的输出追加add到同一个列表中。这为实现“记忆”提供了基础。3.2 节点Node执行单元节点是一个普通的Python函数它接收当前State对其进行修改然后返回更新后的State或其中的一部分。def llm_think_node(state: AgentState) - dict: 一个模拟LLM思考的节点 # 从状态中获取信息 question state[“input”] # 这里模拟LLM的思考过程。实际应用中你会在这里调用ChatOpenAI等模型。 new_thought f“我正在思考问题{question}。我需要搜索更多信息。” # 返回要更新到状态中的部分 return {“thoughts”: [new_thought]} def tool_call_node(state: AgentState) - dict: 一个模拟调用工具的节点 # 模拟调用一个网络搜索工具 search_result “根据网络搜索LangGraph是一个用于构建有状态多Actor应用的工作流库。” return {“tool_outputs”: [search_result]}3.3 边Edge与图Graph编排流程边决定了执行完一个节点后下一步该去哪个节点。ConditionalEdge允许根据状态内容动态决定路径。from langgraph.graph import StateGraph, END # 1. 创建图的工作流构建器 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(“think”, llm_think_node) workflow.add_node(“act”, tool_call_node) workflow.add_node(“finish”, final_response_node) # 假设已定义 # 3. 设置入口点 workflow.set_entry_point(“think”) # 4. 添加普通边无条件转移 workflow.add_edge(“think”, “act”) # 5. 添加条件边 from langgraph.graph import ConditionalEdge def should_continue(state: AgentState) - str: # 根据工具结果判断是否需要继续思考 if “信息不足” in state.get(“tool_outputs”, [“”])[-1]: return “think” # 返回下一个节点的名称 else: return “finish” workflow.add_conditional_edges( “act”, should_continue, # 路由函数 { “think”: “think”, “finish”: “finish” } ) # 6. 添加最终边 workflow.add_edge(“finish”, END) # 7. 编译图得到可执行对象 app workflow.compile()现在一个包含基本循环逻辑的Agent工作流就定义好了。执行app.invoke({“input”: “什么是LangGraph”})即可运行。4. 完整实战案例构建具备记忆的研究助手Agent让我们构建一个更真实的Agent它能够根据复杂问题执行多轮网络搜索并整合信息形成最终报告。4.1 项目结构与设计research_agent/ ├── .env # 存储API密钥 ├── requirements.txt # 项目依赖 ├── config.py # 配置类 ├── agents/ │ ├── __init__.py │ └── research_agent.py # Agent图定义 ├── tools/ │ ├── __init__.py │ └── search_tools.py # 自定义搜索工具 └── main.py # 主程序入口4.2 定义工具与模型首先在tools/search_tools.py中我们定义一个模拟的搜索工具。在实际项目中你可以替换为SerpAPI、Tavily Search API或任何你需要的工具。# tools/search_tools.py from langchain.tools import tool import random tool def web_search(query: str) - str: 执行一次网络搜索。传入一个搜索查询字符串返回搜索结果摘要。 这是一个模拟工具实际应用中请接入真正的搜索API。 # 模拟不同查询返回不同结果 mock_responses [ f“关于‘{query}’的搜索结果1LangGraph是LangChain的一个扩展用于构建有状态、多参与者的应用。”, f“关于‘{query}’的搜索结果2它使用图结构来定义工作流节点代表操作边代表控制流。”, f“关于‘{query}’的搜索结果3主要特性包括循环、分支、持久化状态和人性化交互。”, “未找到更具体的信息。” ] return random.choice(mock_responses[:3]) # 75%概率返回有效信息接下来在config.py中配置大语言模型。# config.py import os from langchain_openai import ChatOpenAI from dotenv import load_dotenv load_dotenv() def get_llm(model_name: str “gpt-3.5-turbo”, temperature: float 0.1): 获取配置好的LLM实例。 temperature调低使输出更确定适合工作流。 api_key os.getenv(“OPENAI_API_KEY”) if not api_key: raise ValueError(“请在.env文件中设置OPENAI_API_KEY”) return ChatOpenAI(modelmodel_name, temperaturetemperature, api_keyapi_key)4.3 定义Agent状态与图这是核心部分在agents/research_agent.py中实现。# agents/research_agent.py from typing import TypedDict, List, Annotated, Literal import operator from langgraph.graph import StateGraph, END, START from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from config import get_llm from tools.search_tools import web_search # 1. 定义状态 class ResearchState(TypedDict): # 消息历史使用LangGraph提供的特殊注解实现累积 messages: Annotated[List, add_messages] # 已执行的搜索次数用于限制循环 search_count: int # 收集到的关键信息点 gathered_facts: Annotated[List[str], operator.add] # 最终报告 report: str # 2. 定义节点函数 llm get_llm() tools [web_search] llm_with_tools llm.bind_tools(tools) def planner_node(state: ResearchState): 规划节点分析问题决定是否需要搜索以及搜索关键词。 system_prompt “““你是一个研究规划员。请分析用户的问题判断是否需要通过网络搜索来获取信息。 如果需要请生成一个最合适的搜索查询词。 如果问题简单或基于常识请直接回答‘无需搜索’。 只输出搜索词或‘无需搜索’。””” messages [SystemMessage(contentsystem_prompt)] state[“messages”] response llm.invoke(messages) plan response.content.strip() return {“plan”: plan} # 注意这里需要更新状态结构我们先简化后面会整合 def search_node(state: ResearchState): 执行搜索节点。 # 从状态或上一步的规划中获取查询词这里简化处理 last_msg state[“messages”][-1] if isinstance(last_msg, HumanMessage): query last_msg.content[:50] # 简单截取用户问题前50字作为查询 else: query “LangGraph tutorial” result web_search.invoke({“query”: query}) return { “gathered_facts”: [f“搜索词‘{query}’的结果{result}”], “search_count”: state.get(“search_count”, 0) 1 } def synthesizer_node(state: ResearchState): 综合节点根据收集到的信息生成初步答案或报告。 facts “\n”.join(state[“gathered_facts”]) user_question state[“messages”][-1].content if state[“messages”] else “” prompt f“““基于以下收集到的信息请回答用户的问题。 用户问题{user_question} 收集到的信息 {facts} 请生成一个清晰、有条理的回答。如果信息不足请明确指出。””” response llm.invoke([HumanMessage(contentprompt)]) return {“report”: response.content} def router_node(state: ResearchState) - Literal[“search”, “synthesize”, “finish”]: 路由节点决定下一步是搜索、综合还是结束。 search_count state.get(“search_count”, 0) # 规则1如果还没搜索过先去搜索 if search_count 0: return “search” # 规则2如果搜索次数少于3次且最新报告里包含“信息不足”则继续搜索 elif search_count 3 and “信息不足” in state.get(“report”, “”): return “search” # 规则3否则进入综合生成最终报告 else: return “synthesize” # 3. 构建图 def create_research_agent(): workflow StateGraph(ResearchState) # 添加节点这里将规划逻辑整合到路由中简化演示 workflow.add_node(“search”, search_node) workflow.add_node(“synthesize”, synthesizer_node) # 设置入口点 workflow.set_entry_point(“search”) # 添加条件边 workflow.add_conditional_edges( “search”, router_node, { “search”: “search”, “synthesize”: “synthesize” } ) workflow.add_conditional_edges( “synthesize”, # 综合后根据路由决定是继续搜索还是结束 lambda s: “search” if router_node(s) “search” else “finish”, { “search”: “search”, “finish”: END } ) return workflow.compile() # 创建图应用实例 research_agent_app create_research_agent()4.4 运行与验证在main.py中编写主程序来测试我们的研究助手。# main.py from agents.research_agent import research_agent_app from langchain_core.messages import HumanMessage def run_research(question: str): print(f“用户问题{question}”) print(“-” * 50) # 初始化状态 initial_state { “messages”: [HumanMessage(contentquestion)], “search_count”: 0, “gathered_facts”: [], “report”: “” } # 执行图 final_state research_agent_app.invoke(initial_state) print(f“执行了 {final_state[‘search_count’]} 次搜索。”) print(“\n收集到的信息”) for i, fact in enumerate(final_state[‘gathered_facts’], 1): print(f“ {i}. {fact}”) print(“\n 最终报告 ) print(final_state[‘report’]) print(“” * 50) if __name__ “__main__”: # 测试一个复杂问题 complex_question “请详细解释LangGraph的工作原理并说明它与普通LangChain Agent的主要区别。” run_research(complex_question)运行python main.py你会看到Agent自动进行了多轮搜索模拟收集信息并最终生成一份综合报告。控制台输出会展示搜索次数、收集到的每条信息以及最终报告。5. 常见问题与排查思路在开发LangGraph应用时你可能会遇到一些典型问题。问题现象可能原因排查思路与解决方案KeyError或状态字段丢失1. 状态TypedDict定义与节点返回值不匹配。2. 节点返回的字典键不在状态定义中。1. 检查class State(TypedDict)的所有字段名。2. 确保每个节点返回的dict的键都是状态中定义的字段。3. 使用print(state)在节点开始处调试。图陷入无限循环1. 条件边ConditionalEdge的逻辑有误始终返回同一个节点。2. 没有设置终止条件指向END的边。1. 仔细检查路由函数如should_continue的逻辑。2. 确保在某个条件下能路由到END。3. 在状态中添加计数器如search_count并设置上限。langchain或langgraph导入错误/版本冲突1.langchain生态包版本不兼容。2. 安装了冲突的langchain-core、langchain-community等子包版本。1. 使用pip list工具Tools无法被LLM识别或调用1. 没有正确使用bind_tools()将工具绑定到LLM。2. 工具的函数文档字符串docstring不清晰导致模型无法理解。3. 模型不支持工具调用如某些本地模型。1. 确保使用llm.bind_tools(tools)。2. 为工具函数编写清晰、结构化的docstring描述输入参数和返回值。3. 对于OpenAI模型确保使用ChatOpenAI且模型版本支持函数调用如gpt-3.5-turbo-1106及以上。记忆Memory没有持久化每次对话都是新的1. 使用的是内存中的图实例应用重启后状态丢失。2. 没有配置持久化存储后端。1. 对于需要长期记忆的场景使用LangGraph的Pregel持久化API。2. 配置checkpointer例如SqliteSaver将图状态保存到数据库。**错误Agent stopped due to iteration limit or time limit.**1. LangGraph默认设置了迭代次数上限以防止无限循环。2. 你的工作流步骤可能过于复杂。1. 在编译图时通过compile(checkpointer..., interrupt_before[...], interrupt_after[...])进行更精细的控制。2. 检查工作流逻辑优化节点设计减少不必要的循环。与langchain-community等扩展包的工具兼容性问题社区工具可能使用了较新的langchain-core接口与主版本不兼容。1. 尝试将所有langchain相关包升级到最新稳定版。2. 如果问题仍在查看工具对应GitHub仓库的Issue或暂时使用自定义工具。6. 最佳实践与工程建议将LangGraph应用到生产环境需要关注以下几点。6.1 状态设计规范最小化与清晰化状态只存储工作流真正需要共享和传递的数据。避免将临时变量或大型对象塞入状态。使用TypedDict和Annotated这不仅是类型提示更是LangGraph理解如何合并状态更新的声明。对于列表使用Annotated[List, operator.add]对于字符串拼接可以使用operator.add或自定义归约函数。区分对话历史与内部状态像messages这样的对话历史使用add_messages注解。像search_count、draft这样的内部控制变量使用普通字段。6.2 节点设计原则单一职责每个节点只做一件事。例如一个节点负责调用LLM另一个节点负责处理工具结果。幂等性与容错节点函数应尽可能设计成幂等的。考虑网络调用或工具调用失败的情况在节点内进行try-catch并将错误信息作为状态的一部分传递给后续节点处理。充分的日志记录在每个节点的开始和结束处记录状态的关键信息这对于调试复杂工作流至关重要。可以使用Python的logging模块并设置不同的日志级别。6.3 图的健壮性设置安全阀对于可能循环的路径必须在状态中设置计数器如iteration并在路由函数中检查超过阈值则强制结束避免产生意外费用对于付费API或死循环。实现人工审核节点Human-in-the-Loop对于关键决策点如发布内容、执行重要操作可以设计一个节点暂停执行等待外部如人工输入后再继续。这通过interrupt机制实现。# 在编译时声明可中断的节点 app workflow.compile(checkpointercheckpointer, interrupt_before[“publish”])持久化与检查点对于长时间运行的任务务必使用SqliteSaver或RedisSaver等检查点保存器。这允许应用在崩溃或重启后从上次中断的地方继续。from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(“:memory:”) # 或你的数据库路径 app workflow.compile(checkpointermemory)6.4 测试与监控单元测试节点函数将每个节点函数当作纯函数尽可能进行测试模拟输入状态验证输出状态。集成测试整个图使用固定的输入测试端到端的工作流断言最终的输出状态符合预期。可视化你的图LangGraph提供了app.get_graph().draw_mermaid_png()方法需要安装mermaid和pygraphviz生成工作流图。可视化有助于理解复杂流程和向团队解释设计。监控与指标在生产环境中记录每个工作流实例的执行时间、节点调用次数、工具调用成功率、Token消耗等指标便于性能分析和成本控制。7. 总结与进阶学习路线通过本文你已经掌握了LangGraph的核心概念用图来定义工作流用状态来共享数据用节点和边来编排逻辑。我们从一个简单的循环示例构建了一个具备多轮搜索和综合能力的研究助手Agent。关键收获状态State是核心使用TypedDict和Annotated清晰地定义数据流。图Graph是蓝图StateGraph帮你将复杂的Agent逻辑可视化、模块化。条件边实现智能控制让Agent能够根据中间结果动态调整执行路径。工具Tools是能力的延伸通过bind_tools将外部API无缝集成到Agent中。下一步可以探索更复杂的模式研究StateGraph的add_edge、add_conditional_edges、add_node的高级用法实现并行执行、动态子图调用。持久化与并发深入学习SqliteSaver将Agent状态保存到数据库构建支持多用户、长会话的Web服务。与Web框架集成将编译好的app实例集成到FastAPI或Flask中提供HTTP API打造你的AI智能体后端服务。替换模型后端尝试使用langchain-ollama、langchain-anthropic等包将OpenAI模型替换为Claude、本地部署的Llama或Qwen等模型掌握多模型切换的能力。探索LangGraph Studio这是一个官方的图形化调试和监控工具可以直观地查看工作流执行过程对于调试复杂Agent非常有帮助。AI Agent的开发是一个将软件工程思维与大语言模型能力相结合的领域。LangGraph提供的这套范式极大地提升了复杂Agent的可控性、可维护性和可观测性。从今天这个具备记忆的研究助手开始你可以继续扩展它的能力例如加入代码执行、文件读写、数据库查询等工具最终构建出能够自主处理真实世界任务的强大智能体。 LangGraph实战：构建具备长期记忆与复杂推理能力的AI智能体
最近在尝试构建一个能自主处理复杂任务的AI智能体时你是否也遇到过这样的困境Agent的逻辑流难以编排状态管理混乱记忆能力时有时无调试起来更是让人头疼。传统的LangChain Agent虽然强大但在处理多步骤、有状态的复杂工作流时往往显得力不从心。这正是LangGraph要解决的问题。本文将为你系统拆解如何结合LangGraph、Langchain和Agent从零构建一个具备长期记忆和复杂推理能力的AI应用。无论你是想入门AI应用开发还是希望将现有项目升级为更智能的Agent系统这套保姆级教程都能提供从概念到部署的完整路径。1. 背景与核心概念为什么需要LangGraph在深入代码之前我们首先要理解这几个核心组件分别解决了什么问题以及它们如何协同工作。1.1 AI Agent与LangChain智能体的基石AI Agent智能体不仅仅是一个聊天机器人。它是一个能够感知环境、进行决策并执行动作以实现特定目标的软件实体。一个典型的Agent包含几个关键部分工具ToolsAgent可以调用的外部能力如搜索网络、执行计算、读写文件。记忆Memory存储对话历史、中间结果和知识使Agent具备上下文感知能力。规划Planning将复杂目标拆解为可执行的步骤序列。执行Execution调用工具并处理结果。LangChain是一个用于开发由大语言模型驱动的应用程序的框架。它提供了构建Agent所需的基础模块标准化接口统一不同大模型如OpenAI、Anthropic、本地模型的调用方式。丰富的工具链内置了大量实用工具搜索引擎、计算器、API调用等也支持自定义。基础记忆系统提供了对话缓存、向量存储等记忆后端。然而当任务流程包含条件分支、循环或需要维护复杂状态时仅用LangChain原生的AgentExecutor会变得非常笨拙。1.2 LangGraph为智能体引入“工作流引擎”这就是LangGraph的用武之地。你可以把它理解为专门为LangChain Agent设计的“工作流引擎”或“状态机框架”。它基于流行的网络分析库NetworkX构建核心思想是将Agent的执行过程建模为一个有向图Graph。节点Nodes代表一个执行单元可以是一个工具调用、一个条件判断或者一次LLM调用。边Edges定义了节点之间的流转路径。边可以是固定的也可以根据上一个节点的输出结果动态决定条件边。状态State一个贯穿整个图执行过程的共享字典。每个节点都可以读取和修改状态这是实现长期记忆和复杂信息传递的关键。LangGraph与LangChain Agent的核心区别LangChain Agent通常是线性的“思考-行动-观察”循环难以处理非线性的任务流。LangGraph允许你显式地定义包含分支、循环、并行和子流程的复杂工作流对执行过程拥有更精细的控制。1.3 核心价值可预测的状态管理与循环LangGraph带来的最大好处是状态管理的清晰化和循环与分支的原生支持。例如构建一个“研究助手”Agent流程可能是1) 理解问题2) 搜索网络3) 判断信息是否足够分支如果不足循环回步骤2如果足够则4) 撰写报告。用LangGraph可以非常直观地建模这个流程并确保“已收集信息”这个状态在整个循环中得以保持和更新。2. 环境准备与版本说明在开始实战之前我们需要搭建一个稳定的开发环境。本文将使用Python作为开发语言。2.1 基础环境配置建议使用Python 3.10或更高版本以获得最佳的兼容性。我们使用venv创建虚拟环境来隔离项目依赖。# 1. 创建项目目录并进入 mkdir langgraph-agent-tutorial cd langgraph-agent-tutorial # 2. 创建Python虚拟环境 python -m venv venv # 3. 激活虚拟环境 # 在Windows上 venv\Scripts\activate # 在macOS/Linux上 source venv/bin/activate # 4. 升级pip pip install --upgrade pip2.2 依赖安装创建requirements.txt文件并填入以下核心依赖。注意LangGraph和LangChain版本迭代较快以下版本为当前推荐组合。# requirements.txt langchain0.1.0 langchain-community0.0.10 langgraph0.0.26 langchain-openai0.0.5 # 用于OpenAI模型 openai1.6.1 # OpenAI官方SDK tiktoken0.5.2 # 用于Token计数 python-dotenv1.0.0 # 用于管理环境变量安装依赖pip install -r requirements.txt重要提示如果你计划使用本地大模型如通过Ollama部署的Llama 2、Mistral等则需要安装langchain-ollama等对应社区包并注释掉langchain-openai。本文示例将主要使用OpenAI GPT模型进行演示因为其稳定性和可用性较高。2.3 配置API密钥为了调用OpenAI的模型你需要一个API密钥。建议使用.env文件管理敏感信息避免将其硬编码在代码中。在项目根目录创建.env文件。在文件中添加你的OpenAI API密钥OPENAI_API_KEYsk-your-actual-openai-api-key-here在代码中通过python-dotenv加载from dotenv import load_dotenv load_dotenv() # 这会从.env文件加载环境变量到os.environ # 现在可以通过 os.getenv(OPENAI_API_KEY) 获取密钥3. 核心概念拆解图、状态与节点理解LangGraph的三个核心抽象是编写代码的基础。3.1 状态State数据的共享总线在LangGraph中状态是一个类似字典的对象它在图的所有节点之间传递和共享。通常我们使用TypedDict来定义状态的模式这有助于类型提示和代码清晰度。from typing import TypedDict, List, Annotated import operator class AgentState(TypedDict): # 用户的原始问题 input: str # Agent的“思考”过程或中间答案 thoughts: Annotated[List[str], operator.add] # 关键这是一个追加式列表 # 从工具获得的结果 tool_outputs: Annotated[List[str], operator.add] # 最终给用户的回答 final_output: str关键点Annotated[List[str], operator.add]。这个注解是LangGraph的魔法所在。它告诉框架当多个节点修改thoughts字段时不是覆盖而是将它们的输出追加add到同一个列表中。这为实现“记忆”提供了基础。3.2 节点Node执行单元节点是一个普通的Python函数它接收当前State对其进行修改然后返回更新后的State或其中的一部分。def llm_think_node(state: AgentState) - dict: 一个模拟LLM思考的节点 # 从状态中获取信息 question state[“input”] # 这里模拟LLM的思考过程。实际应用中你会在这里调用ChatOpenAI等模型。 new_thought f“我正在思考问题{question}。我需要搜索更多信息。” # 返回要更新到状态中的部分 return {“thoughts”: [new_thought]} def tool_call_node(state: AgentState) - dict: 一个模拟调用工具的节点 # 模拟调用一个网络搜索工具 search_result “根据网络搜索LangGraph是一个用于构建有状态多Actor应用的工作流库。” return {“tool_outputs”: [search_result]}3.3 边Edge与图Graph编排流程边决定了执行完一个节点后下一步该去哪个节点。ConditionalEdge允许根据状态内容动态决定路径。from langgraph.graph import StateGraph, END # 1. 创建图的工作流构建器 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(“think”, llm_think_node) workflow.add_node(“act”, tool_call_node) workflow.add_node(“finish”, final_response_node) # 假设已定义 # 3. 设置入口点 workflow.set_entry_point(“think”) # 4. 添加普通边无条件转移 workflow.add_edge(“think”, “act”) # 5. 添加条件边 from langgraph.graph import ConditionalEdge def should_continue(state: AgentState) - str: # 根据工具结果判断是否需要继续思考 if “信息不足” in state.get(“tool_outputs”, [“”])[-1]: return “think” # 返回下一个节点的名称 else: return “finish” workflow.add_conditional_edges( “act”, should_continue, # 路由函数 { “think”: “think”, “finish”: “finish” } ) # 6. 添加最终边 workflow.add_edge(“finish”, END) # 7. 编译图得到可执行对象 app workflow.compile()现在一个包含基本循环逻辑的Agent工作流就定义好了。执行app.invoke({“input”: “什么是LangGraph”})即可运行。4. 完整实战案例构建具备记忆的研究助手Agent让我们构建一个更真实的Agent它能够根据复杂问题执行多轮网络搜索并整合信息形成最终报告。4.1 项目结构与设计research_agent/ ├── .env # 存储API密钥 ├── requirements.txt # 项目依赖 ├── config.py # 配置类 ├── agents/ │ ├── __init__.py │ └── research_agent.py # Agent图定义 ├── tools/ │ ├── __init__.py │ └── search_tools.py # 自定义搜索工具 └── main.py # 主程序入口4.2 定义工具与模型首先在tools/search_tools.py中我们定义一个模拟的搜索工具。在实际项目中你可以替换为SerpAPI、Tavily Search API或任何你需要的工具。# tools/search_tools.py from langchain.tools import tool import random tool def web_search(query: str) - str: 执行一次网络搜索。传入一个搜索查询字符串返回搜索结果摘要。 这是一个模拟工具实际应用中请接入真正的搜索API。 # 模拟不同查询返回不同结果 mock_responses [ f“关于‘{query}’的搜索结果1LangGraph是LangChain的一个扩展用于构建有状态、多参与者的应用。”, f“关于‘{query}’的搜索结果2它使用图结构来定义工作流节点代表操作边代表控制流。”, f“关于‘{query}’的搜索结果3主要特性包括循环、分支、持久化状态和人性化交互。”, “未找到更具体的信息。” ] return random.choice(mock_responses[:3]) # 75%概率返回有效信息接下来在config.py中配置大语言模型。# config.py import os from langchain_openai import ChatOpenAI from dotenv import load_dotenv load_dotenv() def get_llm(model_name: str “gpt-3.5-turbo”, temperature: float 0.1): 获取配置好的LLM实例。 temperature调低使输出更确定适合工作流。 api_key os.getenv(“OPENAI_API_KEY”) if not api_key: raise ValueError(“请在.env文件中设置OPENAI_API_KEY”) return ChatOpenAI(modelmodel_name, temperaturetemperature, api_keyapi_key)4.3 定义Agent状态与图这是核心部分在agents/research_agent.py中实现。# agents/research_agent.py from typing import TypedDict, List, Annotated, Literal import operator from langgraph.graph import StateGraph, END, START from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from config import get_llm from tools.search_tools import web_search # 1. 定义状态 class ResearchState(TypedDict): # 消息历史使用LangGraph提供的特殊注解实现累积 messages: Annotated[List, add_messages] # 已执行的搜索次数用于限制循环 search_count: int # 收集到的关键信息点 gathered_facts: Annotated[List[str], operator.add] # 最终报告 report: str # 2. 定义节点函数 llm get_llm() tools [web_search] llm_with_tools llm.bind_tools(tools) def planner_node(state: ResearchState): 规划节点分析问题决定是否需要搜索以及搜索关键词。 system_prompt “““你是一个研究规划员。请分析用户的问题判断是否需要通过网络搜索来获取信息。 如果需要请生成一个最合适的搜索查询词。 如果问题简单或基于常识请直接回答‘无需搜索’。 只输出搜索词或‘无需搜索’。””” messages [SystemMessage(contentsystem_prompt)] state[“messages”] response llm.invoke(messages) plan response.content.strip() return {“plan”: plan} # 注意这里需要更新状态结构我们先简化后面会整合 def search_node(state: ResearchState): 执行搜索节点。 # 从状态或上一步的规划中获取查询词这里简化处理 last_msg state[“messages”][-1] if isinstance(last_msg, HumanMessage): query last_msg.content[:50] # 简单截取用户问题前50字作为查询 else: query “LangGraph tutorial” result web_search.invoke({“query”: query}) return { “gathered_facts”: [f“搜索词‘{query}’的结果{result}”], “search_count”: state.get(“search_count”, 0) 1 } def synthesizer_node(state: ResearchState): 综合节点根据收集到的信息生成初步答案或报告。 facts “\n”.join(state[“gathered_facts”]) user_question state[“messages”][-1].content if state[“messages”] else “” prompt f“““基于以下收集到的信息请回答用户的问题。 用户问题{user_question} 收集到的信息 {facts} 请生成一个清晰、有条理的回答。如果信息不足请明确指出。””” response llm.invoke([HumanMessage(contentprompt)]) return {“report”: response.content} def router_node(state: ResearchState) - Literal[“search”, “synthesize”, “finish”]: 路由节点决定下一步是搜索、综合还是结束。 search_count state.get(“search_count”, 0) # 规则1如果还没搜索过先去搜索 if search_count 0: return “search” # 规则2如果搜索次数少于3次且最新报告里包含“信息不足”则继续搜索 elif search_count 3 and “信息不足” in state.get(“report”, “”): return “search” # 规则3否则进入综合生成最终报告 else: return “synthesize” # 3. 构建图 def create_research_agent(): workflow StateGraph(ResearchState) # 添加节点这里将规划逻辑整合到路由中简化演示 workflow.add_node(“search”, search_node) workflow.add_node(“synthesize”, synthesizer_node) # 设置入口点 workflow.set_entry_point(“search”) # 添加条件边 workflow.add_conditional_edges( “search”, router_node, { “search”: “search”, “synthesize”: “synthesize” } ) workflow.add_conditional_edges( “synthesize”, # 综合后根据路由决定是继续搜索还是结束 lambda s: “search” if router_node(s) “search” else “finish”, { “search”: “search”, “finish”: END } ) return workflow.compile() # 创建图应用实例 research_agent_app create_research_agent()4.4 运行与验证在main.py中编写主程序来测试我们的研究助手。# main.py from agents.research_agent import research_agent_app from langchain_core.messages import HumanMessage def run_research(question: str): print(f“用户问题{question}”) print(“-” * 50) # 初始化状态 initial_state { “messages”: [HumanMessage(contentquestion)], “search_count”: 0, “gathered_facts”: [], “report”: “” } # 执行图 final_state research_agent_app.invoke(initial_state) print(f“执行了 {final_state[‘search_count’]} 次搜索。”) print(“\n收集到的信息”) for i, fact in enumerate(final_state[‘gathered_facts’], 1): print(f“ {i}. {fact}”) print(“\n 最终报告 ) print(final_state[‘report’]) print(“” * 50) if __name__ “__main__”: # 测试一个复杂问题 complex_question “请详细解释LangGraph的工作原理并说明它与普通LangChain Agent的主要区别。” run_research(complex_question)运行python main.py你会看到Agent自动进行了多轮搜索模拟收集信息并最终生成一份综合报告。控制台输出会展示搜索次数、收集到的每条信息以及最终报告。5. 常见问题与排查思路在开发LangGraph应用时你可能会遇到一些典型问题。问题现象可能原因排查思路与解决方案KeyError或状态字段丢失1. 状态TypedDict定义与节点返回值不匹配。2. 节点返回的字典键不在状态定义中。1. 检查class State(TypedDict)的所有字段名。2. 确保每个节点返回的dict的键都是状态中定义的字段。3. 使用print(state)在节点开始处调试。图陷入无限循环1. 条件边ConditionalEdge的逻辑有误始终返回同一个节点。2. 没有设置终止条件指向END的边。1. 仔细检查路由函数如should_continue的逻辑。2. 确保在某个条件下能路由到END。3. 在状态中添加计数器如search_count并设置上限。langchain或langgraph导入错误/版本冲突1.langchain生态包版本不兼容。2. 安装了冲突的langchain-core、langchain-community等子包版本。1. 使用pip list工具Tools无法被LLM识别或调用1. 没有正确使用bind_tools()将工具绑定到LLM。2. 工具的函数文档字符串docstring不清晰导致模型无法理解。3. 模型不支持工具调用如某些本地模型。1. 确保使用llm.bind_tools(tools)。2. 为工具函数编写清晰、结构化的docstring描述输入参数和返回值。3. 对于OpenAI模型确保使用ChatOpenAI且模型版本支持函数调用如gpt-3.5-turbo-1106及以上。记忆Memory没有持久化每次对话都是新的1. 使用的是内存中的图实例应用重启后状态丢失。2. 没有配置持久化存储后端。1. 对于需要长期记忆的场景使用LangGraph的Pregel持久化API。2. 配置checkpointer例如SqliteSaver将图状态保存到数据库。**错误Agent stopped due to iteration limit or time limit.**1. LangGraph默认设置了迭代次数上限以防止无限循环。2. 你的工作流步骤可能过于复杂。1. 在编译图时通过compile(checkpointer..., interrupt_before[...], interrupt_after[...])进行更精细的控制。2. 检查工作流逻辑优化节点设计减少不必要的循环。与langchain-community等扩展包的工具兼容性问题社区工具可能使用了较新的langchain-core接口与主版本不兼容。1. 尝试将所有langchain相关包升级到最新稳定版。2. 如果问题仍在查看工具对应GitHub仓库的Issue或暂时使用自定义工具。6. 最佳实践与工程建议将LangGraph应用到生产环境需要关注以下几点。6.1 状态设计规范最小化与清晰化状态只存储工作流真正需要共享和传递的数据。避免将临时变量或大型对象塞入状态。使用TypedDict和Annotated这不仅是类型提示更是LangGraph理解如何合并状态更新的声明。对于列表使用Annotated[List, operator.add]对于字符串拼接可以使用operator.add或自定义归约函数。区分对话历史与内部状态像messages这样的对话历史使用add_messages注解。像search_count、draft这样的内部控制变量使用普通字段。6.2 节点设计原则单一职责每个节点只做一件事。例如一个节点负责调用LLM另一个节点负责处理工具结果。幂等性与容错节点函数应尽可能设计成幂等的。考虑网络调用或工具调用失败的情况在节点内进行try-catch并将错误信息作为状态的一部分传递给后续节点处理。充分的日志记录在每个节点的开始和结束处记录状态的关键信息这对于调试复杂工作流至关重要。可以使用Python的logging模块并设置不同的日志级别。6.3 图的健壮性设置安全阀对于可能循环的路径必须在状态中设置计数器如iteration并在路由函数中检查超过阈值则强制结束避免产生意外费用对于付费API或死循环。实现人工审核节点Human-in-the-Loop对于关键决策点如发布内容、执行重要操作可以设计一个节点暂停执行等待外部如人工输入后再继续。这通过interrupt机制实现。# 在编译时声明可中断的节点 app workflow.compile(checkpointercheckpointer, interrupt_before[“publish”])持久化与检查点对于长时间运行的任务务必使用SqliteSaver或RedisSaver等检查点保存器。这允许应用在崩溃或重启后从上次中断的地方继续。from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(“:memory:”) # 或你的数据库路径 app workflow.compile(checkpointermemory)6.4 测试与监控单元测试节点函数将每个节点函数当作纯函数尽可能进行测试模拟输入状态验证输出状态。集成测试整个图使用固定的输入测试端到端的工作流断言最终的输出状态符合预期。可视化你的图LangGraph提供了app.get_graph().draw_mermaid_png()方法需要安装mermaid和pygraphviz生成工作流图。可视化有助于理解复杂流程和向团队解释设计。监控与指标在生产环境中记录每个工作流实例的执行时间、节点调用次数、工具调用成功率、Token消耗等指标便于性能分析和成本控制。7. 总结与进阶学习路线通过本文你已经掌握了LangGraph的核心概念用图来定义工作流用状态来共享数据用节点和边来编排逻辑。我们从一个简单的循环示例构建了一个具备多轮搜索和综合能力的研究助手Agent。关键收获状态State是核心使用TypedDict和Annotated清晰地定义数据流。图Graph是蓝图StateGraph帮你将复杂的Agent逻辑可视化、模块化。条件边实现智能控制让Agent能够根据中间结果动态调整执行路径。工具Tools是能力的延伸通过bind_tools将外部API无缝集成到Agent中。下一步可以探索更复杂的模式研究StateGraph的add_edge、add_conditional_edges、add_node的高级用法实现并行执行、动态子图调用。持久化与并发深入学习SqliteSaver将Agent状态保存到数据库构建支持多用户、长会话的Web服务。与Web框架集成将编译好的app实例集成到FastAPI或Flask中提供HTTP API打造你的AI智能体后端服务。替换模型后端尝试使用langchain-ollama、langchain-anthropic等包将OpenAI模型替换为Claude、本地部署的Llama或Qwen等模型掌握多模型切换的能力。探索LangGraph Studio这是一个官方的图形化调试和监控工具可以直观地查看工作流执行过程对于调试复杂Agent非常有帮助。AI Agent的开发是一个将软件工程思维与大语言模型能力相结合的领域。LangGraph提供的这套范式极大地提升了复杂Agent的可控性、可维护性和可观测性。从今天这个具备记忆的研究助手开始你可以继续扩展它的能力例如加入代码执行、文件读写、数据库查询等工具最终构建出能够自主处理真实世界任务的强大智能体。