Tool Use、流式与结构化契约

一次工具往返的四个对象

工具声明告诉模型可以请求什么；assistant 的 tool_use 说明模型请求了什么；本地执行产生业务结果；user 的 tool_result 带回结果并用 tool_use_id 关联。完整保留原 assistant 消息，随后紧接工具结果，才能维持匹配关系。

沿用上章环境，保存为 tool.py。这是只读库存实验；未知工具或非法参数立即失败，不动态执行任意函数名。

import os,json
from anthropic import Anthropic
c=Anthropic(timeout=30,max_retries=0)
messages=[{'role':'user','content':'查询 A 的库存'}]
tools=[{'name':'stock','description':'读取指定商品库存',
        'input_schema':{'type':'object','properties':{'sku':{'type':'string'}},
                        'required':['sku'],'additionalProperties':False}}]
for _ in range(4):
    r=c.messages.create(model=os.environ['ANTHROPIC_MODEL'],max_tokens=500,
                         tools=tools,messages=messages)
    calls=[b for b in r.content if b.type=='tool_use']
    if r.stop_reason=='max_tokens':
        raise RuntimeError('truncated')
    if not calls:
        if r.stop_reason!='end_turn':
            raise RuntimeError(f'unhandled stop reason: {r.stop_reason}')
        print([b.text for b in r.content if b.type=='text'])
        break
    messages.append({'role':'assistant','content':[b.model_dump() for b in r.content]})
    results=[]
    for b in calls:
        if b.name!='stock' or set(b.input)!={'sku'} or not isinstance(b.input['sku'],str):
            raise ValueError('invalid tool request')
        result={'sku':b.input['sku'],'count':{'A':3}.get(b.input['sku'])}
        results.append({'type':'tool_result','tool_use_id':b.id,
                        'content':json.dumps(result)})
    messages.append({'role':'user','content':results})
else:
    raise RuntimeError('step limit')

并行工具请求必须每个都返回关联结果，不能只处理 content[0]。业务失败可以返回带 is_error=True 的 tool_result，让模型解释或换查询；鉴权失败不应转为成功空数据。写工具的授权与幂等逻辑位于函数执行之前，不能依赖模型自报的“已批准”。

流式参数不能边到边执行

SDK 的 messages.stream 累积事件；文本可从 text_stream 显示，最终对象用 get_final_message 读取。保存为 stream.py：

import os
from anthropic import Anthropic
with Anthropic(timeout=30,max_retries=0).messages.stream(
    model=os.environ['ANTHROPIC_MODEL'],max_tokens=200,
    messages=[{'role':'user','content':'解释库存和订单的区别'}]) as stream:
    for text in stream.text_stream:
        print(text,end='',flush=True)
    final=stream.get_final_message()
print('\nstop:',final.stop_reason)

工具输入流可能是 {"sku": 后接 "A"}。第一个片段不是无效最终请求，只是尚未完成；完成后才解析与校验。不能通过补右括号“修复”半条命令后执行。

结构化抽取与工具调用的目的不同：前者要求一份业务数据，后者提出执行意图。支持原生结构化输出的模型可使用对应 output configuration / SDK parse 能力；不支持时可设计返回抽取结果的只读工具，再对其 input 做 schema 校验。无论哪种，仍需检查拒绝、截断与业务不变量，不能宣称 Tool Use 本身保证数据真实。

参阅 工具实现 与 流式消息。

练习

离线检查两条 tool_use（id 分别 a、b）与结果列表匹配。构造只返回 a 的错误结果，断言校验失败，再补齐 b。

先记录自己的输出与解释，再核对本章答案。