渠道适配器

每个渠道（provider，Web UI 里叫“渠道”）是 hoshimi/core/providers/ 里的一个 Provider 子类：它持有该渠道的账号列表和选项，知道怎样用一个账号向上游发出一轮对话，并把上游的流转换成归一化分片。账号轮换、重试、冷却、模型目录缓存这些共同的部分都在基类里，适配器只写这个站点特有的东西。

渠道的几种形式复制锚点链接

不需要任何账号就能用的渠道把类属性 keyless 设为 True：只要配置没有禁用它，build_providers 就用一个匿名占位账号把它建出来。

Provider 契约复制锚点链接

契约在 hoshimi/core/providers/base.py。

类属性复制锚点链接

max_retries（一轮最多试几个账号）由 server.max_account_retries 设置，默认 3。

方法复制锚点链接

Account 包装一个账号的字典：account.get(key)、account.set(key, value)，account.id 是配置里的 id（没有时为 <provider>-<序号>），account.anonymous 标记匿名占位账号，account.needs_login 为真时渠道池不再使用它，直到重新登录。

归一化分片协议复制锚点链接

stream_raw 产出的每个分片都是带 type 键的字典。不管上游说的是 OpenAI Chat、Anthropic Messages、Gemini 还是 TRAE SOLO，适配器都要转成下面这些：

一轮对话的典型分片序列：

{"type": "reasoning_delta", "text": "先看看文件……"}
{"type": "text_delta", "text": "好的，"}
{"type": "tool_call_start", "id": "call_1", "name": "read", "arguments": "", "index": 0}
{"type": "tool_call_delta", "index": 0, "arguments": "{\"path\": \"README.md\"}"}
{"type": "usage", "usage": {"prompt_tokens": 120, "completion_tokens": 30, "total_tokens": 150}}
{"type": "finish", "reason": "tool_calls"}

输出端由三组函数把分片变成公开协议：chunks.py 的 emit_chat_chunks / aggregate_chat（Chat Completions）与 emit_responses_events / aggregate_responses（Responses），以及 message_chunks.py 的 emit_messages_events / aggregate_messages（Anthropic Messages）。几条规则：

• finish.reason 不是 stop、tool_calls、length 时按 stop 处理；上游没有发出 finish 时，输出端补一个 stop。
• error 分片在流式输出中变成一个错误帧；非流式时变成 UpstreamStreamFailure，再作为 upstream_error（502）返回。
• 只要已经产出过分片，渠道池就不再换账号重试，否则客户端会收到重复的内容。

OpenAI 兼容的站点不必自己解析流：providers/openai_compat.py 的 consume_openai_sse(lines, provider=...) 把 OpenAI 风格的 SSE 转成上面的分片；read_body 还能处理对流式请求返回整个 JSON 的上游。其他协议的转换在 core/translate/（anthropic.py、gemini.py、solo.py、responses.py）。

错误与账号轮换复制锚点链接

适配器用 ProviderError 的子类报告上游的拒绝，渠道池按类型决定下一步：

所有账号都试过、而最后一个错误是“过一会儿可能就好”的那种时，渠道池最多再做 backoff_retries 轮：等待上游给出的时间（最多 backoff_max_seconds），没有给出时随机等待（full jitter）。一旦已经产出过内容，就不再重试。

openai_compat.py 的 raise_for_status(response, provider=...) 按固定顺序把一个 HTTP 回答归类：先看是否是“模型不存在”（404/410 等），再依次判断模型过载、安全拦截、401/403 认证失败、403 权限不足、额度（402/429 或正文里的额度字样）。顺序很重要：过载的模型常以普通的 429 出现，安全拦截常以普通的 403 出现，只看状态码就会把它们误判为额度用尽或凭据错误。流中以 200 返回的错误帧，只有“模型不存在”“过载”“拦截”“额度”几类会变成异常，其余作为 error 分片交给调用方。

失效模型记录（failure.py 的 DeadModelStore）只在内存中：一次记录在 10 分钟后被遗忘；同一窗口内被记三次，记录就不再随时间解除。这个模型再次成功产出第一个分片时，记录被清除。

stream_using 的 on_route(provider_id, account_id, attempt) 回调在某个账号产出第一个分片时触发，用量统计据此记录真正回答的账号。

模型目录与能力复制锚点链接

Provider.models() 管理目录缓存，适配器只实现 fetch_models()：

• 成功的目录缓存 300 秒。
• 取目录失败时保留上一次的结果，30 秒后再试；从来没有成功过时使用 fallback_models()。
• 新目录不到上一次的一半时，视为不完整的回答，保留旧目录一个周期；下一个周期还是这样，才接受它。
• fetch_models() 成功取到目录时设置 catalog_authoritative = True（OpenAiCompatProvider 已经这样做），这样站点返回的空目录被当作真实答案，而不是回退到固定列表。

specs() 返回每个模型的 ModelSpec（hoshimi/core/catalog.py），它是推理强度策略和 GET /api/v1/models 读取的东西：

各站点的能力表放在 catalog.py（zhipu_spec、kilo_spec、cline_spec 等）。默认的 specs() 给每个模型一个惰性的规格（不推理，所以推理强度策略不改动请求体），再从 models.dev 补充“是否推理”和“是否接受图片”这两项；能力表已经写明的不会被覆盖。推理强度的行为见模型行为。

写一个专用适配器复制锚点链接

以智谱 GLM 渠道（providers/zhipu.py）为例，它是一个 OpenAI 兼容站点：

class ZhipuProvider(OpenAiCompatProvider):
    id = "zhipu"
    display_name = "Zhipu GLM (BigModel)"
    keyless = False
    probeable = False
    requires_authorization = True
    base_url = ZHIPU_BASE_URL
    chat_path = "/chat/completions"
    models_path: str | None = "/models"
    fallback_model_list: ClassVar[list[str]] = list(ZHIPU_FREE_MODELS)

    def build_headers(self, account: Account, model: str) -> dict[str, str]:
        headers = super().build_headers(account, model)
        headers["Authorization"] = f"Bearer {self._token(account)}"
        return headers

    def specs(self) -> dict[str, ModelSpec]:
        return {model_id: zhipu_spec(model_id, self.id) for model_id in self.models()}

def test_a_turn_goes_to_the_station_with_the_bearer() -> None:
    seen: dict[str, str] = {}

    def handler(request: httpx.Request) -> httpx.Response:
        seen["path"] = request.url.path
        seen["authorization"] = request.headers.get("authorization", "")
        return httpx.Response(200, text=SSE)

    provider = ZhipuProvider(httpx.Client(transport=httpx.MockTransport(handler)), [_account()])
    chunks = list(provider.stream_raw(provider.accounts[0], "glm-4.7-flash", {"messages": [{"role": "user", "content": "x"}]}))
    assert seen["path"] == "/api/paas/v4/chat/completions"
    assert seen["authorization"] == "Bearer sk-zhipu-1"
    assert [chunk["type"] for chunk in chunks] == ["text_delta", "text_delta", "usage", "finish"]

添加一个按数据定义的服务复制锚点链接

只需 API 密钥、协议常规的服务不必写类：在 core/provider_definitions.py 的 CANONICAL_PROVIDERS 中加一行即可，例如：

ProviderDefinition('groq', 'Groq', 'https://api.groq.com/openai/v1', protocol='chat', auth=('api_key',), plan='api', catalog='/models', adapter='configured', key_env='GROQ_API_KEY'),

configured 的服务由 regular_provider_classes() 生成一个 ConfiguredProvider 子类；provider_forms.py 自动为支持 api_key 的服务提供“API 密钥 + 地址 + 模型”的表单。目录接口特殊的服务在 providers/catalog_profiles.py 描述，请求体需要调整的在 providers/request_profiles.py 描述。tests/core/test_provider_definitions.py 检查同一品牌的不同服务不会因为名字或地址相同而合并。

交互式登录复制锚点链接

需要浏览器登录或设备码的站点，在 core/login/ 里写一个登录流程。命令行的 hoshimi login <id> 和 Web UI 的登录对话框共用它。

契约复制锚点链接

LoginFlow = Callable[[httpx.Client, ReportUrl], dict[str, Any]]

流程拿到一个 HTTP 客户端和一个报告函数 report。站点一给出浏览器地址，流程就必须调用 report(url)：命令行打印它并打开浏览器，API 把它交给 Web UI，两者都不必知道站点用的是设备码、外部登录链接还是本地回调端口。然后流程阻塞，直到凭据就绪，返回渠道认识的账号字典；失败时抛出 LoginError，消息会直接显示给用户。

report 是一个 LoginReporter，还提供：

LOGIN_TIMEOUT_SECONDS 是 600 秒；管理器给整个流程多留 60 秒，让流程自己的超时先触发。

例子：Kilo 的设备码登录复制锚点链接

def start_kilo_login(client: httpx.Client, report: Callable[[str], None]) -> dict[str, Any]:
    response = client.post(DEVICE_URL, timeout=30)
    if response.is_error:
        raise LoginError(f"kilo: device authorization failed (HTTP {response.status_code})")
    data = response.json()
    code, url = data.get("code"), data.get("verificationUrl")
    # ...拒绝不完整的响应
    lifetime = min(float(data.get("expiresIn") or LOGIN_TIMEOUT_SECONDS), LOGIN_TIMEOUT_SECONDS)
    if isinstance(report, LoginReporter):
        report.device_code(url, code, lifetime)
    else:
        report(url)
    deadline = time.monotonic() + lifetime
    while time.monotonic() < deadline:
        if isinstance(report, LoginReporter):
            report.wait(3)
        else:
            poll_pause(3)
        polled = client.get(f"{DEVICE_URL}/{code}", timeout=30)
        if polled.status_code == 202:
            continue
        # ...拒绝、过期与其他错误抛出 LoginError
        payload = polled.json()
        if payload.get("status") == "approved" and isinstance(payload.get("token"), str) and payload["token"]:
            return {"access_token": payload["token"]}
    raise LoginError("kilo: authorization timed out")

（节选，见 hoshimi/core/login/kilo.py。）Kilo 返回的是长期有效的密钥，所以流程不编造 refresh_token 或 expires_at，免得为它安排根本不存在的续期。

注册与写回复制锚点链接

在 core/login/manager.py 的 _flows() 中加入 "<id>": start_<id>_login（流程在这里惰性导入，一个站点的问题不会影响启动）。LoginManager 负责其余的事：

• 每次登录是一个会话（pending、succeeded、failed），在后台线程中运行，开始一小时后被清理；会话可以取消，等待授权码的会话可以提交授权码。
• 成功后，凭据写入 config.json 中该渠道的 accounts，并启用这个渠道。凭据里带有稳定身份（user_id、account_id、userid、uid、sub、access_key_id、api_key、access_token 中的第一个）时，同一身份再次登录会替换原账号，而不是新增一个。
• 写入之后渠道池立即按新配置重建，新账号马上可用。

需要续期的凭据，在适配器中实现 refresh_account。refresh.py 的定时器每 30 分钟检查一次所有账号，在记录的过期时间前一小时内（没有记录过期时间的账号每次都）续期，并只把变化的那个账号写回 config.json。