在本篇文章中，您将学习如何使用 LangGraph 在 Python 中构建完整的代理工作流，从单个模型调用到具备持久对话记忆的工具使用代理。
我们将涵盖以下主题：
- 状态、节点和边如何组合定义 LangGraph 代理的执行流程。
- 如何注册工具并通过图的推理循环路由模型的工具调用。
- 检查点如何在单独的图调用之间持久保存对话历史。
我们不再浪费时间了。
引言
大多数 AI 代理设置都能很好地处理单轮情况：接收问题、调用模型并返回答案。之后更难的问题很快就会出现。代理可能需要查询您的数据库、记住之前消息的上下文，或让您清楚地看到模型做了什么决定以及原因。在不为每个用例构建自定义管道的情况下解决这些挑战，是许多实现开始崩溃的地方。
LangGraph 为处理这些问题提供了清晰的结构。代理被表示为一个图，其中节点是工作单元，边定义下一步运行什么，共享状态对象在每一步都携带完整的消息历史。模型在节点内运行，因此每个推理步骤、工具调用和响应都成为图状态的一部分。这使得整个执行流程可见、可检查，并可供之后运行的任何节点使用。
在本文中，您将学习如何理解每个 LangGraph 图构建所基于的状态、节点和边原语；使用 MessagesState 自动管理对话历史；在节点内调用语言模型并将其连接到图；注册工具并将工具调用路由回模型；跟踪完整的消息序列以查看模型在每一步的确切操作；以及使用检查点在单独调用之间持久保存对话。我们将从头开始构建图，从安装步骤开始。
设置环境
安装所需包：
pip install langgraph langchain-openai python-dotenv |
然后在项目根目录创建 .env 文件，写入您的 OpenAI API 密钥：
OPENAI_API_KEY="your_key_here" |
在脚本顶部加载它，优先于任何 LangChain 或 LangGraph 导入：
from dotenv import load_dotenv load_dotenv() |
python-dotenv 读取 .env 文件并将密钥设置为环境变量。
理解状态、节点和边
每个 LangGraph 图都由以下三个组件构建。提前正确理解它们可以在图变得更复杂时避免混淆。
状态是一个 TypedDict，充当整个图的共享内存。每个节点从它读取并将更新写回。除了这种方式，没有其他数据在节点之间传递。您未在节点中更新的字段保持不变；您只返回希望修改的内容。
节点是普通的 Python 函数。节点将当前状态作为参数并返回一个字典，包含它希望更新的字段。使用 add_node 注册函数即可使其成为图的一部分，无需特殊装饰器或基类。如果您只传递函数而不传递名称字符串，LangGraph 会自动使用函数名。
边定义执行顺序。add_edge(A, B) 表示：节点 A 完成后，运行节点 B。add_conditional_edges 表示：节点 A 完成后，调用路由函数并前往其指向的位置。每个图都需要 START 作为入口点，并至少有一条通往 END 的路径。
默认情况下，当节点返回状态字段的值时，该值会替换原有内容。对于应在节点间累积的字段（如日志、消息历史），您需要用 reducer 函数 标注该字段。在以下示例中，列表字段上的 operator.add 表示追加而非替换：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 |
from typing import Annotated import operator from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END class TicketState(TypedDict): customer_message: str log: Annotated[list, operator.add] def log_received(state: TicketState) -> dict: return {"log": [f"Received: {state['customer_message']}"]} def log_assigned(state: TicketState) -> dict: return {"log": ["Assigned to support queue"]} builder = StateGraph(TicketState) builder.add_node("log_received", log_received) builder.add_node("log_assigned", log_assigned) builder.add_edge(START, "log_received") builder.add_edge("log_received", "log_assigned") builder.add_edge("log_assigned", END) graph = builder.compile() result = graph.invoke({"customer_message": "My invoice looks wrong", "log": []}) print(result) |
输出结果：
{'customer_message': 'My invoice looks wrong', 'log': ['Received: My invoice looks wrong', 'Assigned to support queue']} |
两个节点都写入了日志，且两条记录都存在。customer_message 未被修改，因为两个节点都没有返回它。这正是 MessagesState 使用略微特殊的 reducer add_messages（同时处理消息对象的去重和排序）来处理其 messages 字段的方式。
使用 MessagesState 管理对话历史
LangGraph 图中的每个节点都会读取当前状态并将更新写回。对于对话代理，状态需要携带完整的消息历史——用户输入、模型响应、工具输出——以便模型在决定下一步操作时始终拥有所需上下文。
LangGraph 提供了一个专门为此设计的内置状态类型：MessagesState。它是一个带有单一 messages 字段的 TypedDict，使用 add_messages reducer 而非简单覆盖。每次节点返回新消息时，这些消息都会追加到现有列表中，而不是替换它。您无需手动拼接对话历史。
from langgraph.graph import MessagesState |
这是大多数单代理图所需的状态定义。您可以扩展它以添加其他字段，例如 customer_id、priority 标志等节点所需的内容。但 messages 字段已存在且已配置为自动累积。
在节点内调用模型
状态就绪后，任何 LangGraph 代理的核心节点都是一个将当前消息列表传递给模型并追加其响应的函数。模型返回 AIMessage；将其放入以“messages”为键的字典中返回，即可将其添加到状态。
from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage llm = ChatOpenAI(model="gpt-4o-mini") def run_model(state: MessagesState) -> dict: system = SystemMessage("You are a support agent for a SaaS product. " "Be concise and helpful.") response = llm.invoke([system] + state["messages"]) return {"messages": [response]} |
ChatOpenAI 使用 LangChain 的标准聊天模型接口封装 OpenAI API。切换到其他提供商（如 Anthropic、Google 或通过 Ollama 使用的本地模型）只需更改导入和模型字符串；节点其余部分保持不变。SystemMessage 在每次调用时设置模型角色，而不存储在状态中，从而保持持久历史干净。
将其连接到图并运行：
from langgraph.graph import StateGraph, START, END from langchain_core.messages import HumanMessage builder = StateGraph(MessagesState) builder.add_node("run_model", run_model) builder.add_edge(START, "run_model") builder.add_edge("run_model", END) graph = builder.compile() result = graph.invoke({"messages": [HumanMessage("My dashboard isn't loading. What should I try?")]}) print(result["messages"][-1].content) |
result["messages"] 是完整列表：原始 HumanMessage 加上模型生成的 AIMessage。[-1] 获取最新的一条。
注册工具并路由工具调用
模型可以从其训练数据中回答一般问题，但任何与您的数据相关的特定内容——账户详情、订阅级别、工单历史——都需要 工具调用。模型决定何时需要工具；您的代码定义工具的功能。
使用 @tool 装饰器定义工具：
from langchain_core.tools import tool @tool def get_customer_tier(customer_id: str) -> str: """Look up the subscription tier for a customer by their ID. Returns 'free', 'pro', or 'enterprise'.""" tiers = { "cust_1001": "enterprise", "cust_2002": "pro", "cust_3003": "free", } return tiers.get(customer_id, "not found") |
文档字符串是模型在决定是否调用此工具以及传递什么参数时读取的内容。保持精确很重要，因为模糊的文档字符串会导致调用遗漏或参数格式错误。
将工具绑定到模型，使其知道工具存在，并更新节点：
tools = [get_customer_tier] llm_with_tools = llm.bind_tools(tools) def run_model(state: MessagesState) -> dict: system = SystemMessage("You are a support agent for a SaaS product. " "Use available tools when you need account-specific information.") response = llm_with_tools.invoke([system] + state["messages"]) return {"messages": [response]} |
bind_tools 将工具的 schema 与每个请求一起发送给模型。当模型决定使用工具时，响应会以带有 tool_calls 字段的 AIMessage 返回，而不是纯文本内容。
添加 ToolNode 来处理执行并连接路由：
from langgraph.prebuilt import ToolNode, tools_condition tool_node = ToolNode(tools) builder = StateGraph(MessagesState) builder.add_node("run_model", run_model) builder.add_node("tools", tool_node) builder.add_edge(START, "run_model") builder.add_conditional_edges("run_model", tools_condition) builder.add_edge("tools", "run_model") graph = builder.compile() |
ToolNode 从最后一个 AIMessage 读取 tool_calls，使用模型指定的参数运行匹配的函数，并将结果包装为 ToolMessage 追加到状态。tools_condition 在每次模型调用后检查最后一个 AIMessage。如果 tool_calls 非空，则路由到“tools”，否则路由到“__end__”。从“tools”回到“run_model”的边是闭合循环的关键：它将工具结果发送回模型，以便模型生成最终答案。
追踪推理循环
在继续之前，请考虑当模型使用工具时图内实际发生的情况，因为实际发生的事情比最终输出所显示的更多。
result = graph.invoke({"messages": [ HumanMessage("Can you check what plan customer cust_1001 is on?") ]}) for msg in result["messages"]: print(type(msg).__name__, ":", msg.content or msg.tool_calls) |
示例输出：
HumanMessage : Can you check what plan customer cust_1001 is on? AIMessage : [{'name': 'get_customer_tier', 'args': {'customer_id': 'cust_1001'}, 'id': 'call_Rx7kLmNpQ2wJtA3s', 'type': 'tool_call'}] ToolMessage : enterprise AIMessage : Customer cust_1001 is on the enterprise plan. |
这里我们有四条消息和两次模型调用。第一次模型调用产生一个带有 tool_calls 字段且 content 为空的 AIMessage。模型正在发出其意图的信号，而不是立即回答。tools_condition 检测到这一点，路由到 ToolNode，后者执行 get_customer_tier("cust_1001") 并追加包含结果的 ToolMessage。
回到 run_model 的边再次触发。现在模型在上下文中拥有之前的三条消息，理解查找成功，并写入最终的 AIMessage（答案在 content 中）。tools_condition 再次运行，未发现工具调用，结束图。
这个循环——模型调用、工具执行、再次模型调用——是标准的 ReAct 模式。每次工具使用需要两次模型调用：一次决定要查找什么，一次解释结果。在考虑添加更多工具时的延迟和成本时，这是一个有用的信息。
在调用之间持久保存对话
上面的每次 graph.invoke() 都从全新的图状态开始。如果没有 持久化，模型不会记住之前的交互。
要持久化调用之间的状态，请在编译图时附加检查点：
from langgraph.checkpoint.memory import InMemorySaver checkpointer = InMemorySaver() graph = builder.compile(checkpointer=checkpointer) |
然后在每次调用时传递相同的 thread_id：
config = {"configurable": {"thread_id": "ticket-7741"}} graph.invoke( {"messages": [HumanMessage("Hi, I can't access my account.")]}, config, ) result = graph.invoke( {"messages": [HumanMessage("My ID is cust_2002, can you check my plan?")]}, config, ) print(result["messages"][-1].content) |
示例输出：
You're on the pro plan, cust_2002. Since you're having trouble accessing your account, I'd recommend resetting your password first. Pro accounts also have priority support available if the issue continues. |
第二次调用可以看到第一次的对话，因为检查点在执行前恢复了该线程的状态，并在执行后保存了更新后的状态。使用不同的 thread_id 会从一个单独的空状态开始。
InMemorySaver 将检查点存储在进程内存中，适合开发和测试。在生产环境中，您通常会将其替换为由数据库或其他持久存储支持的持久检查点。您的其余图代码保持不变。
Checkpointers 为线程持久化 图状态。如果您的应用还需要独立于任何对话持久化数据，例如用户配置文件、偏好或跨多个线程共享的长期记忆，请使用 Store。Stores 通过提供图在执行期间可以访问的持久应用级存储来补充检查点。
总结
在本文中，您从头开始构建了一个完整的 LangGraph 代理。在此过程中，您学习了状态如何在图中流动、节点如何执行工作、工具如何融入执行循环，以及检查点如何在单独调用之间保留对话。这些相同的构建块可以从简单的聊天机器人扩展到更复杂的代理工作流。
LangGraph 的优势之一是每个部分都是独立的。您可以交换语言模型、注册新工具或更改对话持久化方式，而无需重新设计图的其余部分。所有内容都通过共享状态通信，这使图保持可预测且易于扩展。
同样的思想也适用于 多代理系统。将请求路由到专业代理的协调器仍然是一个具有状态、节点和条件边的图。架构会变大，但底层原语保持不变。
如果您想进一步探索，以下资源是很好的起点：
祝您构建愉快！
0 Comments
Log in to join the conversation.No comments yet. Be the first to share your thoughts.