Building Agentic Workflows in Python with LangGraph

使用 LangGraph 在 Python 中构建智能代理工作流

By

Bala Priya C

on

2026 年 7 月 20 日

in

人工智能

0

分享

文章

在本文中，你将学习如何使用 LangGraph 从单次模型调用到具有持久对话记忆的工具使用代理，构建完整的智能代理工作流。

我们将涵盖的主题包括：

• 状态、节点和边如何组合以定义 LangGraph 代理的执行流程。
• 如何注册工具并通过图的推理循环路由模型的工具调用。
• 检查指针如何在多次图调用之间持久化对话历史。

我们不再浪费时间。

介绍

大多数 AI 代理设置都能很好地处理单轮对话场景：接收问题，调用模型，返回答案。但随后会出现更复杂的挑战。代理可能需要查询你的数据库、记住早期消息的上下文，或向你展示模型做出决策的具体过程和原因。在不为每个使用场景构建自定义管道的情况下解决这些挑战，是许多实现开始出现瓶颈的地方。

LangGraph 为处理这些问题提供了清晰的结构。代理被表示为图，其中节点是工作单元，边定义下一步执行内容，共享状态对象则携带完整的消息历史贯穿每个步骤。模型在节点内运行，因此每个推理步骤、工具调用和响应都成为图状态的一部分。这使得整个执行流程可见、可检查，并可供后续运行的任何节点使用。

在本文中，你将学习如何理解构成每个 LangGraph 图的基础状态、节点和边原语；使用 MessagesState 自动管理对话历史；在节点内调用语言模型并将其连接到图；注册工具并将工具调用路由回模型；追踪完整消息序列以查看模型在每个步骤的具体操作；以及使用检查指针在多次调用之间持久化对话。我们将从头开始构建这个图，从安装步骤开始。

环境搭建

安装所需包：

pip install langgraph langchain-openai python-dotenv

1

pip

install

langgraph

langchain

-

openai

python

dotenv

然后在项目根目录创建 .env 文件并填入你的 OpenAI API 密钥：

OPENAI_API_KEY="your_key_here"

OPENAI_API_KEY

=

"your_key_here"

在导入任何 LangChain 或 LangGraph 模块之前，在脚本顶部加载该文件：

from dotenv import load_dotenv load_dotenv()

2

from

import

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 表示追加而非替换：

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)

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

typing

Annotated

operator

typing_extensions

TypedDict

.

graph

StateGraph

,

START

END

class

TicketState

:

customer_message

str

log

[

list

add

]

def

log_received

state

->

dict

return

{

"log"

f

"Received: {state['customer_message']}"

}

log_assigned

"Assigned to support queue"

builder

add_node

"log_received"

"log_assigned"

add_edge

compile

result

invoke

"customer_message"

"My invoice looks wrong"

print

输出结果为：

{'customer_message': 'My invoice looks wrong', 'log': ['Received: My invoice looks wrong', 'Assigned to support queue']}

'customer_message'

'My invoice looks wrong'

'log'

'Received: My invoice looks wrong'

'Assigned to support queue'

两个节点都向 log 字段写入内容，且两条记录都保留了下来。customer_message 字段未被修改，因为没有节点返回该字段。这正是 MessagesState 处理消息字段的方式，它使用一个更专业的 reducer（add_messages）来处理消息对象的去重和排序。

使用 MessagesState 管理对话历史

LangGraph 图中的每个节点都会读取当前状态，并将更新写回状态。对于对话代理，状态需要携带完整的对话历史（用户输入、模型响应、工具输出），以便模型在决定下一步操作时始终拥有必要的上下文。

LangGraph 提供了一种内置的状态类型来处理这种情况：MessagesState。它是一个 TypedDict，包含一个使用 add_messages 还原器的 messages 字段，而不是简单的覆盖。每次节点返回新消息时，它们会追加到现有列表中，而不是替换它。你不需要手动拼接对话历史。

from langgraph.graph import MessagesState

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]}

langchain_openai

ChatOpenAI

langchain_core

messages

SystemMessage

llm

model

"gpt-4o-mini"

run_model

system

"You are a support agent for a SaaS product. "

"Be concise and helpful."

response

+

"messages"

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)

HumanMessage

"run_model"

"My dashboard isn't loading. What should I try?"

content

result["messages"] 是完整列表：原始 HumanMessage 加上模型生成的 AIMessage。[-1] 获取最新的消息。

注册工具并路由工具调用

模型可以回答来自其训练数据的一般性问题，但任何特定于你数据的问题（如账户详情、订阅层级、工单历史）都需要工具调用。模型决定何时需要工具；你的代码定义工具的作用。

使用 @tool 装饰器定义工具：

from langchain_core.tools import tool @tool def get_customer_tier(customer_id: str) -> str: """通过客户 ID 查找客户的订阅层级。返回 'free'、'pro' 或 'enterprise'。""" tiers = { "cust_1001": "enterprise", "cust_2002": "pro", "cust_3003": "free", } return tiers.get(customer_id, "not found")

tools

tool

@

get_customer_tier

customer_id

""

"通过客户 ID 查找客户的订阅层级。

返回 'free'、'pro' 或 'enterprise'。"

tiers

"cust_1001"

"enterprise"

"cust_2002"

"pro"

"cust_3003"

"free"

get

"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]}

llm_with_tools bind_tools "Use available tools when you need account-specific information."

bind_tools 会将工具的模式与每个请求一起发送给模型。当模型决定使用工具时，响应会以包含 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()

prebuilt ToolNode tools_condition tool_node "tools" add_conditional_edges

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)

"Can you check what plan customer cust_1001 is on?" for msg type __name__ ":" or 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.

Can you check what plan customer cust_1001 is ? AIMessage 'name' 'get_customer_tier' 'args' 'customer_id' 'cust_1001' 'id' 'call_Rx7kLmNpQ2wJtA3s' 'type' 'tool_call' ToolMessage enterprise the

这里我们有四条消息和两次模型调用。第一次模型调用生成了一个包含 tool_calls 且内容为空的 AIMessage。模型正在表明它想要执行的操作，而不是直接回答问题。tools_condition 检测到这一情况后，将流程路由到 ToolNode，ToolNode 会执行 get_customer_tier("cust_1001") 并追加一个包含结果的 ToolMessage。

边缘再次触发run_model。此时模型已将前三条历史消息纳入上下文，理解了查询操作已成功，并生成包含答案的最终AIMessage。tools_condition再次运行，未发现工具调用，流程结束。

这种模型调用、工具执行、模型再次调用的循环是标准的ReAct模式。每次工具使用都需要两次模型调用：一次决定要查询的内容，一次解释查询结果。在添加更多工具时，了解这一点对评估延迟和成本非常重要。

跨调用持久化对话

上述每个graph.invoke()都从全新的图状态开始。没有持久化机制时，模型不会记住之前的对话。

要实现调用间的持久化，编译图时需附加检查点记录器：

from langgraph.checkpoint.memory import InMemorySaver checkpointer = InMemorySaver() graph = builder.compile(checkpointer=checkpointer)

checkpoint

memory

InMemorySaver

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)

config

"configurable"

"thread_id"

"ticket-7741"

"Hi, I can't access my account."

"My ID is cust_2002, can you check my plan?"

你正在使用专业版，cust_2002。由于你无法访问账户，我建议你先重置密码。如果问题持续，专业版账户还可享受优先技术支持。

're on the pro plan, cust_2002. Since you'

re

having

trouble

accessing

your

account

I

'

d

recommend

resetting

password

first

Pro

accounts

also

have

priority

support

available

if

issue

continues

第二次调用能看见第一次对话内容，是因为检查点记录器在执行前恢复了线程状态，并在执行后保存了更新后的状态。使用不同的thread_id会从独立的空白状态开始。

InMemorySaver在进程内存中存储检查点，适合开发和测试。生产环境通常会替换为基于数据库或其他持久化存储的检查点记录器。其余图代码保持不变。

检查点记录器为线程持久化图状态。如果应用还需要独立于对话的持久化数据（如用户档案、偏好设置或跨线程的长期记忆），请使用Store。Store通过提供持久化的应用级存储，与检查点记录器形成互补，使图在执行时可以访问这些数据。

总结

在本文中，你从零开始构建了一个完整的LangGraph代理。过程中你学习了状态如何在图中流动、节点如何执行任务、工具如何融入执行循环，以及检查点记录器如何在不同调用间保持对话。这些基础构建模块可扩展至从简单聊天机器人到更复杂的代理工作流。

LangGraph的优势之一是各组件相互独立。你可以更换语言模型、注册新工具或更改对话持久化方式，而无需重新设计整个图。所有组件通过共享状态进行通信，这使图保持可预测性和易于扩展性。

这些理念同样适用于多智能体系统。将请求路由到专业智能体的协调器仍然是一个包含状态、节点和条件边的图结构。架构规模会变大，但底层的基本元素保持不变。

如果您希望进一步探索，以下资源是继续学习的好去处：

• LangGraph 持久化文档
• LangChain 中的工具调用
• MessagesState 与 add_messages
• LangGraph 预置组件

愉快地构建吧！

更多相关内容

• 使用 LangGraph 构建 ReAct 智能体：初学者指南
• 通过管道自动化机器学习工作流…
• 自动化机器学习简介：自动化机器学习…
• 2025 年用于机器学习工作流的 7 个 AI 智能体框架
• cuML 的实践入门：用于 GPU 加速的…
• AI 如何为数据科学降低成本并增加价值…

/.entry