原生结构化输出与 PDF 引用
从工具意图转到数据契约
上一节的 tool_use 表示请求执行函数。订单抽取需要的是对象，不应为了取得 JSON 自动下单。Anthropic SDK 1.8.0 的 messages.parse(output_format=Order) 把 Pydantic 类型转换成请求 output_config.format，返回 ParsedMessage，其 parsed_output 是类型化对象。仍先检查终态，再检查业务范围；原生 schema 不能证明库存或身份。
沿用第一章环境，保存 native.py。SDK 1.8.0 此处使用 httpx2；MockTransport 是其真实 HTTP 测试传输，固定响应不联网。它验证 SDK 请求序列化和解析，不证明云端模型支持或遵循 schema。
import json
import httpx2 as httpx
from anthropic import Anthropic
from pydantic import BaseModel,ConfigDict
class Order(BaseModel):
model_config=ConfigDict(extra='forbid',strict=True)
sku:str
count:int
def checked(message):
if message.stop_reason!='end_turn' or message.parsed_output is None:
raise ValueError('not a completed structured result')
order=message.parsed_output
if order.sku not in {'A','B'} or not 1<=order.count<=5:raise ValueError('business violation')
return order
def fixture(reason,text):
def reply(request):
body=json.loads(request.content)
assert body['output_config']['format']['type']=='json_schema'
return httpx.Response(200,json={'id':'msg_fixture','type':'message','role':'assistant',
'model':'fixture','content':[{'type':'text','text':text}], 'stop_reason':reason,
'stop_sequence':None,'usage':{'input_tokens':1,'output_tokens':1}})
with Anthropic(api_key='fixture',http_client=httpx.Client(transport=httpx.MockTransport(reply))) as c:
return c.messages.parse(model='fixture',max_tokens=100,output_format=Order,
messages=[{'role':'user','content':'2 A'}])
if __name__=='__main__':
assert checked(fixture('end_turn','{"sku":"A","count":2}')).count==2
for reason,text in [('max_tokens','{"sku":"A","count":2}'),('refusal','I cannot'),('end_turn','{"sku":"A","count":9}'),('end_turn','{"sku":"A","count":true}'),('end_turn','{"sku":"A","count":"2"}'),('end_turn','{"sku":"A","count":2.0}')]:
try:checked(fixture(reason,text))
except ValueError as e: print(reason,type(e).__name__)
else:raise AssertionError('bad output accepted')此模型启用 strict=True：count 只接受 JSON 整数，拒绝 true、字符串 "2"，也拒绝 JSON 数值 2.0；这里保留输入形态，避免转换后再检查丢失类型证据。正常分支产生 Order(count=2)。完整 JSON 配 max_tokens 仍被拒绝；refusal 文本可能在 SDK 解析阶段已抛出 ValidationError，因此不能假设所有拒绝都返回可读取的 parsed_output。业务数量9即使 schema 合法也不能提交。协议层、解析层、业务层的失败需要分别记录，不统一伪装成空订单。
真实调用单独保存 online_native.py，需账户支持原生结构化的 ANTHROPIC_MODEL 和 ANTHROPIC_API_KEY，会产生费用，此入口未在线运行：
import os
from anthropic import Anthropic
from native import Order,checked
with Anthropic(timeout=30,max_retries=0) as c:
message=c.messages.parse(model=os.environ['ANTHROPIC_MODEL'],max_tokens=500,
messages=[{'role':'user','content':'要2本A'}],output_format=Order)
print(checked(message))旧beta示例可能给output_format传schema字典；此版本中类型交给parse的output_format，原始schema交给create的output_config.format。改参数位置不改变“完成→解析→业务核验”的顺序。
PDF 是 document，不是 image
用户提供的 PDF 先在服务端取得授权，再读 bytes。MIME 名称和 %PDF- 头仅作早期输入校验，不证明文件安全、完整或不存在恶意内容；文件大小、页数和请求限制以选用模型文档核对。保存 pdf.py，命令 python pdf.py authorized.pdf 是付费在线入口，本次未发送文件：
import base64,os,sys
from pathlib import Path
from anthropic import Anthropic
def document(raw,media_type='application/pdf'):
if media_type!='application/pdf' or not raw.startswith(b'%PDF-'):
raise ValueError('expected PDF bytes and media type')
return {'type':'document','title':'授权资料',
'source':{'type':'base64','media_type':media_type,
'data':base64.b64encode(raw).decode('ascii')},
'citations':{'enabled':True}}
if __name__=='__main__':
raw=Path(sys.argv[1]).read_bytes()
with Anthropic(timeout=60,max_retries=0) as c:
r=c.messages.create(model=os.environ['ANTHROPIC_MODEL'],max_tokens=700,
messages=[{'role':'user','content':[document(raw),
{'type':'text','text':'仅根据资料概括退款条件并给引用。'}]}])
if r.stop_reason!='end_turn':raise RuntimeError('document answer incomplete')
for b in r.content:
if b.type!='text':continue
print(b.text)
for citation in b.citations or []:
print(citation.model_dump())PDF引文的page_location使用document_index与页范围，其他文本来源可能使用字符或内容块位置；不要把所有引用都假定为网页URL。展示时将document_index映射回本次授权文档，并保留引用文本供人工核验。引用证明模型指向了某个片段，不自动证明该片段支持结论。开启引用与结构化输出的组合支持有模型限制，两个实验分别运行，不能随意混合配置。
练习
离线使用native.py新增“额外字段price”和“sku未知”响应，确认均不提交订单；用pdf.py的document函数分别输入正确头PDF、PNG伪装PDF、正确PDF配image/png，检查后两者拒绝。给出为什么头校验不能替代真实PDF解析的解释。