实现模型适配器#
遇到新协议，或者需要一个确定性的测试替身时，请使用自定义适配器。对于与 OpenAI 兼容的 HTTP 服务，先考虑用 OpenAIProvider 搭配自定义 base_url，可能并不需要新适配器。
协议#
AgentModel接收一个完整组装好的 ModelRequest，返回 ModelEvent 的异步迭代器。它带有一个由模型持有的 RetryOptions 策略。协议方法的声明是普通的 def，返回 AsyncIterator；实现可以用 async def加 yield来满足这个契约。
适配器只需要 retry 和 stream。另有一个可选属性 native_deferred_tools: bool，由 NativeDeferredToolsModel 声明。只有当你的端点既能把延迟加载的工具挡在模型上下文之外、又能在搜索结果点名时加载它（如 Anthropic 的 defer_loading），才把它设为 True：此时适配器会收到所有延迟加载的定义，并自行把它们标记为延迟加载。如果模型只能从搜索结果中得知延迟加载的工具（如 Responses API），就不要声明它；ToolSearchExtension 随后会把这些定义从请求中去掉。
from collections.abc import AsyncIterator
from zett_agent.messages import AssistantMessage
from zett_agent.model import (
ModelEvent,
ModelRequest,
ModelResponse,
RetryOptions,
)
class FixedModel:
retry = RetryOptions(max_retries=0)
async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
yield ModelEvent.text("Hello")
yield ModelEvent.completed(
ModelResponse(AssistantMessage(content="Hello"))
)
必须满足的流式约定#
产出零个或多个文本、推理或工具调用片段。
成功的流式传输必须以恰好一个已完成的 ModelResponse 结束。
这个响应里要包含完整的回答和完整的 ToolCalls，即使它们已经作为增量流式传出。运行时不从展示文本重建最终消息。
在最终响应之后不要再发出任何内容。
传输或协议失败时请抛出异常；不要把失败伪装成成功的空回答。
ModelRequest 包含这一步所需的完整上下文。请显式支持适用的消息角色和带类型的图像部分。AgentMessage 是内部输入，不是助手输出；内置适配器会把它映射成面向厂商 API 的用户输入。如果端点协议会缓存提示词前缀，请把cache_key声明给它；OpenAI 适配器在它的两种协议上都会把它作为 prompt_cache_key发送。
工具调用与用量#
用 ToolCallDelta 表示 ID、名称或参数的部分片段，用 ToolCall 表示最终解析出的参数。工具结果请按调用 ID 关联到对应的 ToolMessage，而不是按列表位置。用量在最终响应上通过 ModelUsage 报告；输入包含缓存读取和写入，输出在模型提供商报告时包含推理。缺失的计数器保持为零，但如果某个协议完全不报告任何缓存计数器，就必须传入 cache_reported=False，这样调用方才能区分未知的缓存用量和实测的未命中。
只对同一个提供商/模型标识重放签名的厂商块。请把它们保存在 AssistantMessage.replay_blocks 里，而不是拼接到文本中去。
重试策略由这里负责#
只在一个事件逃逸给调用方之前重试瞬时失败。在部分文本或工具增量之后重启会重复可见输出，还可能干扰工具调用的组装。校验失败或认证失败不应盲目重试。当消费方提前停止时，请在 finally 或上下文管理器里关闭 SDK 流。