使用 Python 和 LangGraph 构建代理工作流！#大数据 #分析 #数据科学 #AI #机器学习 #自然语言处理 #LLM #物联网 #工业物联网 #PyTorch #Python #RStats #TensorFlow #Java #JavaScript #ReactJS #GoLang #云计算 #无服务器 #数据科学家 #Linux #编程 #编码 #100DaysofCode https://geni.us/LangGraphWorkflows…

使用 Python 和 LangGraph 构建代理工作流 - MachineLearningMastery.com

来源：https://machinelearningmastery.com/building-agentic-workflows-in-python-with-langgraph/ 在本文中，你将学习如何使用 Python 和 LangGraph 构建一个完整的代理工作流，从单个模型调用到带有持久对话记忆的、使用工具的代理。

我们将涵盖以下主题：

• 状态、节点和边如何组合起来定义 LangGraph 代理的执行流程。
• 如何注册一个工具，并通过图的推理循环路由模型的工具调用。
• 检查点器如何跨独立的图调用持久化对话历史。

我们不再浪费时间了。

使用 Python 和 LangGraph 构建代理工作流

简介

大多数AI代理（https://www.ibm.com/think/topics/ai-agents）设置能够很好地处理单轮情况：接收一个问题，调用模型，并返回答案。更难的问题很快就会随之出现。代理可能需要查询你的数据库，记住之前消息的上下文，或者让你清楚了解模型确切决定的内容及其原因。在不针对每个用例构建自定义机制的情况下解决这些挑战，是许多实现开始出现问题的地方。

LangGraph（https://www.langchain.com/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（https://pypi.org/project/python-dotenv/）读取 .env 文件并将密钥设置为环境变量。

理解状态、节点和边

每个 LangGraph 图都由以下三个组件（https://docs.langchain.com/oss/python/langgraph/graph-api#graphs）构建而成。正确理解它们可以避免在图变得更加复杂时产生混淆。

状态是一个 TypedDict（https://typing.python.org/en/latest/spec/typeddict.html），充当整个图的共享内存。每个节点从中读取数据，并将更新写回。节点之间没有其他方式传递数据。你在节点中未更新的字段保持不变；你只返回你想要修改的内容。

节点是普通的 Python 函数。一个节点将当前状态作为其参数，并返回一个字典，包含它想要更新的字段。使用 add_node（https://reference.langchain.com/python/langgraph/graph/state/StateGraph/add_node）注册一个函数使其成为图的一部分，无需特殊的装饰器或基类。如果你只传递函数而不带名称字符串，LangGraph 会自动使用函数名。

边定义执行顺序。add_edge(A, B)（https://reference.langgraph.com/python/langgraph/graph/state/StateGraph/add_edge）意味着：在节点 A 完成后，运行节点 B。add_conditional_edges（https://reference.langchain.com/python/langgraph/graph/state/StateGraph/add_conditional_edges）意味着：在节点 A 完成后，调用一个路由函数并前往它指向的任何地方。每个图都需要 START 作为入口点，并且至少有一条路径通向 END。

默认情况下，当一个节点为一个状态字段返回一个值时，该值会替换原有的内容。对于应该在节点之间累积的字段——比如日志、消息历史——你可以使用 归约器函数（https://docs.langchain.com/oss/python/langgraph/graph-api#reducers）来注释该字段。在下面的例子中，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']}

两个节点都写入了 log，并且两个条目都在。customer_message 保持不变，因为两个节点都没有返回它。这正是 MessagesState 处理其 messages 字段的方式，只是使用了一个稍微更专门的归约器 add_messages，它还能处理消息对象的去重和排序。

使用 MessagesState 管理对话历史

在 LangGraph 图中，每个节点都会读取当前状态并将更新写回。对于对话代理，状态需要携带完整的消息历史——用户输入、模型响应、工具输出——这样模型在决定下一步做什么时始终拥有所需的上下文。

LangGraph 为此内置了一个状态类型：MessagesState（https://reference.langchain.com/python/langgraph/graph/message/MessagesState）。它是一个 TypedDict，包含一个 messages 字段，该字段使用了 add_messages 归约器而不是简单的覆盖。每次节点返回新消息时，它们会被追加到现有列表中，而不是替换它。你不需要手动拼接对话历史。

from langgraph.graph import MessagesState

这是大多数单代理图所需的状态定义。你可以通过添加其他字段来扩展它，比如 customer_id、priority 标志，任何你的节点需要的东西。但 messages 已经存在并且已经设置为累积。

在节点内调用模型

状态就绪后，任何 LangGraph 代理的核心节点都是一个函数，它将当前消息列表传递给模型并追加其响应。模型返回一个 AIMessage（https://reference.langchain.com/python/langchain-core/messages/ai/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 的标准聊天模型接口（https://reference.langchain.com/python/langchain-openai/chat_models/base/ChatOpenAI）封装了 OpenAI API。切换到不同的提供者——Anthropic、Google、通过 Ollama 的本地模型——只需要更改导入和模型字符串；节点的其余部分保持不变。SystemMessage（https://reference.langchain.com/python/langchain-core/messages/system/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] 获取最近的消息。

注册工具并路由工具调用

模型可以回答来自其训练数据的一般性问题，但任何特定于你的数据的问题——账户详情、订阅等级、票务历史——都需要工具调用（https://www.ibm.com/think/topics/tool-calling）。模型决定何时需要工具；你的代码定义它做什么。

使用 @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（https://reference.langchain.com/python/langchain-core/language_models/chat_models/BaseChatModel/bind_tools）在每次请求时将工具的模式发送给模型。当模型决定使用它时，响应会以一个填充了 tool_calls 字段的 AIMessage 形式返回，而不是纯文本内容。

添加一个 ToolNode（https://reference.langchain.com/python/langgraph.prebuilt/tool_node/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.

这里我们有四条消息和两次模型调用。第一个模型调用产生一个 AIMessage，其 tool_calls 已填充，而内容为空。模型正在表明它想做什么，还没有回答。tools_condition 看到这一点，路由到 ToolNode，后者运行 get_customer_tier("cust_1001") 并追加一个包含结果的 ToolMessage。

返回到 run_model 的边再次触发。现在模型在上下文中拥有全部三条先前的消息，理解查找成功，并写入最终的 AIMessage，答案在内容中。tools_condition 再运行一次，发现没有工具调用，并结束图。

这个循环——模型调用、工具执行、再次模型调用——是标准的 ReAct 模式（https://www.ibm.com/think/topics/react-agent）。每次使用工具都会产生两次模型调用：一次决定查找什么，一次解释结果。当你在考虑延迟和成本并添加更多工具时，这是一个有用的信息。

跨调用持久化对话

上面每次 graph.invoke()（https://reference.langchain.com/python/langgraph/pregel/main/Pregel/invoke）都是以一个全新的图状态开始的。没有持久化（https://docs.langchain.com/oss/python/langgraph/persistence），模型就不会记住之前的交互。

为了在调用之间持久化状态，在编译图时附加一个检查点器：

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 将检查点存储在进程内存中，这对于开发和测试很有用。在生产环境中，你通常用由数据库或其他持久存储支持的持久化检查点器替换它。你图代码的其余部分保持不变。

检查点器（https://docs.langchain.com/oss/python/langgraph/checkpointers）为线程持久化图状态。如果你的应用程序还需要独立于任何对话持久化数据，例如用户配置文件、偏好设置或跨多个线程共享的长期记忆，请使用一个 Store。存储（https://docs.langchain.com/oss/python/langgraph/stores）通过提供持久化的应用程序级存储来补充检查点器，图可以在执行期间访问这些存储。

总结

在本文中，你从头开始构建了一个完整的 LangGraph 代理。在