File size: 5,896 Bytes
59af925 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """第三方接入模型的**文本对话**路由（把 /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:
# 上游的 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",
]
|