揭示工具调用的真实机制(不是 LLM 直接调用而是框架中介),对比 PydanticAI、LangChain、MCP 的实现差异和代理框架的真实价值。
工具调用究竟是如何发生的?
这就是促使我开始这项研究的问题。
我不断看到这样的示例:
@agent.tool
def get_weather(city: str) -> dict:
return weather_api.fetch(city)
然后,相关解释往往会轻描淡写地说:
“LLM 调用了 get_weather 函数。”
LLM 会进入我的 Python 进程并执行这个函数吗?
它知道这个函数位于内存中的什么位置吗?
它能直接访问我的数据库、本地文件或私有 API 吗?
如果 LLM 只是接收输入并生成输出,它怎么可能调用任何东西?
而且,如果模型提供商已经支持原生工具调用,为什么我们还需要 PydanticAI、LangChain 或 LangGraph 这样的智能体框架?
这又引出了更多问题:
究竟是谁在执行函数?
函数的执行结果如何传回模型?
工具调用是否只是一种复杂的提示技术?
LangChain 是否只是在 LLM 外围添加了一些指令?
我能否自己构建整个循环?
除了这个循环之外,智能体框架还提供了什么?
MCP 与工具调用是一回事吗?
每个 LLM 应用都需要智能体框架吗?
我最初的思维模型非常简单:
输入 → LLM → 输出
这个模型似乎与下面这样的说法互不相容:
LLM 调用了一个工具。
所以,我决定不再把这个过程当作魔法,而是深入真实的框架代码,追踪数据的完整流转过程。🔍
本文记录了我的发现。
首先:我是如何验证本文观点的 🔬
在解释架构之前,我想先明确说明本文的证据来源。
本文并非仅仅基于架构图、AI 生成的解释或二手教程。当然,我在创作本文时使用了 AI 的帮助,但我可以向你保证,我在研究过程中验证了这些结果,因此你可以信赖本文的结论。
PydanticAI 和 LangChain 的公开文档。
当前的智能体编排实现。
验证模型生成的工具参数的代码。
调用已注册工具的确切代码行。
将函数执行结果重新转换为模型消息的代码。
固定到具体 commit 的修订版本,以便即使 main 分支发生变化,相关证据仍然可供检查。
源码审查时间
本文的源码审查完成于:
August 3, 2026 at 20:42 CEST
已检查的源码快照
下面列出的 commit,是我在审查时针对特定文件找到的最新修订版本。它们是文件级别的修订版本,不一定是整个仓库的 HEAD commit。
本文会在相关位置链接到固定了 commit 的文件。
这一点很重要,因为框架的内部代码会发生变化。
指向 main 分支的链接展示的是项目当前包含的内容。固定到具体 commit 的链接则准确展示了我撰写本文时所检查的内容。
实际示例使用 Python,但这种架构并不局限于 Python。
TypeScript、Go、C#、Rust 及其他语言中都存在相同的循环:
模型请求
→ 结构化工具请求
→ 运行时验证
→ 普通函数执行
→ 工具结果消息
→ 再次发起模型请求
Python 只是为我们提供了可读性较高、便于检查的框架实现。
一段话说清答案 💡
LLM 不会直接执行我的函数。
它生成的输出可能表示一个函数调用请求。
我的应用程序——或者运行在其中的智能体框架——会读取该请求,将它与可用工具进行匹配,验证参数,检查权限,执行真正的函数,捕获结果,将结果返回给模型,然后继续循环。
模型提出一个动作。运行时负责验证、管控并执行它。
即使是原生工具调用,本质上仍然是结构化的模型输出。
模型响应中可能包含逻辑上类似以下内容的结构:
{
"name": "get_weather",
"arguments": {
"city": "Tokyo"
}
}
get_weather() 此时并没有运行。
没有请求任何天气 API。
没有调用任何 Python 函数。
没有产生任何外部副作用。
模型只是生成了一个结构化请求。
必须由其他东西来执行它。
这个“其他东西”可能是应用程序代码、框架代码,或者——对于某些由提供商托管的工具——由提供商运营的运行时基础设施。
它并不是神经网络本身。
参与其中的四个角色 🧩
当我把相关组件拆分开来后,这套架构就清晰多了:
提供商 API 或模型适配器。
应用程序或智能体框架。
实际的工具实现。
它们相互协作,但职责并不相同。
从本质上讲,模型执行的过程在概念上类似于:
输入 token → 模型计算 → 输出 token
根据提供商的不同,这些输出 token 可能被解释为:
自然语言文本。
结构化内容块。
文本与动作请求的混合内容。
原始模型无法直接:
读取我计算机上的任意文件。
使用我的账户发送电子邮件。
修改应用程序状态。
重启我的生产服务器。
它可以生成描述这些动作的文本。
它也可以生成结构化输出,请求外围软件执行其中某个动作。
这种区别是工具调用的基础。
模型提供商定义了相应规范,用于向模型展示工具,以及向应用程序返回工具请求。
模型请求中可能包含:
对话消息。
工具参数的 JSON Schema。
工具选择设置。
提供商的响应中可能包含:
普通的助手文本。
提供商特有的内容块。
文本与工具请求的组合。
提供商 API 让应用程序代码更容易解释模型生成的动作请求。
它不会自动拥有或执行运行在我的应用程序内部的本地函数。
提供商可能知道以下内容:
{
"name": "get_customer",
"description": "Retrieve a customer by ID.",
"parameters": {
"type": "object",
"properties": {
"customer_id": {
"type": "string"
}
},
"required": ["customer_id"]
}
}
但它不会自动拥有以下实现:
def get_customer(customer_id: str) -> dict:
return my_private_database.find_customer(customer_id)
Schema 和具体实现是两回事。
一个重要的例外:提供商托管的工具
一些提供商会提供运行在自身基础设施上的工具,例如托管式网页搜索、文件搜索或代码执行。
在这些情况下,工具可能由提供商的软件执行。
但即便如此,也不是模型权重在直接操作服务器。
模型生成动作请求,然后由受控的提供商运行时执行该请求。
该架构仍然会区分:
模型决策
外部执行
基于提示的工具使用与原生工具调用
在原生工具 API 普及之前,应用程序通常会这样告诉模型:
如果需要使用工具,请返回包含以下内容的 JSON:
- 工具名称
- 参数
模型可能会这样回答:
{
"name": "get_customer",
"arguments": {
"customer_id": "cust-1001"
}
}
然后,应用程序会:
解析工具名称。
验证参数。
执行函数。
将结果发送回模型。
使用原生工具调用时,应用程序会通过专用 API 字段发送工具定义,提供商则会返回专用的工具调用结构。
{
"tool_calls": [
{
"id": "call_123",
"name": "get_customer",
"arguments": {
"customer_id": "cust-1001"
}
}
]
}
这两种方式都涉及由模型生成动作请求。
原生工具调用通常可以提供:
更完善的消息结构。
由提供商支持的调用 ID。
更可靠的参数处理。
更方便的多次调用支持。
更清晰的文本与动作分离。
更稳健的输出解析。
但它并没有消除执行循环。
应用程序或智能体框架负责整个编排过程。
它通常会执行以下操作:
向模型发送消息和可用工具的定义。
读取模型响应。
检测工具请求。
将每个请求中的名称与具体实现进行匹配。
验证参数。
检查授权和审批规则。
执行实际工具。
捕获工具的返回结果或异常。
创建工具结果消息。
将更新后的对话再次发送给模型。
持续执行,直到模型返回最终答案。
生产级框架还可能处理:
无效参数纠正。
依赖注入。
这一编排层正是智能体框架大部分价值的来源。
工具本身就是普通代码:
def get_weather(city: str) -> dict[str, object]:
return weather_api.fetch(city)
async function getWeather(city: string): Promise<Weather> {
return weatherApi.fetch(city);
}
语言并不改变架构边界。
函数之所以运行,是因为某个运行时调用了它。
神经网络无法直接进入语言运行时并调用它。
让我们在整篇文章中使用一个工具:
def get_customer(customer_id: str) -> dict[str, str]:
...
customer cust-1001 使用的是什么订阅计划?
以下是完整的生命周期。
步骤 1:应用向模型发送工具定义
应用发送:
get_customer 工具定义。一个简化后的定义可能看起来像:
{
"name": "get_customer",
"description": "Retrieve a customer by customer ID.",
"parameters": {
"type": "object",
"properties": {
"customer_id": {
"type": "string"
}
},
"required": ["customer_id"]
}
}
模型现在可以推断出 get_customer 可能有助于回答这个问题。
它仍然无法直接执行实现代码。
步骤 2:模型生成工具请求
响应可能被规范化为:
{
"tool_calls": [
{
"id": "call_123",
"name": "get_customer",
"arguments": {
"customer_id": "cust-1001"
}
}
]
}
此时,customer 函数仍然没有运行。
模型实际上是在写一张请求单据:
请运行 get_customer,参数为 customer_id="cust-1001"。
步骤 3:运行时验证请求
应用或框架会问:
get_customer 存在吗?customer_id 参数存在吗?cust-1001?只有在这些检查之后,运行时才应该执行该工具。
步骤 4:普通软件调用函数
框架执行逻辑上等价于以下代码的操作:
result = get_customer(customer_id="cust-1001")
函数可能返回:
{
"id": "cust-1001",
"name": "Alice",
"plan": "pro"
}
这是实际工作发生的时刻。
数据库被应用代码查询了。
模型生成了提议。
步骤 5:运行时创建工具结果消息
函数的返回值必须在对话中表示:
{
"role": "tool",
"tool_call_id": "call_123",
"content": {
"id": "cust-1001",
"name": "Alice",
"plan": "pro"
}
}
提供商的格式不同,但目的相同:
步骤 6:再次调用模型
下一个模型请求包含:
模型现在可以回答:
Customer cust-1001 is using the Pro plan.
完整的序列看起来像这样:
sequenceDiagram
actor User
participant Runtime as Application / Agent Framework
participant Model as LLM Provider
participant Tool as Tool Implementation
User->>Runtime: What plan is cust-1001 using?
Runtime->>Model: Messages + get_customer schema
Model-->>Runtime: Tool request: get_customer(cust-1001)
Note over Runtime,Tool: The model requested an action.<br/>The function has not run yet.
Runtime->>Runtime: Validate arguments and permissions
Runtime->>Tool: Invoke get_customer("cust-1001")
Tool-->>Runtime: {"plan": "pro"}
Runtime->>Model: Tool-result message
Model-->>Runtime: Customer cust-1001 uses the Pro plan
Runtime-->>User: Final response
最重要的细节是缺少的直接连接:
LLM ─────X────→ local function
运行时介于模型和可执行代码之间。
一旦理解了数据流,基础循环看起来小得惊人。
以下是有意不依赖于特定提供商的伪代码:
async def run_agent(user_message: str) -> str:
messages = [
{
"role": "user",
"content": user_message,
}
]
for _ in range(10):
response = await call_model(
messages=messages,
tools=tool_schemas,
)
messages.append(response.message)
if not response.tool_calls:
return response.text
for call in response.tool_calls:
tool = tool_registry[call.name]
arguments = validate_arguments(
tool=tool,
arguments=call.arguments,
)
authorize_tool_call(
tool=tool,
arguments=arguments,
current_user=current_user,
)
result = await execute_tool(
tool=tool,
arguments=arguments,
)
messages.append(
{
"role": "tool",
"tool_call_id": call.id,
"content": serialize(result),
}
)
raise RuntimeError("Maximum agent steps exceeded")
这是许多 Agent 框架的核心。
这将应用消息和工具模式转换为提供商的请求格式。
它也将提供商响应转换为应用可以检查的对象。
这些调用是模型生成的输出。它们可能:
结构化输出不会自动成为可信的输出。
tool_registry[call.name]
运行时将生成的字符串解析为可执行的实现:
tool_registry = {
"get_customer": get_customer,
"search_orders": search_orders,
}
这是一个关键的安全边界。
模型只能请求应用暴露和解析的工具。
validate_arguments(...)
验证检查生成的参数是否与预期的结构匹配。
class GetCustomerArguments(BaseModel):
customer_id: str
这可以证明 customer_id 是一个字符串。
它不能证明当前用户是否被允许访问该客户。
authorize_tool_call(...)
授权是一个业务决策,不是模式决策。
if customer_id not in current_user.allowed_customers:
raise PermissionError("Customer access denied")
这是实际副作用发生的地方。
execute_tool(...)
执行者可能需要支持:
步骤限制
在没有停止护栏的情况下,模型可能会反复请求工具:
model → tool → model → tool → model → tool → ...
最大步骤限制是普通的防御性工程。
循环本身并不神奇。循环周围的工程才是框架赚钱的地方。
现在让我们超越概念性伪代码,检查一个真实的框架。
公开的起点是:
文档显示 PydanticAI 可以从 Python 函数的签名和文档字符串中推导工具模式。
当前样式的示例看起来像:
import asyncio
import os
from pydantic_ai import Agent
CUSTOMERS = {
"cust-1001": {
"id": "cust-1001",
"name": "Alice",
"plan": "pro",
}
}
agent = Agent(
os.environ["MODEL"],
instructions=(
"Answer questions about customer subscriptions. "
"Use get_customer when customer data is required."
),
)
@agent.tool_plain
async def get_customer(customer_id: str) -> dict[str, str]:
"""Retrieve a customer by customer ID."""
customer = CUSTOMERS.get(customer_id)
if customer is None:
return {
"id": customer_id,
"error": "Customer not found",
}
return customer
async def main() -> None:
result = await agent.run(
"What subscription plan is customer cust-1001 using?"
)
print(result.output)
if __name__ == "__main__":
asyncio.run(main())
公开 API 很便利,但它不能告诉我们谁调用了这个函数。
为此,我们需要追踪内部执行路径。
PydanticAI 步骤 1:模型响应进入图
在被检查的 _agent_graph.py 版本中,包含工具调用的模型响应被路由到 CallToolsNode。
相关的转换在逻辑上由以下代码表示:
return CallToolsNode[DepsT, NodeRunEndT](last_message)
这告诉我们一些重要的事情:
模型响应不能自己执行 Python 函数。
响应被传递给另一个运行时节点,该节点负责处理工具请求。
PydanticAI 步骤 2:验证和执行是分离的操作
在被检查的 _tool_execution.py 版本中,框架首先验证模型生成的调用:
validated = await self.tool_manager.validate_tool_call(call)
然后它执行验证后的调用:
tool_result = await self.tool_manager.execute_tool_call(validated)
这种职责分离清楚地证明了实际的边界:
模型生成的工具请求
→ 框架验证
→ 框架执行
模型并未执行其中任何一项操作。
PydanticAI 第 3 步:工具管理器调用可执行代码
在所检查版本的 tool_manager.py 中,经过验证的调用最终会被委托给已注册的工具集:
return await self.toolset.call_tool(
name,
validated.validated_args,
validated.ctx,
validated.tool,
)
这才是真正的执行交接点。
接收生成的工具请求。
验证其参数。
调用已注册的可执行代码。
至此,这部分调用路径已经没有任何谜团。
PydanticAI 第 4 步:结果转换为模型消息
执行完成后,_tool_execution.py 会创建一个 ToolReturnPart。
从概念上讲,该部分包含:
返回的内容。
随后,该结果会被包含在后续的模型请求中。
PydanticAI 文档也通过消息历史呈现了这一流程:
ToolCallPart
→ 实际执行工具
→ ToolReturnPart
→ 后续模型响应
这让框架的循环过程变得清晰可见。
PydanticAI 的执行路径
AI 智能体接收已注册的工具
↓
生成工具 schema
↓
提供商适配器将 schema 发送给模型
↓
模型返回 ToolCallPart
↓
AI 智能体图将响应路由到 CallToolsNode
↓
ToolManager 验证调用
↓
ToolManager 执行已注册的工具
↓
创建 ToolReturnPart
↓
通过另一个模型请求发送结果
↓
模型生成最终文本或另一个工具调用
值得关注的源码符号包括:
ModelRequestNode
CallToolsNode
ToolManager
ToolCallPart
ToolReturnPart
validate_tool_call
execute_tool_call
这些都是内部实现细节,而不是稳定的公共 API。
如果某个符号在未来版本中发生移动,应搜索代码仓库,而不要假定其路径永远不变:
rg "CallToolsNode" pydantic-ai
rg "validate_tool_call" pydantic-ai
rg "execute_tool_call" pydantic-ai
rg "ToolReturnPart" pydantic-ai
现在,让我们在当前的 LangChain AI 智能体技术栈中追踪同一个操作。
官方起点:
LangChain AI 智能体文档
LangChain 工具文档
当前 LangChain 中实现相同用例的示例如下:
import os
from langchain.agents import create_agent
from langchain.tools import tool
CUSTOMERS = {
"cust-1001": {
"id": "cust-1001",
"name": "Alice",
"plan": "pro",
}
}
@tool
def get_customer(customer_id: str) -> dict[str, str]:
"""Retrieve a customer by customer ID."""
customer = CUSTOMERS.get(customer_id)
if customer is None:
return {
"id": customer_id,
"error": "Customer not found",
}
return customer
agent = create_agent(
model=os.environ["MODEL"],
tools=[get_customer],
system_prompt=(
"Answer questions about customer subscriptions. "
"Use get_customer when customer data is required."
),
)
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": (
"What subscription plan is customer "
"cust-1001 using?"
),
}
]
}
)
print(result["messages"][-1].content)
同样,公共 API 隐藏了循环过程中的大部分细节。
因此,让我们继续追踪源码。
LangChain 第 1 步:create_agent 构建循环
在所检查版本的 LangChain agents/factory.py 中,create_agent 的文档直接说明了它的用途:
"""Creates an agent graph that calls tools in a loop until a stopping condition is met."""
它并不只是一个提示词模板。
该工厂围绕以下内容构建了基于图的编排:
其实现导入了 LangGraph 组件,包括 ToolNode,以及 ToolMessage 等消息类型。
LangChain 第 2 步:模型返回工具调用
模型不会执行工具。
它返回一条包含结构化工具调用数据的 AI 消息。
{
"name": "get_customer",
"args": {
"customer_id": "cust-1001"
},
"id": "call_123"
}
图路由会检查该消息。
如果其中存在工具调用,执行流程便会转移到工具节点。
如果不存在工具调用,图就可以停止执行并返回最终答案。
LangGraph 第 3 步:ToolNode 调用已注册的工具
决定性的证据出现在所检查版本的 LangGraph tool_node.py 中。
对于同步工具,其执行路径包含:
response = tool.invoke(call_args, config)
对于异步工具,其执行路径包含:
response = await tool.ainvoke(call_args, config)
这才是真正的函数执行边界。
模型生成了请求。
ToolNode 解析出已注册的工具,并通过常规运行时代码调用它。
源码使我们无须猜测究竟是谁执行了工具。
LangGraph 第 4 步:结果转换为 ToolMessage
调用完成后,ToolNode 会将结果规范化为 ToolMessage。
该消息包含如下信息:
返回的内容。
状态或错误信息。
消息会被追加到图状态中。
随后,图会返回模型节点,让模型读取工具结果,并决定是否:
请求另一个工具。
使用不同的参数重试。
继续执行工作流的另一个分支。
LangChain 和 LangGraph 的执行路径
Python 函数被转换为工具
↓
工具 schema 被绑定到聊天模型
↓
模型返回包含 tool_calls 的 AIMessage
↓
图路由检测到 tool_calls
↓
ToolNode 解析出所请求的工具
↓
ToolNode 调用 tool.invoke() 或 tool.ainvoke()
↓
结果被转换为 ToolMessage
↓
ToolMessage 被添加到图状态
↓
图返回模型节点
↓
没有剩余工具调用时停止执行