Agent开发人工智能

LangGraph 入门：用 StateGraph 构建 Agent 的五步流程

从定义状态、节点、连边到条件路由和编译运行，用 LangGraph 把 Stage 3 手写的 ReAct 循环变成一张可执行的图。核心记忆口诀：State 是数据，Node 是动作，Edge 是流程，条件边是 agent 的自主性所在。

2026/6/17
13 分钟阅读
目录
用 LangGraph 写Agent的流程思路：1、定义模型和工具2、定义状态3、定义模型节点4、定义工具节点5、 定义结束逻辑6、构建并编译Agent程序

用 LangGraph 写Agent的流程思路：

  1. 定义状态 State（TypedDict）——图的共享数据/记忆，用 reducer（operator.add）让消息历史自动累积。
  2. 定义节点 Node（普通函数）——模型节点 llm_call（大脑，负责判断调不调工具）、工具节点 tool_node（手，负责执行工具）。
  3. 创建并配置图：StateGraph(State) 建画布 → add_node 把节点挂上去。
  4. 连边定义流转：普通边 add_edge 定固定走向（START→llm_call、tool_node→llm_call）；条件边 add_conditional_edges + 路由函数定动态分支（继续调工具 or 结束）——这是循环和自停止的核心。
  5. 编译运行：compile() 定稿成可执行 agent，invoke() 跑起来。

一句话记忆：State 是数据，Node 是动作，Edge 是流程，条件边是 agent 的"自主性"所在。

1、定义模型和工具

用的deepseek-v4-pro模型，并定义工具

import os

from langchain.tools import tool
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv

# 从根目录 .env 读取 DeepSeek 配置（key / base_url / 模型名）
load_dotenv()

# init_chat_model：LangChain 的通用模型初始化器。
# 这里接 DeepSeek（OpenAI 兼容），temperature=0 让输出更稳定、可复现。
model = init_chat_model(
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url=os.environ["DEEPSEEK_BASE_URL"],
    model=os.environ["MODEL"],
    temperature=0
)


# 定义工具：@tool 装饰器把一个普通 Python 函数变成 agent 可调用的工具。
# 关键：函数的 docstring 会成为工具给模型看的“说明书”——模型靠它判断何时调用、怎么传参。
# 这正是 Stage 3 练过的“工具 schema”，只是 LangChain 帮你从函数签名+docstring 自动生成了。
@tool
def multiply(a: int, b: int) -> int:
    """Multiply `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a * b


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

    Args:
        a: First int
        b: Second int
    """
    return a + b


@tool
def divide(a: int, b: int) -> float:
    """Divide `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a / b


# 把三个工具登记成列表
tools = [add, multiply, divide]
# 名字 -> 工具对象 的映射，后面 tool_node 执行时按名字查找用
tools_by_name = {tool.name: tool for tool in tools}
# bind_tools：把工具“绑”到模型上。绑定后模型在回答时就能输出 tool_calls（要调哪个工具+参数）
# 注意：绑定不等于执行——模型只负责“决定调用”，真正执行仍由我们的代码（tool_node）来做。
model_with_tools = model.bind_tools(tools)

2、定义状态

图的状态用于存储消息和 LLM 调用次数。

from langchain.messages import AnyMessage
from typing_extensions import TypedDict, Annotated
import operator


# 图的“状态”（State）：贯穿整张图、在节点之间传递的共享数据。
# 每个节点读它、改它，再把更新返回——这就是 LangGraph 的数据流核心。
class MessagesState(TypedDict):
    # messages：对话历史。Annotated[..., operator.add] 是关键——
    # 它告诉 LangGraph：节点返回的新消息要【追加】到列表，而不是覆盖。
    # （这就是 Stage 3 手写循环里“messages 越滚越长”那件事，框架用 reducer 替你做了。）
    messages: Annotated[list[AnyMessage], operator.add]
    # llm_calls：记录调用了几次 LLM，方便观察循环转了几圈 / 控制成本。
    llm_calls: int

3、定义模型节点

模型节点用于调用 LLM 并决定是否调用工具。

from langchain.messages import SystemMessage

# 模型节点（agent 的“大脑”）：让 LLM 看完当前对话后，决定“调工具”还是“直接回答”。
def llm_call(state: dict):
    """LLM decides whether to call a tool or not"""

    return {
        # 调用绑了工具的模型。输入 = 系统提示 + 到目前为止的全部对话历史。
        # 返回的 AIMessage 里可能带 tool_calls（要调工具），也可能是纯文本（直接回答）。
        "messages": [
            model_with_tools.invoke(
                [
                    SystemMessage(
                        content="You are a helpful assistant tasked with performing arithmetic on a set of inputs."
                    )
                ]
                + state["messages"]  # 拼上历史消息，模型才有上下文
            )
        ],
        # 每经过一次本节点就 +1，等于给“心跳”计数
        "llm_calls": state.get('llm_calls', 0) + 1
    }

4、定义工具节点

工具节点用于调用工具并返回结果。

from langchain.messages import ToolMessage


# 工具节点（agent 的“手”）：真正去执行上一步模型决定调用的工具。
def tool_node(state: dict):
    """Performs the tool call"""

    result = []
    # 取最后一条消息（来自 llm_call 的 AIMessage），遍历它要求的每个工具调用。
    # 一条消息里可能有多个 tool_calls —— 模型可以一次要求并行调多个工具。
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]          # 按名字找到对应函数
        observation = tool.invoke(tool_call["args"])      # 用模型给的参数真正执行
        # 把结果包成 ToolMessage 返回。tool_call_id 必须和发起调用的 id 配对，
        # 模型才知道这条结果是哪次调用的回执（Stage 3 手写时要自己管，这里框架替你对齐）。
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return {"messages": result}

5、 定义结束逻辑

条件边函数用于根据 LLM 是否发出工具调用来路由到工具节点或终点。

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


# 条件边函数：决定循环“继续”还是“结束”——这就是 agent 的【自停止条件】。
# 返回值是下一个要去的节点名（字符串），LangGraph 据此决定走向。
def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    """Decide if we should continue the loop or stop based upon whether the LLM made a tool call"""

    messages = state["messages"]
    last_message = messages[-1]

    # 模型还要调工具 → 去 tool_node 执行，然后还会绕回来继续想（循环继续）
    if last_message.tool_calls:
        return "tool_node"

    # 模型不再调工具（只给文字）→ 结束，把答案返回给用户
    # 这等价于 Stage 3 手写循环里的 `if finish_reason == "stop": break`
    return END

6、构建并编译Agent程序

该Agent是使用该类构建的StateGraph，并使用该compile方法进行编译的。

# 构建工作流：StateGraph 是“图”的画布，传入状态类型让节点共享同一份 State
agent_builder = StateGraph(MessagesState) # 记忆

# 添加节点：给图登记两个“工序”——大脑(llm_call) 和 手(tool_node)
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)

# 连边：定义节点之间怎么走（这就是把 Stage 3 的循环“画”出来）
agent_builder.add_edge(START, "llm_call")           # 入口 → 先进大脑
# 条件边：从 llm_call 出来后，由 should_continue 决定去 tool_node 还是 END
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    ["tool_node", END]                              # 可能的去向(工具或者结束)
)
agent_builder.add_edge("tool_node", "llm_call")     # 工具执行完 → 回到大脑（这条边形成“循环”）

# 编译：把图“定稿”成可执行的 agent
agent = agent_builder.compile()

# 可视化：画出这张图的结构，直观看到 llm_call ⇄ tool_node 的循环
from IPython.display import Image, display
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

# 运行：给一个任务，invoke 会自动按图把循环跑完，直到 should_continue 返回 END
from langchain.messages import HumanMessage
messages = [HumanMessage(content="Add 3 and 4.")]
messages = agent.invoke({"messages": messages})
# pretty_print 逐条打印消息，能清楚看到：用户提问 → 模型决定调 add → 工具结果 → 模型给出答案
for m in messages["messages"]:
    m.pretty_print()
返回文章列表