| """第三方接入模型的**文本对话**路由（把 /v1/chat/completions 接到中转站上）。 |
| |
| 第三方接入（`services/third_party_image_service.py`）本来就是"一个中转站 key 下的 |
| 模型清单"，里面的模型可能只能生图，也可能只能对话（能力字段含 ``chat``）。 |
| |
| 这个模块负责协议侧的那一半： |
| |
| * 判断一次对话请求该不该走第三方（模型命中 + 能力含 ``chat`` + 权限通过）； |
| * 把上游的 OpenAI 兼容响应规范成本站 `chat.completion` / `chat.completion.chunk`； |
| * 上游报错时抛出可读的异常，交给 `LoggedCall` 记日志并回 502。 |
| |
| **计费**：与站内文本一致 —— 文本对话本来就只记日志、不扣积分，第三方文本同样不扣。 |
| （生图侧才走 `ImageBillingReservation`。） |
| """ |
|
|
| from __future__ import annotations |
|
|
| import time |
| import uuid |
| from typing import Any, Iterator |
|
|
| from fastapi import HTTPException |
|
|
| from services.protocol.openai_v1_chat_complete import completion_chunk, completion_response |
| from services.third_party_image_service import ThirdPartyRoute, third_party_image_service |
| from utils.helper import is_image_chat_request |
|
|
|
|
| def resolve_route(body: dict[str, Any], identity: dict[str, Any] | None) -> ThirdPartyRoute | None: |
| """这次对话请求命中的第三方路由；不该接管时返回 ``None``。 |
| |
| 以下情况一律不接管（交回原来的账号池逻辑）： |
| |
| * 没有身份（内部调用，如 Skills 生成、反推提示词）； |
| * 请求本身是**生图**请求（模型是生图模型，或 ``modalities`` 含 image）； |
| * 模型名没有命中任何能力含 ``chat`` 的第三方模型。 |
| """ |
|
|
| if not isinstance(identity, dict) or not identity: |
| return None |
| if is_image_chat_request(body): |
| return None |
| model = str(body.get("model") or "").strip() |
| if not model: |
| return None |
| route = third_party_image_service.find_chat_route(model) |
| if route is None: |
| return None |
| try: |
| third_party_image_service.assert_access(identity, route) |
| except PermissionError as exc: |
| raise HTTPException(status_code=403, detail={"error": str(exc)}) from exc |
| return route |
|
|
|
|
| def _upstream_failure_message(payload: object) -> str: |
| if isinstance(payload, dict): |
| error = payload.get("error") |
| if isinstance(error, dict) and error.get("message"): |
| return str(error["message"]) |
| if isinstance(error, str) and error: |
| return error |
| if payload.get("message"): |
| return str(payload["message"]) |
| return "第三方对话接口返回了无法解析的响应" |
|
|
|
|
| def _first_message(payload: dict[str, Any]) -> dict[str, Any]: |
| choices = payload.get("choices") |
| first = choices[0] if isinstance(choices, list) and choices and isinstance(choices[0], dict) else {} |
| message = first.get("message") |
| return message if isinstance(message, dict) else {} |
|
|
|
|
| def chat_completion_response(body: dict[str, Any], route: ThirdPartyRoute) -> dict[str, Any]: |
| """非流式：转发给第三方，规范成本站的 `chat.completion`。""" |
|
|
| model = str(body.get("model") or "") |
| messages = body.get("messages") if isinstance(body.get("messages"), list) else None |
| raw = third_party_image_service.chat_completion(dict(body), route) |
| if not isinstance(raw, dict) or not isinstance(raw.get("choices"), list) or not raw.get("choices"): |
| raise RuntimeError(_upstream_failure_message(raw)) |
|
|
| first = raw["choices"][0] if isinstance(raw["choices"][0], dict) else {} |
| message = _first_message(raw) |
| content = message.get("content") |
| normalized = completion_response(model, content if isinstance(content, str) else "", messages=messages) |
| target = normalized["choices"][0]["message"] |
| |
| for key in ("tool_calls", "reasoning_content", "annotations", "images"): |
| if message.get(key): |
| target[key] = message[key] |
| normalized["choices"][0]["finish_reason"] = first.get("finish_reason") or "stop" |
| usage = raw.get("usage") |
| if isinstance(usage, dict) and usage: |
| |
| normalized["usage"] = usage |
| normalized["model"] = model |
| return normalized |
|
|
|
|
| def stream_chat_chunks(body: dict[str, Any], route: ThirdPartyRoute) -> Iterator[dict[str, Any]]: |
| """流式：把上游 SSE 的增量包成 `chat.completion.chunk`。""" |
|
|
| model = str(body.get("model") or "") |
| completion_id = f"chatcmpl-{uuid.uuid4().hex}" |
| created = int(time.time()) |
| sent_role = False |
| sent_any = False |
| for item in third_party_image_service.stream_chat_completion(dict(body), route): |
| choices = item.get("choices") |
| first = choices[0] if isinstance(choices, list) and choices and isinstance(choices[0], dict) else {} |
| delta = first.get("delta") |
| if not isinstance(delta, dict): |
| continue |
| sent_any = True |
| if not sent_role: |
| sent_role = True |
| delta = {"role": "assistant", **delta} |
| yield completion_chunk(model, dict(delta), first.get("finish_reason"), completion_id, created) |
| if not sent_any: |
| yield completion_chunk(model, {"role": "assistant", "content": ""}, None, completion_id, created) |
| yield completion_chunk(model, {}, "stop", completion_id, created) |
|
|
|
|
| def handle(body: dict[str, Any], route: ThirdPartyRoute): |
| """按请求体里的 ``stream`` 分派到流式 / 非流式。""" |
|
|
| if body.get("stream"): |
| return stream_chat_chunks(body, route) |
| return chat_completion_response(body, route) |
|
|
|
|
| __all__ = [ |
| "resolve_route", |
| "handle", |
| "chat_completion_response", |
| "stream_chat_chunks", |
| ] |
|
|
