快速演示如何使用 LangGraph 的 Graph API 构造一个可以进行 tool-calling 的有状态的模型执行链路。

1. 需要的Python依赖:

typing-extensions
pyyaml
langgraph(1.0.3)
langchain-openai(1.0.3)
langchain(1.0.8)

2. 具体代码：(不算特别长，关键节点添加了中文注释)

import yaml
import operator
from pathlib import Path
from typing import Literal
from typing_extensions import TypedDict, Annotated

from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langchain.messages import SystemMessage,HumanMessage,ToolMessage,AnyMessage


config_path = Path(__file__).parent.parent.parent / "config.yaml"
with open(config_path, 'r', encoding='utf-8') as f:
    config = yaml.safe_load(f)

# 初始化模型（从配置文件读取模型配置）
model = ChatOpenAI(
    model=config['model'],
    api_key=config['api_key'],
    base_url=config['base_url'],
    temperature=config['temperature']
)

# 定义 tools
@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
tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)

# 定义 State 存储 messages 和模型调用次数
class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add] # 追加方式将AI TOOL Human等类型 Message 追加到 State 的 messages 中
    llm_calls: int

# 定义调用模型的 Node
def llm_call(state: dict):
    """LLM decides whether to call a tool or not"""
    return {
        "messages": [
            model_with_tools.invoke(
                [
                    SystemMessage(
                        content="You are a helpful assistant tasked with performing arithmetic on a set of inputs."
                    )
                ]
                + state["messages"] # 这里会将调用模型之前 State 存储的 messages 都给到模型用于分析决策
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1 # 调用模型次数+1
    }


# 定义调用 tool 的 Node
def tool_node(state: dict):
    """Performs the tool call"""
    result = []
    for tool_call in state["messages"][-1].tool_calls: # 如果调用 tool_node 的上一步模型决策了调用 tool_calls 则进入
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"]) # 通过模型提供的参数调用我们的 tool 方法
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"])) # 封装 tool 结果到 State 的 messages 列表中
    return {"messages": result}

# 定义条件边：根据上一步模型是否决定进行工具调用，来路由到 tool_node 或结束。
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]

    if last_message.tool_calls:
        return "tool_node"

    return END

# 创建工作流
agent_builder = StateGraph(MessagesState)

agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)

agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    ["tool_node", END]
)
agent_builder.add_edge("tool_node", "llm_call")

agent = agent_builder.compile()

messages = [HumanMessage(content="Add 3 and 4.")]
messages = agent.invoke({"messages": messages})
for m in messages["messages"]:
    m.pretty_print()

3. 如何进行 tools 绑定

model_with_tools = model.bind_tools(tools) 会将 tools 列表封装进 ChatOpenAI Client 里面，每次发送请求到模型的时候都会带着以下tools列表：

"tools": [
  {
    "type": "function",
    "function": {
      "name": "add",
      "description": "Adds `a` and `b`.\n\n    Args:\n        a: First int\n        b: Second int",
      "parameters": {
        "properties": {
          "a": {
            "type": "integer"
          },
          "b": {
            "type": "integer"
          }
        },
        "required": [
          "a",
          "b"
        ],
        "type": "object"
      }
    }
  },
  ......
]

4. 运行结果

======= Human Message=======
Add 3 and 4.
========= Ai Message =========
Tool Calls:
add (toolu_bdrk_01FE3FwxHotJXhdmMYZigTw9)
Call ID: toolu_bdrk_01FE3FwxHotJXhdmMYZigTw9
Args:
a: 3
b: 4
======== Tool Message ========
7
========= Ai Message =========
The sum of 3 and 4 is 7.

5. MessagesStaten内容解析

• 第一条是 messages = agent.invoke({"messages": messages}) 传递进来的用户需求
• 第二条是调用llm_call Node 模型根据用户需求决策启用tool_calls调用add
• 第三条是通过should_continue判断出模型启用了tool_calls然后调用tool_node Node进行工具执行
• 第四条是tool_calls将结果返回给llm_call ，模型决策终止并返回最终结果

[
    HumanMessage(
        content='Add 3 and 4.', 
        additional_kwargs={}, 
        response_metadata={}
    ), 

    AIMessage(
        content='', 
        additional_kwargs={'refusal': None}, 
        response_metadata={'token_usage': {'completion_tokens': 68, 'prompt_tokens': 792, 'total_tokens': 860, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': None, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': None}}, 'model_provider': 'openai', 'model_name': 'us.anthropic.claude-sonnet-4-5-20250929-v1:0', 'system_fingerprint': None, 'id': 'msg_bdrk_01AhhYfDuETVpPaaSkT8jjqe', 'finish_reason': 'tool_calls', 'logprobs': None}, 
        id='lc_run--6798fe8d-04cb-4dc5-a789-a1722dd20881-0', 
        tool_calls=[{'name': 'add', 'args': {'a': 3, 'b': 4}, 'id': 'toolu_bdrk_01MhfmzVR8NzTT4vENew8bVy', 'type': 'tool_call'}], usage_metadata={'input_tokens': 792, 'output_tokens': 68, 'total_tokens': 860, 'input_token_details': {}, 'output_token_details': {'reasoning': 0}}
    ), 

    ToolMessage(
        content='7', 
        tool_call_id='toolu_bdrk_01MhfmzVR8NzTT4vENew8bVy'
    ), 

    AIMessage(
        content='The sum of 3 and 4 is **7**.', 
        additional_kwargs={'refusal': None}, 
        response_metadata={'token_usage': {'completion_tokens': 17, 'prompt_tokens': 874, 'total_tokens': 891, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': None, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': None}}, 'model_provider': 'openai', 'model_name': 'us.anthropic.claude-sonnet-4-5-20250929-v1:0', 'system_fingerprint': None, 'id': 'msg_bdrk_019TXKeRfw4iFZ9hfFiAakJ7', 'finish_reason': 'stop', 'logprobs': None}, 
        id='lc_run--3059b9eb-6e90-4fc1-a34b-ad5075d06e0b-0', 
        usage_metadata={'input_tokens': 874, 'output_tokens': 17, 'total_tokens': 891, 'input_token_details': {}, 'output_token_details': {'reasoning': 0}}
    )
]

6. 三个 invoke() 方法解析

代码里面一共调用了 invoke() 方法了三次，每次功能都不一样

6.1. ChatOpenAI.invoke()

用于进行模型API的调用，即和模型对话

1. 将 LangChain 消息对象 → JSON 格式
2. 添加模型参数（model, temperature）
3. 添加绑定的工具信息（tools）
4. 发送 HTTPS POST 请求到您的网关
5. 等待响应（阻塞）
6. 解析 JSON 响应
7. 将 JSON → LangChain AIMessage 对象
8. 返回消息对象

6.2. Tool.invoke()

执行我们的 add tool,即执行 Python 代码

1. 接收参数 {“a”: 3, “b”: 4}
2. 调用 Python 函数 add(a=3, b=4)
3. 执行函数体 return a + b → 7
4. 将结果转换成字符串 “7”
5. 返回结果

6.3. CompiledGraph.invoke()

执行 StateGraph 工作作流，即进行 start -> node -> edge -> node -> end

1. 初始化状态
2. 从 START 开始遍历图
3. 每次执行一个节点函数（llm_call 或 tool_node）
4. 将节点返回的更新合并到状态中
5. 根据条件边（should_continue）决定下一个节点
6. 循环执行直到到达 END
7. 返回最终的完整状态

参考文档：LangGraph Quickstart