上一篇，我们给 get_inventory() 写了 Tool Schema，也验证了 API 能接收它、模型能理解它。但那时模型只看到了工具的“说明书”：程序没有执行函数，也没有把查询结果交还给模型。
这一篇要第一次走完工具调用的全过程：
模型看到 Schema → 提出调用请求 → 程序执行函数 → 程序送回结果 → 模型根据结果回答用户。
这还不是 Agent Loop，也就是让模型反复判断是否还要调用工具的循环。本篇只处理“最多一次工具调用”：模型要么直接回答，要么请求 get_inventory，程序执行后再让模型生成最终回答。连续调用多个工具，是后面才会解决的问题。
文章对应代码仓库 agent-from-zero 的 v0.1.2-single-tool-call。可以切换到这个 tag，对照 src/agent/agent.py、src/agent/llm.py 与 tests/test_agent.py 阅读。
先复用已有的请求和统计代码
之前的 chat() 同时负责发送请求和生成 ChatResult。加入工具调用后，程序会用另一套流程请求模型。如果直接复制整段代码，异常处理、延迟统计和成本估算就会出现两份实现。这里先把这两部分拆开：
def _complete(messages, temperature=0.0, tools=None):
started_at = time.perf_counter()
try:
kwargs = dict(model=DEFAULT_MODEL, temperature=temperature, messages=messages)
if tools:
kwargs["tools"] = tools
response = get_client().chat.completions.create(**kwargs)
except AuthenticationError as e:
raise AuthenticationFailure(str(e)) from e
except APIConnectionError as e:
raise NetworkFailure(str(e)) from e
except BadRequestError as e:
raise BadRequestFailure(str(e)) from e
latency_ms = (time.perf_counter() - started_at) * 1000
return response, latency_ms
def _to_chat_result(response, latency_ms) -> ChatResult:
usage = response.usage
cache_hit_tokens = usage.prompt_cache_hit_tokens or 0
cache_miss_tokens = usage.prompt_cache_miss_tokens or 0
tokens_out = usage.completion_tokens
return ChatResult(
text=response.choices[0].message.content,
tokens_in=usage.prompt_tokens,
tokens_out=tokens_out,
cache_hit_tokens=cache_hit_tokens,
cache_miss_tokens=cache_miss_tokens,
latency_ms=latency_ms,
estimated_cost_usd=_estimate_cost_usd(cache_hit_tokens, cache_miss_tokens, tokens_out),
)_complete() 负责请求模型并测量延迟，_to_chat_result() 负责提取文本、Token 和估算成本。原来的 chat() 只需组合它们，对外行为不变：
def chat(user_message, system=DEFAULT_SYSTEM, temperature=0.0) -> ChatResult:
messages = [
{"role": "system", "content": system},
{"role": "user", "content": user_message},
]
response, latency_ms = _complete(messages, temperature=temperature)
return _to_chat_result(response, latency_ms)这样，工具调用就能直接复用前面写好的异常处理和统计代码。
用 `ask()` 完成一次工具调用
新的 ask() 函数写在 src/agent/agent.py：
TOOLS = [GET_INVENTORY_SCHEMA]
TOOL_FUNCTIONS = {"get_inventory": get_inventory}
def ask(user_message: str, system: str = llm.DEFAULT_SYSTEM) -> ChatResult:
messages = [
{"role": "system", "content": system},
{"role": "user", "content": user_message},
]
response, latency_ms = llm._complete(messages, tools=TOOLS)
first_result = llm._to_chat_result(response, latency_ms)
message = response.choices[0].message
if not message.tool_calls:
return first_result
tool_call = message.tool_calls[0]
args = json.loads(tool_call.function.arguments)
tool_result = TOOL_FUNCTIONS[tool_call.function.name](**args)
messages.append({
"role": "assistant",
"content": message.content,
"tool_calls": [{
"id": tool_call.id,
"type": "function",
"function": {
"name": tool_call.function.name,
"arguments": tool_call.function.arguments,
},
}],
})
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(tool_result),
})
response, latency_ms = llm._complete(messages)
second_result = llm._to_chat_result(response, latency_ms)
return _sum_chat_results(first_result, second_result)顺着代码读，ask() 只有两个分支：
- —没有
tool_calls：模型已经直接回答，返回第一次请求的结果。 - —有
tool_calls：程序解析参数、执行函数，把调用消息和工具结果加入对话记录，再请求一次模型生成最终回答。
三条消息怎样对上号
这一段协议最容易混淆的不是函数调用本身，而是三份数据之间的关系：
需要留意以下细节：
- 1.
tool_call.function.arguments是 JSON 字符串，因此要先用json.loads()转成 Python 字典。 - 2.第二次请求要带回模型生成的 assistant 消息，其中保留
tool_calls[].id、函数名和原始参数。 - 3.工具结果是一条
role: "tool"消息。它的tool_call_id必须与调用请求的id相同。这一版把工具结果作为文本放进content，因此用json.dumps()把 Python 字典转换成 JSON 字符串。这个转换过程也叫“序列化”。
本篇手动重建了纯字典形式的 assistant 消息，而没有把 SDK 对象直接放回 messages。这样对话记录可以直接转换成 JSON。以后无论要保存完整的调用记录，还是保存程序运行到哪一步，都不需要依赖某个 SDK 对象的内部结构。
第二次 _complete(messages) 没有再传 tools。这是本篇为了“最多调用一次工具”有意做出的限制，并不是 Tool Calling 协议要求这样写。程序预期第二次请求直接生成最终回答，也没有准备处理新的工具调用。真正的 Agent Loop 会在后续请求中继续提供工具，并检查模型是否还要调用。
模型提出调用，程序负责执行
“Tool Calling”这个名字容易让人以为模型会直接运行 get_inventory()。在本篇采用的自定义函数调用流程里，模型只生成调用请求，应用程序负责执行函数：
tool_result = TOOL_FUNCTIONS[tool_call.function.name](**args)
模型根据 Schema 给出函数名与参数；程序收到请求后，决定是否接受、如何校验、调用哪个本地函数，并拿到真实结果。DeepSeek 的官方示例也采用相同分工。
这样的分工非常重要。模型生成的参数只是一个建议，不能直接当作已经确认无误的指令。今天的工具只是读取内存中的库存；以后接入订单、支付或数据库时，程序必须先检查权限和参数，并记录执行过程，然后才能操作这些系统。
查看实际发送的两次请求
给 HTTP 客户端加一个函数，在每次请求发出时把内容记录下来，就能看到 ask("A100 还有多少库存？") 实际发送了什么：
captured = []
def log_request(request):
captured.append(json.loads(request.content))
llm._client = OpenAI(
api_key=os.environ["DEEPSEEK_API_KEY"],
base_url=llm.DEEPSEEK_BASE_URL,
http_client=httpx2.Client(event_hooks={"request": [log_request]}),
)
result = ask("A100 还有多少库存？")第一次请求带上用户问题和 Tool Schema（2026-09-20）：
{
"messages": [
{"role": "system", "content": "You are a helpful assistant for a small industrial equipment company."},
{"role": "user", "content": "A100 还有多少库存？"}
],
"model": "deepseek-chat",
"temperature": 0.0,
"tools": [{
"type": "function",
"function": {
"name": "get_inventory",
"description": "Look up the current stock level and unit price for a product, given its product ID. Use this whenever the user asks about stock, availability, or price for a specific product.",
"parameters": {
"type": "object",
"properties": {
"product_id": {
"type": "string",
"description": "The product ID, e.g. 'A100' or 'B200'."
}
},
"required": ["product_id"]
}
}
}]
}模型没有直接回答库存，而是返回一个调用请求：
{
"choices": [{
"finish_reason": "tool_calls",
"message": {
"content": "",
"role": "assistant",
"tool_calls": [{
"id": "call_00_bej7uwqVwn3nm4gNtEQg4078",
"function": {
"name": "get_inventory",
"arguments": "{\"product_id\": \"A100\"}"
},
"type": "function"
}]
}
}],
"model": "deepseek-flash",
"usage": {
"completion_tokens": 40,
"prompt_tokens": 335,
"prompt_cache_hit_tokens": 128,
"prompt_cache_miss_tokens": 207
}
}程序执行 get_inventory("A100") 后，第二次请求会多出 assistant 调用消息和工具结果。
先说明一个记录上的细节：下面的第一次响应和第二次请求摘自两次独立运行，所以生成的 call_... 不同。这两段只用于展示消息结构，不能拼成同一次完整过程。在同一次真实运行中，第二次请求里的 tool_calls[].id 和 tool_call_id 必须使用第一次响应返回的同一个值。
{
"messages": [
{"role": "system", "content": "You are a helpful assistant for a small industrial equipment company."},
{"role": "user", "content": "A100 还有多少库存？"},
{
"role": "assistant",
"content": "",
"tool_calls": [{
"id": "call_00_luvHOu8Y1L4ze41twVA08026",
"type": "function",
"function": {
"name": "get_inventory",
"arguments": "{\"product_id\": \"A100\"}"
}
}]
},
{
"role": "tool",
"tool_call_id": "call_00_luvHOu8Y1L4ze41twVA08026",
"content": "{\"name\": \"Industrial Sensor A100\", \"price\": 100, \"stock\": 20}"
}
],
"model": "deepseek-chat",
"temperature": 0.0
}从这些实际请求和响应中还可以看到：
- —第一次响应的
content是空字符串，真正有意义的内容在tool_calls中。 - —
role: "tool"的content是本地函数真实返回值序列化后的字符串，不是模型生成的库存数据。 - —这次请求使用
deepseek-chat，响应中记录的模型名称是deepseek-flash；335 个输入 Token 中有 128 个命中缓存。这只是这一次实验看到的情况，不代表服务端以后每次都会选择同一个模型，也不代表每次都会命中相同数量的缓存。
一个问题用工具，一个问题不用
现在用两类问题观察模型是否会采用不同的处理方式：
r1 = ask("A100 还有多少库存？")
print(r1.text)
r2 = ask("What is 2+2?")
print(r2.text)真实调用结果（deepseek-chat，2026-09-19）：
--- "A100 还有多少库存？" ---
A100（工业传感器 A100）目前还有 20 件库存。
tokens_in=431 tokens_out=57 latency_ms=1668 cost_usd=0.000099
--- "What is 2+2?" ---
2 + 2 = 4.
（只发生了一次请求）第一个问题调用了工具，其中“20”来自 products["A100"]["stock"]；第二个问题由模型直接回答，只发生了一次请求。这次结果与 Schema 中“在用户询问库存、可用性或价格时使用”的描述一致。不过，一次实验不能证明模型以后每次都会做出同样选择，还需要重复实验和自动化测试来继续验证。
汇总两次请求的成本与延迟
触发工具后，一次用户问答包含两次模型请求。用户只看到最终文本，但模型侧的 Token、费用和请求时间来自两次请求：
def _sum_chat_results(first: ChatResult, second: ChatResult) -> ChatResult:
return ChatResult(
text=second.text,
tokens_in=first.tokens_in + second.tokens_in,
tokens_out=first.tokens_out + second.tokens_out,
cache_hit_tokens=first.cache_hit_tokens + second.cache_hit_tokens,
cache_miss_tokens=first.cache_miss_tokens + second.cache_miss_tokens,
latency_ms=first.latency_ms + second.latency_ms,
estimated_cost_usd=first.estimated_cost_usd + second.estimated_cost_usd,
)如果只记录第二次请求，模型调用的成本和延迟都会被低估。这是第三篇介绍的统计方法第一次用在包含多次请求的问答中。
这里也要看清统计边界：latency_ms 累加的是两次模型请求的时间，不包含 get_inventory() 的执行时间；estimated_cost_usd 估算的也是模型费用，不包含外部数据库或第三方服务可能产生的费用。当前工具只是读取内存字典，差别很小；接入真实系统后，工具自身的耗时和费用也需要单独记录。
测试程序流程，不要固定模型的回答
pytest -v当时的实际结果：
tests/test_agent.py::test_ask_returns_directly_when_model_does_not_call_a_tool PASSED
tests/test_agent.py::test_ask_sends_the_tool_schema_on_the_first_call PASSED
tests/test_agent.py::test_ask_calls_get_inventory_and_returns_the_real_stock_number PASSED
tests/test_agent.py::test_ask_feeds_the_real_tool_result_back_to_the_model PASSED
tests/test_agent.py::test_ask_sums_cost_and_latency_across_both_calls PASSED
...（加上此前的 17 个）
22 passed in 0.44s这些测试检查的是可以稳定验证的程序行为：是否发送 Schema、是否发起两次请求、tool_call_id 是否正确对应、程序是否送回真实结果，以及是否汇总两次请求的数据。
模型回答“A100 还有 20 件库存”，也可能回答“当前库存是 20 件”。两句话表达的事实相同，只是说法不同。因此，测试不应该要求模型每次说出完全相同的一句话，也不应该把某次联网调用的完整回答写死在测试代码里。
留给你的三个实验
- 1.把 Tool Schema 的
description临时改成含糊的"Get inventory."，重复运行库存问题和What is 2+2?，记录工具选择是否变化。不要只跑一次就下结论。 - 2.设法让模型把商品名而不是商品 ID 填入
product_id，观察get_inventory()返回None后模型怎样回应。这里只观察，下一篇再处理缺失数据与幻觉。 - 3.输入“A100 和 B200 各有多少库存？”，观察只读取
message.tool_calls[0]的实现会丢掉什么。再思考为什么真正的 Agent 需要循环。
小结
本篇第一次走完了工具调用的全过程：模型根据 Schema 提出请求，程序解析参数并执行函数，再用关联 ID 把真实结果交给模型。这个过程有两个容易忽略的事实：
- —在自定义函数调用中，模型只提出调用；是否执行、如何执行，都由程序决定。
- —一次工具问答可能包含多次模型请求，模型调用的成本、Token 和延迟必须合并统计；工具自身的消耗需要另行记录。
现在 ask("A100 还有多少库存？") 已经能根据真实数据回答问题，但程序仍有明显缺口：参数错误或商品不存在时，工具可能返回 None。下一篇会专门制造这种情况，观察模型会诚实说明“没有数据”，还是生成一个没有依据的答案。