连接模型提供商

选择适配器

所有适配器都实现同一套协议，因此替换其中一个适配器不会影响代码的其他部分。

适配器	对接服务	你需要提供
OpenAIProvider	OpenAI Chat Completions 或 Responses	模型名称和 API 密钥
AnthropicProvider	Anthropic messages API	模型名称和 API 密钥
GoogleProvider	Google GenAI	模型名称和 API 密钥
DeepSeekProvider	DeepSeek Chat Completions 或 Responses	模型名称和 API 密钥
OllamaProvider	一个正在运行的 Ollama 服务器	已安装的模型（非本地部署时还需提供服务器 URL）

适配器负责统一消息、工具和事件，但不统一每个模型的能力。请选择与你的工作流相匹配的端点和模型。尤其是，图像、可见推理、托管工具和工具搜索并非在所有端点上可用。本运行时不提供通用的 response_format 或结构化最终输出设置；如果需要经过校验的数据，请使用带类型的工具，或在应用中校验最终答案。

配置你选定的模型提供商

从环境变量中读取凭据和模型名称。每个标签页都会创建一个可传给 create_agent 的 model；使用完毕后用 await model.aclose() 关闭它。这些代码片段配置的是真实服务，并非离线示例。

OpenAI

import os
from zett_agent.providers.openai import OpenAIProvider

model = OpenAIProvider(
    model=os.environ["OPENAI_MODEL"],
    api_key=os.environ["OPENAI_API_KEY"],
    response=True,
)

response=True 会选择 Responses API。省略它则使用 Chat Completions（默认）。请选择端点支持的协议；启用这个标志并不会给网关增加 Responses 支持。

Anthropic

import os
from zett_agent.providers.anthropic import AnthropicProvider

model = AnthropicProvider(
    model=os.environ["ANTHROPIC_MODEL"],
    api_key=os.environ["ANTHROPIC_API_KEY"],
)

使用 Messages API。不支持 response=True。

Google

import os
from zett_agent.providers.google import GoogleProvider

model = GoogleProvider(
    model=os.environ["GOOGLE_MODEL"],
    api_key=os.environ["GOOGLE_API_KEY"],
)

使用 API 密钥访问 Google GenAI。base_url 可以把它指向代理或本地测试服务器。内置构造函数不提供 Vertex AI 项目/位置配置。

DeepSeek

import os
from zett_agent.providers.deepseek import DeepSeekProvider

model = DeepSeekProvider(
    model=os.environ["DEEPSEEK_MODEL"],
    api_key=os.environ["DEEPSEEK_API_KEY"],
)

默认使用 Chat Completions。当你的服务支持时，response=True 会选择兼容 Responses 的端点。

Ollama

import os
from zett_agent.providers.ollama import OllamaProvider

model = OllamaProvider(
    model=os.environ["OLLAMA_MODEL"],
    base_url="http://127.0.0.1:11434",
)

请单独启动 Ollama 并安装所选模型。构造适配器不会启动服务器，也不会下载模型。本地服务无需 API 密钥；此适配器只使用 Ollama 的 chat 协议。

一次完整的工具调用请求

这个程序会发起真实的网络请求，可能产生费用。请在 shell 中导出环境变量，不要把密钥粘贴到源码或日志里。程序会注册一个加法工具、流式输出响应，并在请求失败时也关闭模型提供商。

"""Opt-in network example. Requires credentials and may incur provider charges."""

import asyncio
import os

from zett_agent.client import create_agent
from zett_agent.dispatcher import AgentEventDispatcher
from zett_agent.events import AgentEvent
from zett_agent.model import RetryOptions
from zett_agent.providers.openai import OpenAIProvider
from zett_agent.tools.base import tool


@tool(guidelines="Use for exact integer addition.")
def add(left: int, right: int) -> int:
    """Add two integers exactly."""
    return left + right


class ConsoleEvents(AgentEventDispatcher):
    async def on_text_delta_event(self, event: AgentEvent) -> None:
        print(event.delta, end="", flush=True)


async def main() -> None:
    model = OpenAIProvider(
        model=os.environ["OPENAI_MODEL"],
        api_key=os.environ["OPENAI_API_KEY"],
        retry=RetryOptions(base_delay=0.5, max_delay=8, max_retries=3),
    )
    try:
        client = await create_agent(model, tools=[add], event_dispatcher=ConsoleEvents())
        await client.run("Use the add tool to calculate 20 plus 22, then explain the result.")
        print()
    finally:
        await model.aclose()


if __name__ == "__main__":
    asyncio.run(main())

下载 real_provider.py。

export OPENAI_MODEL="your-tool-capable-model"
export OPENAI_API_KEY="your-api-key"
uv run python docs/_examples/real_provider.py

如果使用已安装的包，请下载该文件并运行 python real_provider.py。具体措辞以及模型是否遵循工具请求，取决于所选的模型。要确定性地验证相同的工具往返，请运行 流式展示一次工具往返。

连接兼容的网关

对于实现了现有协议的服务，请使用 base_url，而不必编写新的适配器：

model = OpenAIProvider(
    model=os.environ["GATEWAY_MODEL"],
    api_key=os.environ["GATEWAY_API_KEY"],
    base_url=os.environ["GATEWAY_BASE_URL"],
    send_prompt_cache_key=False,
)

URL 应指向 API 根路径，例如服务的 .../v1，而不是它的 .../chat/completions 路由。请让协议与网关匹配。有些网关会拒绝可选字段，比如 prompt_cache_key 或某些推理级别；请只配置受支持的选项，并测试你的组合。适配器不会从端点 URL 自动探测模型能力。

发送额外的请求头、查询参数或请求体字段

网关常常依据适配器没有建模的信息来路由、计费或区分版本。每个适配器都接受 extra_headers；OpenAI 风格的适配器和 Anthropic 适配器还接受 extra_query 和 extra_body，Google 接受 extra_body 和 base_url，Ollama 接受 extra_query。

model = OpenAIProvider(
    model=os.environ["GATEWAY_MODEL"],
    api_key=os.environ["GATEWAY_API_KEY"],
    base_url=os.environ["GATEWAY_BASE_URL"],
    extra_headers={"X-Tenant": "acme"},
    extra_query={"api-version": "2025-01-01"},
    extra_body={"service_tier": "flex"},
)

请求头和查询参数会附加在适配器自身的参数之后，用于每一个请求，因此这里给出的同名参数会替换 SDK 的默认值（可用于 User-Agent 或 anthropic-beta 标志）。extra_body 最后才合并进请求体：它可以添加适配器没有建模的选项，也可以覆盖适配器已有的选项，例如 DeepSeek 的 thinking。这些值会在构造提供商时被复制并校验，且 extra_body 必须是 JSON。

推理强度

在运行上设置与提供商无关的级别，而不是使用厂商专属选项：

from zett_agent.model import ReasoningEffort

reply = await client.run("Plan the migration.", reasoning_effort=ReasoningEffort.HIGH)

级别	含义
MINIMAL	在模型支持时请求最小的推理预算。
LOW / MEDIUM	较低的或中等的强度；MEDIUM 是运行时默认值。
HIGH / XHIGH	对于困难的多步骤任务请求更高的强度。
MAX / ULTRA	请求的最高级别；适配器可能将它们映射到同一个受支持的设置。
OFF	根据适配器的不同，省略或禁用显式的推理配置。

当前的 Responses 适配器会把 XHIGH、MAX 和 ULTRA 映射为 high。基于预算的适配器也会把最高几个级别归为一组。OpenAI Chat Completions 会原样转发具名级别，因此如果模型拒绝其中某个级别，仍可能抛出端点错误。运行时不会就支持的级别进行协商。

OFF 并不是零推理的通用保证：OpenAI 适配器会省略显式强度配置，而其他一些适配器则关闭思考。端点的默认设置仍可能进行推理。提高级别绝不保证出现可见的推理文本、更好的答案或特定的价格。可以用 create_agent(model, reasoning_effort=...) 为整个智能体设置默认值，并按请求覆盖它。

重试与代理

RetryOptions 只在第一个流式事件之前重试，因此答案进行到一半时发生的失败绝不会重放已经可见的文本或工具片段。

OpenAIProvider(
    model=os.environ["OPENAI_MODEL"],
    api_key=os.environ["OPENAI_API_KEY"],
    retry=RetryOptions(base_delay=0.5, max_delay=8, max_retries=3),
)

默认传输会遵循 HTTP_PROXY、HTTPS_PROXY、ALL_PROXY 和 NO_PROXY。当你需要自定义路由时，请传入自己的传输。

max_retries=3 最多允许四次传输尝试：一次初始请求加三次重试。认证和校验失败不会被盲目重试。该策略属于模型适配器，而不是 create_agent。如需整个请求的总截止时间，请参阅 集成到应用。静默的提供方由下一节的超时限制，而且超时不会重试。

当提供方不再响应

以前，如果提供方收下请求之后一直不回应，会被等待十分钟，然后再被请求三次；Google 和 Ollama 适配器则会永远等下去。现在每个适配器都会放弃，并说明是哪一段等待超时了。TimeoutOptions 为每个“不会有任何数据到达”的阶段各设置了一个限制：

限制项	默认值	限制的是什么
connect	5 秒	建立连接。可以访问的主机会在几毫秒内应答。
first_byte	60 秒	从发出请求到收到响应*头*。流式 API 会在模型产出任何内容之前就返回响应头，所以这里等了很久，说明有网关收下了请求却什么也没做。
idle	600 秒	响应开始之后，两个数据块之间允许的最长间隔。它设得很长，因为推理模型可能会安静地思考。

from zett_agent.model import TimeoutOptions

OpenAIProvider(
    model=os.environ["OPENAI_MODEL"],
    api_key=os.environ["OPENAI_API_KEY"],
    timeout=TimeoutOptions(first_byte=30, idle=900),
)

我们刻意不限制*总*时长：持续流出的长回答是正常的。把某个限制设为 None 即可关闭它。某个限制耗尽时会抛出 ProviderTimeoutError，其 phase 为 "connect"、"first_byte" 或 "idle"。

超时不会重试，连接失败除外。重试一个静默的提供方，要把整个限制时间再等一遍，而且第一次请求可能仍在对方那里运行并计费。如果响应已经开始后才安静下来，已经到达的内容会被送达，所以界面上已经显示了；此时请求会失败，而不是重新开始。

备注

Google 和 Ollama 的客户端要等到被索取第一个数据块时才发出请求，所以对它们来说，first_byte 涵盖连接、响应头和第一个数据块，连接本身没有单独的限制。Ollama 的默认值给第一个数据块留了十分钟，因为本地模型可能需要先从磁盘加载。

当提供方回答的是一个网页

在你和提供方之间，往往还有反向代理、网关、防火墙或强制门户。它出故障时，会返回 HTML 页面、一行纯文本或者什么都没有，而这类失败以前都会被简化成“Provider stream request failed”。现在，状态码、提供方自己的解释，以及它是一个网页这件事，都会写进错误信息：

openai returned HTTP 400: model [gpt-6-Astra] service info not found
openai answered HTTP 502 with a web page ("502 Bad Gateway") instead of an API
response. A proxy, gateway, firewall or login page in front of the API usually
does this; check the base_url and the network path to it.

失败会以 ProviderHTTPError 的形式到达，它带有 status_code、reason（当响应体是 JSON 错误时，提供方自己的话）、preview 和 is_html。被拒绝的凭据仍然是 ProviderAuthError，并附上同样的细节。SDK 自己的错误保留在 __cause__ 上。

对流式请求，只要回答的是网页或 JSON，无论状态码是什么，也都算错误。以前，带登录页的 200，或者因为网关不支持流式而返回的 JSON，会被当作一次没有文本的成功回复；现在它们会被抛出，并指出内容类型和响应体说了什么。其他任何内容类型都按流来读取，因为有些网关会把流以 text/plain 发送；一个流如果连一个事件都没有就结束，会作为错误抛出。

提前结束或携带损坏工具调用的流

过去有三种失败会让请求终止，现在模型提供商会让请求在这些失败中继续。

• 格式错误或被截断的工具参数。 如果工具调用的 JSON 无效，例如响应达到了输出上限或网关截断了流，这次调用就不会执行。它会变成一条失败的工具结果，说明发生了什么以及如何重试，这样模型就可以重新发送一个更小的调用。轻微格式错误但完整的 JSON 仍会被静默修复。被截断的载荷绝不会靠猜测补全，因为那样可能写入被截断的内容。AnthropicProvider(max_tokens=...) 设置每次响应的输出预算（默认 16384）；写入量大时可调高，模型上限较小时可调低。

• Responses 的 ``error`` 事件。 流中途出现的 error 事件（限流、过载）会带着自己的代码和消息抛出，而不会表现为缺少终止响应。

• 没有终止事件的 Responses 流。 如果网关在每个输出项都完成之后丢弃了 response.completed，那么已完成的项会被使用，finish_reason 为 "incomplete"。如果有某个项从未完成，或者什么都没到达，流就会抛出 ProviderResponseError；空流会被重试，因为此时还没有展示任何内容。

提示词缓存

运行时从会话派生出提示词缓存键，因此在支持该功能的端点上，同一对话的每一步都可以使用相同的路由提示。缓存复用和节省仍由提供商决定，而不是运行时的保证。在 AgentRunConfig 上设置 cache_key，可以把多个对话路由到同一个前缀；当端点拒绝该字段时，可以用 send_prompt_cache_key=False 构造提供商。

OpenAI 默认会发送该键。DeepSeek 默认省略它，因为其自有服务会自动缓存；只有网关需要时才应启用。其他适配器可能不提供缓存路由。

当提供商报告缓存命中时，ModelUsage.cache_read_tokens 和 cache_hit_rate 会将其暴露出来。经过协议转换网关的请求往往会省略明细；此时 ModelUsage.cache_reported 为 False，从而区分“未知”和实测的 0% 未命中。

始终释放适配器

客户端并不拥有你传入的提供商。应用结束时请关闭它，即使多个客户端共享同一个提供商：

await model.aclose()

对话中途更换模型

把 model=other_model 传给 run 或 stream，即可为单个请求覆盖默认值。普通消息和本地工具仍然与提供商无关，但视觉支持和托管工具必须与新端点匹配。带签名的推理/重放块只保留给能够消费它们的提供商/模型标识；不要把它们当作可移植的指令。

下一步：把流式输出接入终端或界面，或者按照 实现模型适配器编写你自己的适配器。