随着前面的文章对 LangGraph 背景、理论与应用场景的铺垫，本章将带你真正动手构建一个可以运行的示例，体验从“链”转向“图”的思维转变。我们不仅会演示 Graph API 和 Functional API 两种方式的使用，还会深入解释 StateGraph、START/END 节点等核心概念，并探讨编译和执行背后的原理。阅读完本篇文章，你将能够独立搭建简单的 LangGraph 工作流，并理解第一次 graph.invoke() 的内部机制。

最小可运行示例

为了让你快速上手，我们先构建一个简单的“聊天计算器”。它接收一个用户输入，例如 2 + 3 * 4，通过调用语言模型决定是加法、乘法还是除法，再调用对应的工具完成计算，最后返回结果。这个例子取自官方快速入门文档并进行了改进，以便与后面的文章贯通 。

Step 1：定义工具

LangGraph 本身不提供计算能力，我们需要通过 LangChain 提供的 @tool 装饰器定义工具函数。每个工具都具有明确的输入和输出。

from langchain.tools import tool

@tool
def add(a: int, b: int) -> int:
    """Adds `a` and `b`."""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """Multiplies a and b."""
    return a * b

@tool
def divide(a: int, b: int) -> float:
    """Divides a by b."""
    return a / b

我们随后将这些工具绑定给聊天模型，使模型可以通过函数调用来使用它们.

Step 2：定义状态（State）

在 LangGraph 中，状态是灵魂。它记录当前执行的上下文，可在节点间共享。我们定义一个 MessagesState，其中包含用户发送的消息列表以及调用次数：

from typing import TypedDict
from typing_extensions import Annotated
import operator

class MessagesState(TypedDict):
    messages: Annotated[list, operator.add]  # 更新时列表相加
    llm_calls: int

这里 Annotated 和 operator.add 表示在多个节点并行更新同一字段时采用相加策略。LangGraph 支持自定义 reducer，还可使用 Pydantic 模型定义状态 。

Step 3：定义节点（Nodes）

一个节点就是一个接收 state 输入并返回部分状态的函数；它可能会调用 LLM、工具或执行普通逻辑。我们需要两个节点：

1. llm_node：调用 LLM 分析用户输入，决定是否调用工具。
2. tool_node：根据 LLM 的 tool_calls 输出，调用相应的工具并将结果包装为消息。

from langchain.chat_models import init_chat_model
from langchain.messages import SystemMessage, ToolMessage, AnyMessage

model = init_chat_model("claude-sonnet-4-5-20250929", temperature=0)
model_with_tools = model.bind_tools([add, multiply, divide])

def llm_node(state: MessagesState) -> dict:
    """调用 LLM，让模型判断是否需要调用工具"""
    system_prompt = SystemMessage(
        content="你是一个帮用户进行算术计算的助手，可以调用工具实现加法、乘法或除法。"
    )
    response = model_with_tools.invoke([system_prompt] + state["messages"])
    return {
        "messages": [response],
        "llm_calls": state.get("llm_calls", 0) + 1
    }

def tool_node(state: MessagesState) -> dict:
    """执行 LLM 指定的工具调用，并返回工具结果"""
    last_message = state["messages"][-1]
    results: list[ToolMessage] = []
    for tool_call in last_message.tool_calls:
        tool_fn = tools_by_name[tool_call["name"]]
        observation = tool_fn.invoke(tool_call["args"])
        results.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return {"messages": results}

Step 4：定义条件边（Edges）

边用来决定接下来执行哪个节点，这对代理的控制流至关重要 。我们写一个路由函数，检查 LLM 输出是否包含 tool_calls：

from typing import Literal

def should_continue(state: MessagesState) -> Literal["tool_node", "end"]:
    # 如果有工具调用，则去执行工具；否则结束
    return "tool_node" if state["messages"][-1].tool_calls else "end"

当返回的值为节点名称时，LangGraph 会自动跳转到对应节点；如果返回 END，表示流程结束 。

Step 5：搭建并编译图（Graph）

下面使用 Graph API 构建完整的状态图：

from langgraph.graph import StateGraph, START, END

builder = StateGraph(MessagesState)
builder.add_node("llm_node", llm_node)
builder.add_node("tool_node", tool_node)

# 从 START 进入第一个节点
builder.add_edge(START, "llm_node")

# 使用条件边，根据 should_continue 的返回值决定是否转向工具节点或结束
builder.add_conditional_edges("llm_node", should_continue, {"tool_node": "tool_node", "end": END})

# 工具调用完成后再次回到 llm_node
builder.add_edge("tool_node", "llm_node")

# 编译图
graph = builder.compile()

编译是 LangGraph 的必要步骤，它会检查图结构是否合理（例如不存在孤立节点）并允许传入运行时配置。

Step 6：运行图

最后，通过 invoke 方法执行图。此方法接收一个状态字典作为输入，返回更新后的状态。可以选择 stream_mode 以流式查看每一步状态更新：

initial_state = {"messages": [SystemMessage(content="2 + 3 * 4")], "llm_calls": 0}
result = graph.invoke(initial_state, stream_mode="updates")
for update in result:
    print(update)

在首次执行过程中，LangGraph 会按照节点和边的顺序调用 llm_node → tool_node → llm_node，直到没有工具调用为止，此时返回结果。

这个示例展示了构建图式代理的完整流程，同时体现了三个核心概念：状态、节点和边。当掌握了这些基础，就可以定义更复杂的流程，如并行执行、多代理协作、持久化和人机协同等。

Graph API 与 Functional API 对比

LangGraph 提供两套 API：Graph API 和 Functional API。它们共享同一运行时，但设计思想和适用场景不同 。下面从用途、开发体验和可视化三个方面进行对比。

Graph API：显式控制与可视化

Graph API 鼓励开发者通过声明式方式定义节点、边和状态，适合复杂决策树、多分支和并行任务的场景 。它的特点包括：

1. 清晰的流程图：Graph API 可以生成图形化表示，便于调试和团队协作。当工作流包含多个决策点时，图的结构使得分支逻辑一目了然 。
2. 共享状态管理：多个节点可以访问和修改同一份状态，在多工具、多模型协作时非常重要 。
3. 并行处理：Graph API 支持在同一 super-step 中运行多个节点，并在下一个 super-step 中同步结果 。
4. 团队开发友好：不同团队成员可以专注于各自负责的节点或子图，最后通过 StateGraph 组合成整体流程 。

然而，Graph API 需要显式编写状态定义和边的逻辑，对于简单线性流程而言显得繁琐。

Functional API：最小侵入式集成

Functional API 面向已有代码或简单流程，它通过装饰器将函数标记为任务（@task）或入口点（@entrypoint），并支持在函数体内直接使用普通的 Python 控制流 。适用场景包括：

1. 现有代码集成：如果你已有使用 if/else、for 等流程的代码，只需在入口点和关键函数上添加装饰器即可启用 LangGraph 特性，例如持久化、人机协同和流式处理 。
2. 线性流程：对于简单的顺序任务，例如撰写文章、上传文件等，Functional API 可以减少模板代码，使开发体验接近普通函数调用 。
3. 快速原型：当你需要快速迭代想法、不想定义复杂状态结构时，可以使用 Functional API 快速编排逻辑 。
4. 局部状态管理：状态仅在函数内部使用，不需要被多个节点共享时，Functional API 更加合适 。

Functional API 牺牲了一定的可视化和显式控制，但保留了 LangGraph 的核心特性，并可与 Graph API 组合使用。例如，可以在 Graph API 中调用一个 Functional API 的入口函数来处理子任务 。

示例对比：一个“审批流程”

下面通过一个简化的审批流程示例比较两种 API。在流程中，系统先处理用户申请，如果需要人工审批则暂停并等待用户反馈，否则直接结束。

Graph API 实现：

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Literal

def process_application(state):
    # 处理申请逻辑
    approved = state["score"] >= 60
    return {"approved": approved}

# 人工审批节点
from langgraph.graph import interrupt

def wait_for_approval(state):
    # 调用 interrupt 等待人工反馈
    feedback = interrupt({"action": "请审批该申请", "data": state}).result()
    # 根据反馈更新状态
    return {"approved": feedback == "agree"}

def decide_next(state) -> Literal["wait", END]:
    # 如果系统自动审批未通过，则等待人工审批
    return "wait" if not state["approved"] else END

class AppState(TypedDict):
    score: int
    approved: bool

builder = StateGraph(AppState)
builder.add_node("process", process_application)
builder.add_node("wait", wait_for_approval)

# 起始节点
builder.add_edge(START, "process")
# 根据系统审批结果决定是否转人工
builder.add_conditional_edges("process", decide_next, {"wait": "wait", END: END})
# 人工审批后结束
builder.add_edge("wait", END)

app_graph = builder.compile()
# 运行
result = app_graph.invoke({"score": 50, "approved": False}, stream_mode="updates")

Graph API 通过条件边实现自动与人工审批的切换，状态对象明确地定义了 score 和 approved 两个字段。若审批未通过，interrupt 会暂停流程等待人工输入。

Functional API 实现：

from langgraph.func import entrypoint, task
from langgraph.graph import interrupt

@task
def process_application(score: int) -> dict:
    return {"approved": score >= 60}

@task
def wait_for_approval(data: dict) -> dict:
    feedback = interrupt({"action": "请审批该申请", "data": data}).result()
    return {"approved": feedback == "agree"}

@entrypoint()
def approval_workflow(score: int) -> dict:
    # 调用任务并获取结果
    res = process_application(score).result()
    if not res["approved"]:
        res = wait_for_approval(res).result()
    return res

# 调用入口点
result = approval_workflow.invoke(75)

Functional API 将相同的逻辑写成一个普通的 Python 函数 approval_workflow，利用 if 判断是否需要人工审批，并在需要时调用 interrupt。不需要显式定义状态结构，代码更接近日常编程体验。此示例能帮助你根据业务复杂程度选择合适的 API。

StateGraph的作用与内部结构

StateGraph 是 LangGraph 中的核心类，它负责管理整个图的状态定义、节点和边，并在编译后生成可执行的图实例。官方文档指出，StateGraph 类接收一个用户定义的状态对象作为参数 。其作用包括：

1. 维护状态结构：StateGraph 在初始化时接受一个 TypedDict 或 Pydantic 模型，定义全局状态的字段和 reducer 。
2. 注册节点：通过 add_node(name, node_fn) 将函数或可运行对象绑定为节点，每个节点都有唯一名称。
3. 建立边：使用 add_edge、add_conditional_edges 等方法指定节点之间的连接关系。你还可以从虚拟 START 节点或向 END 节点添加边 。
4. 设置入口和终止点：除了使用 add_edge(START, first_node) 方式指定入口外，也可以通过 set_entry_point() 和 set_finish_point() 来直接声明 。
5. 编译图：调用 .compile() 将构建器转为可执行对象。这一步会检查图的完整性、设置运行时参数、注册缓存或持久化配置 。

在内部，LangGraph 采用 Pregel 风格的 message passing 算法：节点通过边传递消息（包含部分状态），每一个 super-step 对多个节点并行处理，当所有节点均处于 inactive 状态且没有消息在通道中流动时，图执行结束 。

START与 END节点的真实含义

在 Graph API 中，START 和 END 是两个特殊的虚拟节点，用于标记流程的入口和出口 。
• START：表示向图发送初始输入的入口。通过 add_edge(START, “node_a”) 或 set_entry_point(“node_a”)，你告诉 LangGraph 第一个要执行的节点是什么 。
• END：表示流程的终点。将某个节点连接到 END，意味着在执行完该节点后停止流程 。

这些节点的存在使得 LangGraph 在建图时不必关心外部输入和输出的源头，只需声明 START 与第一个节点连接、某些节点与 END 连接即可。这样，编译后的图可以独立运行，也方便对流程进行迁移或嵌入其他系统。

第一次 graph.invoke()背后发生了什么

当你调用 graph.invoke(initial_state) 时，LangGraph 会启动运行时调度流程并返回执行结果。理解这一过程有助于优化性能和调试。

1. 状态初始化：传入的字典会被转换为内置的状态对象，初始化 reducer 的默认值。
2. 激活入口节点：运行时将 START 节点发送的消息传递给声明的入口节点，这些节点进入 active 状态 。
3. 执行 super-step：运行时按照拓扑顺序执行处于 active 状态的节点。每个节点执行完毕后产生一个或多个状态更新，并通过边发送消息给下一个节点或 END。
4. 更新全局状态：消息中的部分状态会通过 reducer 合并到全局状态。如果节点返回的是 Send 对象，则会为下游节点创建独立的状态副本，用于并行处理 。
5. 判断终止条件：当所有节点都标记为 inactive 且没有消息在通道中时，运行时结束执行 。如果某条边连接至 END，对应的状态会立即结束并不再往下传递。
6. 返回结果：invoke 的返回值通常是最终的状态；如果使用 stream_mode=“updates”，则会返回迭代器，每次迭代输出节点返回的状态补丁，方便实时查看进度。
7. 持久化与回溯：如果编译时提供了 checkpointer，则运行时会在每一步自动保存状态，支持持久化恢复、时间旅行和人机协同等高级特性，后续文章将详述。

这种基于消息传递的执行方式带来了显式的调度控制，使得代理既可以在单机上运行，也可以在分布式环境中并行调度。此外，LangGraph 还允许在 invoke 时传入配置，如自定义异步执行、调试模式、缓存策略等，使得调试和性能优化更灵活。

小结与展望

本篇文章通过构建一个“聊天计算器”示例，详细介绍了 LangGraph 从定义工具到构建节点与边，再到编译和执行的完整流程。我们比较了 Graph API 与 Functional API 的差异，强调 StateGraph 在状态管理与流程编排中的核心作用，解释了 START 和 END 节点的意义，并揭示了 graph.invoke() 的内部执行机制。通过阅读本篇文章，你应该具备以下能力：
• 理解 LangGraph 的基本概念：状态、节点、边，以及它们如何组合成工作流。
• 使用 Graph API 快速搭建一个简单的有向图代理，并在节点间传递消息。
• 选择适合的 API（Graph 或 Functional）来满足不同的业务需求。
• 理解编译和执行过程，能够调试和优化运行时行为。

在下一篇文章中，我们将深入研究 状态管理 和 节点设计，探讨如何通过合理划分状态和封装逻辑，让 Agent 系统既可扩展又易于维护。