输入封装与跨平台映射
上一节区分了业务任务 ID、模型调用 ID 和工具调用 ID。本节继续处理一个容易被隐藏的边界:应用内部的输入对象,不能直接等同于任意供应商的消息数组。先把应用语义固定下来,再把它映射到具体 API;否则一个平台没有的字段会被静默丢弃,问题只在生产请求中出现。
先定义内部输入合同
内部合同保留四类信息:system 约束、按顺序排列的对话轮次、带 MIME 和来源的内容块、调用级元数据。调用级元数据用于超时、租户和追踪,不应混进用户可见文本。role 决定消息责任,但不能把用户文本中的“system:”当成系统消息。
python
from __future__ import annotations
from dataclasses import dataclass
from typing import Literal
Role = Literal['system', 'user', 'assistant', 'tool']
@dataclass(frozen=True)
class Part:
type: Literal['text', 'image', 'file', 'tool_result']
value: str
mime: str | None = None
source_id: str | None = None
@dataclass(frozen=True)
class Message:
role: Role
parts: tuple[Part, ...]
@dataclass(frozen=True)
class Request:
task_id: str
call_id: str
messages: tuple[Message, ...]
tenant: str
timeout_ms: int = 30_000
def validate(request: Request) -> None:
if not request.task_id or not request.call_id or not request.tenant:
raise ValueError('identity is required')
if request.timeout_ms <= 0:
raise ValueError('timeout must be positive')
if not request.messages or request.messages[0].role != 'system':
raise ValueError('system policy must be explicit')
for message in request.messages:
if not message.parts:
raise ValueError('empty message')
for part in message.parts:
if part.type in {'image', 'file'} and not part.mime:
raise ValueError('media needs MIME type')
request = Request(
task_id='ticket-7', call_id='call-1', tenant='tenant-a',
messages=(
Message('system', (Part('text', 'Answer with evidence.'),)),
Message('user', (Part('text', 'Summarize the report.'),)),
),
)
validate(request)
try:
validate(Request('ticket-7', 'call-2', (), 'tenant-a'))
except ValueError as error:
assert str(error) == 'system policy must be explicit'
else:
raise AssertionError('missing system policy accepted')
print('internal request contract passed')1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
这里的 Request 不是某家 SDK 的替代品,而是应用边界。真实适配器可以把 Part('text', ...) 映射到文本字段,把 Part('image', ..., mime='image/png') 映射到图片内容块;如果目标 API 不支持文件或工具结果,适配器应返回明确的能力错误,不应把二进制转成一段看似成功的字符串。
映射时保留不可逆差异
OpenAI Responses 的输入 item、Claude 的 message content block、Gemini 的 parts 都能表达文本,但角色、工具结果、缓存和多模态字段并不完全同构。适配器至少要记录目标平台、请求版本、能力矩阵版本和被拒绝的扩展字段。不能因为三个平台都有 content 就声称行为等价。
变化实验:把工具结果当用户文本
将 tool_result 改成 user 的文本后,结构校验可能仍然通过,但调用关联和权限语义已经丢失。用 source_id 指向业务账本时,后续审计无法判断这个结果是模型陈述还是服务端实际返回。因此工具结果必须保留调用 ID、工具名、结果状态和来源版本。
练习
- 为
Part增加tool_result的call_id,拒绝没有调用 ID 的工具结果。 - 编写一个
map_for_text_only_provider,遇到 image/file 时返回可观察的unsupported_content,不要偷偷改成文本。 - 给
Request增加request_version,说明为什么重试时task_id不变而call_id要变化。
完成后继续上下文窗口与历史裁剪。