Skip to content

输入封装与跨平台映射 ​

上一节区分了业务任务 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')

这里的 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、工具名、结果状态和来源版本。

练习 ​

  1. 为 Part 增加 tool_result 的 call_id,拒绝没有调用 ID 的工具结果。
  2. 编写一个 map_for_text_only_provider,遇到 image/file 时返回可观察的 unsupported_content,不要偷偷改成文本。
  3. 给 Request 增加 request_version,说明为什么重试时 task_id 不变而 call_id 要变化。

完成后继续上下文窗口与历史裁剪。

用心记录每一步成长