"""XTC 简化接口:/v1/xtc/providers、/v1/xtc/models、/v1/xtc/chat。 完全兼容前端 XTC-AI/src/common/api.js 调用。 """ from __future__ import annotations import json import re from typing import Any, AsyncIterator, Optional from fastapi import APIRouter, Depends, Header, Query, Request from fastapi.responses import StreamingResponse from sse_starlette.sse import EventSourceResponse from ..adapters import gemini_api, openai as openai_adapter from ..auth import require_access_key from ..cache import get_cached_models, set_cached_models from ..config import GEMINI_DEFAULT_MODEL, get_settings from ..errors import HttpError from ..media.imagefix import fix_images_in_messages, infer_fix_mode from ..providers.policy import is_model_allowed from ..services.config_store import load_config from ..utils.sse import sse_data, sse_done from ..utils.thought import ( extract_thought_and_answer, extract_thought_from_reasoning_content, ) from ._common import ( CORS_HEADERS, normalize_chat_body, ok_with_cors, record_usage, select_provider_and_key, ) router = APIRouter(prefix="/v1/xtc", tags=["xtc"]) # simple 流式:thought 标签识别(提到模块级,避免每次调用重新 import + compile) _THOUGHT_OPEN_RE = re.compile(r"\s*$", re.IGNORECASE) _THOUGHT_CLOSE_RE = re.compile(r"^\s*", re.IGNORECASE) @router.get("/providers") async def list_providers(_key: str = Depends(require_access_key)) -> dict: config = await load_config() items = [p.public() for p in config.providers if p.enabled] return ok_with_cors( { "defaultProviderId": config.default_provider_id, "providers": items, "count": len(items), } ) @router.get("/models") async def list_models( _key: str = Depends(require_access_key), provider: Optional[str] = Query(default=None, alias="provider"), all: int = Query(default=0, alias="all"), x_provider: Optional[str] = Header(default=None, alias="x-provider"), ) -> dict: # 前端 api.js 的 fetchModels 通过 x-provider 头指定厂商(与 /v1/xtc/chat # 一致),此处需同时识别,否则模型选择器永远只展示默认厂商的模型。 # 优先取 query 参数(更显式),其次取头。 provider = provider or (x_provider.strip() if x_provider else x_provider) config = await load_config() enabled = [p for p in config.providers if p.enabled] if all and not provider: # 合并所有厂商:单个厂商失败不阻断聚合请求(仅跳过并记日志) out: list[dict] = [] for p in enabled: try: models = await _fetch_provider_models(p) except HttpError as e: print(f"[xtc/models] provider {p.id} list failed: {e.message}") continue for m in models: mid = m.get("id") or "" if not mid or not is_model_allowed(p, mid): continue full_id = f"{p.model_prefix}{mid}" if p.model_prefix else mid out.append({"id": full_id, "name": m.get("name") or full_id, "provider": p.id}) return ok_with_cors({"models": out, "count": len(out), "all": True}) # 单厂商 target = None if provider: target = next((p for p in enabled if p.id == provider), None) if not target: target = config.find_default_provider() if not target: raise HttpError( "No enabled provider. Add one in /admin.", status=503, code="service_unavailable", ) models = await _fetch_provider_models(target) out = [] for m in models: mid = m.get("id") or "" if not mid or not is_model_allowed(target, mid): continue full_id = f"{target.model_prefix}{mid}" if target.model_prefix else mid out.append({"id": full_id, "name": m.get("name") or full_id}) return ok_with_cors( {"provider": target.id, "models": out, "count": len(out)} ) async def _fetch_provider_models(provider) -> list[dict]: """拉取单个厂商的模型列表。 上游失败时抛 HttpError(由全局异常处理器转为统一错误响应),不再静默吞掉 异常返回 []——否则前端无法区分"该厂商无模型"与"上游报错",也会让 503/超时 等瞬时故障被误判为永久空列表(与旧后端行为一致:旧后端会向客户端回传 {ok:false,error:{...}})。 仅缓存非空结果,避免上游瞬时返回空列表时缓存被"毒化"30s。 """ cached = get_cached_models(provider.id) if cached is not None: return cached if not provider.api_keys: raise HttpError( f"Provider '{provider.id}' has no api_keys configured", status=503, code="server_misconfigured", ) if provider.type == "gemini": data = await gemini_api.list_models_raw(api_key=provider.api_keys[0]) else: data = await openai_adapter.list_models_raw( base_url=provider.base_url, api_key=provider.api_keys[0] ) # 只缓存非空结果,避免上游瞬时返回空列表时缓存被"毒化" if data: set_cached_models(provider.id, data) return data # ===== /v1/xtc/chat ===== @router.post("/chat") async def xtc_chat( request: Request, _key: str = Depends(require_access_key), x_provider: Optional[str] = Header(default=None, alias="x-provider"), accept: Optional[str] = Header(default=None), ): """简化聊天:支持 json 与 multipart。 关键字段(兼容前端 api.js): - messages: list[{role, content}] - model: str - provider: str - stream: bool - stream_mode: "simple" | "openai" - image_fix: "mirror_h" | "rotate_180" | "mirror_h_rotate_180" - image_camera_facing: "front" | "back" - files / file_names: list[str] 附加文本拼到 messages - thoughts: bool 是否返回 thought - max_tokens / temperature / top_p / seed / stop / n """ content_type = (request.headers.get("content-type") or "").lower() if "multipart/form-data" in content_type: body = await _parse_multipart(request) else: try: body = await request.json() except Exception: body = {} if not isinstance(body, dict): raise HttpError("invalid body", status=400, code="bad_request") # 统一归一化:兼容 input+images 简化形态 与 messages OpenAI 形态 messages, model_from_body, provider_id = normalize_chat_body(body) if not messages: raise HttpError("input or messages is required", status=400, code="bad_request") model = model_from_body or GEMINI_DEFAULT_MODEL provider_id = provider_id or x_provider # 图片修正 image_fix_mode = body.get("image_fix") camera_facing = body.get("image_camera_facing") if not image_fix_mode and camera_facing: image_fix_mode = infer_fix_mode(camera_facing) image_fixed = False image_fix_failed = 0 image_warning: Optional[str] = None if image_fix_mode: success, failed, image_warning = await fix_images_in_messages(messages, image_fix_mode) image_fixed = success > 0 image_fix_failed = failed config = await load_config() provider, api_key, clean_model = await select_provider_and_key( config=config, provider_id=provider_id, model=model, fallback_model=GEMINI_DEFAULT_MODEL, ) stream = bool(body.get("stream")) stream_mode = str(body.get("stream_mode") or "openai").lower() # 构造上游 body(透传 OpenAI 风格参数) upstream_body = _build_upstream_body(body, clean_model, messages) if stream: return _stream_xtc_chat( provider, api_key, clean_model, upstream_body, messages, stream_mode, image_fix_mode, image_fixed, image_fix_failed, image_warning, _key, ) # 非流式 try: if provider.type == "gemini": openai_resp = await gemini_api.chat_completions( api_key=api_key, model=clean_model, messages=messages, body=upstream_body, stream=False, ) else: openai_resp = await openai_adapter.chat_completions( base_url=provider.base_url, api_key=api_key, body=upstream_body, stream=False, ) record_usage( access_key=_key, provider=provider.id, model=clean_model, usage=(openai_resp or {}).get("usage"), ok=True, ) return _build_xtc_response( openai_resp, provider=provider.id, model=clean_model, image_fix_mode=image_fix_mode, image_fixed=image_fixed, image_fix_failed=image_fix_failed, image_warning=image_warning, camera_facing=camera_facing, ) except HttpError as e: record_usage( access_key=_key, provider=provider.id, model=clean_model, usage=None, ok=False, error_code=e.code, ) raise def _build_upstream_body(body: dict, clean_model: str, messages: list) -> dict: """从 XTC body 构造上游 OpenAI 风格 body。""" out = { "model": clean_model, "messages": messages, } for k in ( "temperature", "top_p", "top_k", "max_tokens", "max_completion_tokens", "presence_penalty", "frequency_penalty", "seed", "stop", "n", "response_format", "tools", "tool_choice", "reasoning_effort", "stream_options", "extra_body", "google", "user", ): if k in body and body[k] is not None: out[k] = body[k] return out def _build_xtc_response( openai_resp: dict, *, provider: str, model: str, image_fix_mode: Optional[str], image_fixed: bool, image_fix_failed: int, image_warning: Optional[str], camera_facing: Optional[str], ) -> dict: """从 OpenAI ChatCompletion 构造 XTC 简化响应。""" choices = openai_resp.get("choices") or [] choice = choices[0] if choices else {} message = choice.get("message") or {} raw_text = message.get("content") or "" reasoning_content = message.get("reasoning_content") thought, text = extract_thought_and_answer(raw_text) if not thought and reasoning_content: thought = extract_thought_from_reasoning_content(reasoning_content) if not text: text = raw_text payload: dict[str, Any] = { "ok": True, "provider": provider, "model": model, "thought": thought, "text": text, "raw": raw_text, "usage": openai_resp.get("usage") or {}, "finish_reason": choice.get("finish_reason"), } if image_fix_mode: payload["image_fix_mode"] = image_fix_mode payload["image_fixed"] = image_fixed payload["image_fix_failed_count"] = image_fix_failed payload["image_camera_facing"] = camera_facing if image_warning: payload["warning"] = image_warning headers = {"x-xtc-provider": provider, "x-xtc-model": model} if image_fix_mode: headers["x-xtc-image-fix-mode"] = image_fix_mode return ok_with_cors(payload, extra_headers=headers) async def _stream_xtc_chat( provider, api_key: str, clean_model: str, upstream_body: dict, messages: list, stream_mode: str, image_fix_mode: Optional[str], image_fixed: bool, image_fix_failed: int, image_warning: Optional[str], access_key: str, ): """流式:simple 模式输出 {type,delta};openai 模式直接透传 OpenAI chunk。""" from ..utils.sse import passthrough_with_usage # 让上游在最后一个流式 chunk 里返回 usage,否则 token 计数永远为 0。 upstream_body = { **upstream_body, "stream_options": {**(upstream_body.get("stream_options") or {}), "include_usage": True}, } headers = { **CORS_HEADERS, "x-xtc-provider": provider.id, "x-xtc-model": clean_model, "Cache-Control": "no-cache", "x-accel-buffering": "no", } if image_fix_mode: headers["x-xtc-image-fix-mode"] = image_fix_mode if stream_mode == "simple": captured: dict = {} return EventSourceResponse( _simple_stream_gen( provider, api_key, clean_model, upstream_body, messages, image_warning, access_key, captured, ), headers=headers, ) # openai 模式:透传 if provider.type == "gemini": captured_g: dict = {} async def gen_openai() -> AsyncIterator[str]: try: chunk_iter = await gemini_api.chat_completions( api_key=api_key, model=clean_model, messages=messages, body=upstream_body, stream=True, ) async for chunk in chunk_iter: if isinstance(chunk, dict) and chunk.get("usage"): captured_g["usage"] = chunk["usage"] yield sse_data(chunk) yield sse_done() except HttpError as e: record_usage( access_key=access_key, provider=provider.id, model=clean_model, usage=None, ok=False, error_code=e.code, ) yield sse_data({"error": {"code": e.code, "message": e.message, "status": e.status}}) yield sse_done() else: record_usage( access_key=access_key, provider=provider.id, model=clean_model, usage=captured_g.get("usage"), ok=True, ) return EventSourceResponse(gen_openai(), headers=headers) # OpenAI 厂商透传 captured_pt: dict = {} async def gen_passthrough() -> AsyncIterator[bytes]: try: chunk_iter = await openai_adapter.chat_completions( base_url=provider.base_url, api_key=api_key, body=upstream_body, stream=True ) async for chunk in passthrough_with_usage(chunk_iter, captured_pt): yield chunk except HttpError as e: record_usage( access_key=access_key, provider=provider.id, model=clean_model, usage=None, ok=False, error_code=e.code, ) err = sse_data({"error": {"code": e.code, "message": e.message, "status": e.status}}) yield err.encode("utf-8") yield sse_done().encode("utf-8") else: record_usage( access_key=access_key, provider=provider.id, model=clean_model, usage=captured_pt.get("usage"), ok=True, ) return StreamingResponse(gen_passthrough(), media_type="text/event-stream", headers=headers) async def _simple_stream_gen( provider, api_key: str, clean_model: str, upstream_body: dict, messages: list, image_warning: Optional[str], access_key: str, captured: Optional[dict] = None, ) -> AsyncIterator[str]: """simple 模式:把 OpenAI chunk 流转换为 {type: text/thought/done, delta}。""" in_thought = False thought_open_sent = False thought_close_sent = False open_tag_re = _THOUGHT_OPEN_RE close_tag_re = _THOUGHT_CLOSE_RE try: if provider.type == "gemini": chunk_iter = await gemini_api.chat_completions( api_key=api_key, model=clean_model, messages=messages, body=upstream_body, stream=True, ) async for chunk in chunk_iter: if isinstance(chunk, dict) and chunk.get("usage") and captured is not None: captured["usage"] = chunk["usage"] event = _simple_from_openai_chunk(chunk) if event: yield sse_data(event) else: # OpenAI 厂商:直接读原始 SSE # 必须用 parse_sse_stream 跨 chunk 缓冲解析,否则跨块切断的 # data: 行会 json.loads 失败被丢弃,导致回复少字、换行被破坏。 from ..utils.sse import parse_sse_stream raw_iter = await openai_adapter.chat_completions( base_url=provider.base_url, api_key=api_key, body=upstream_body, stream=True ) async for sse in parse_sse_stream(raw_iter): if sse.get("__done__"): yield sse_data({"type": "done", "delta": "", "finish_reason": "stop"}) return if sse.get("__raw__") is not None: continue if isinstance(sse, dict) and sse.get("usage") and captured is not None: captured["usage"] = sse["usage"] event = _simple_from_openai_chunk(sse) if event: yield sse_data(event) yield sse_data({"type": "done", "delta": "", "finish_reason": "stop"}) except HttpError as e: record_usage( access_key=access_key, provider=provider.id, model=clean_model, usage=None, ok=False, error_code=e.code, ) yield sse_data( { "type": "error", "error": {"code": e.code, "message": e.message, "status": e.status}, } ) return else: record_usage( access_key=access_key, provider=provider.id, model=clean_model, usage=(captured or {}).get("usage"), ok=True, ) if image_warning: yield sse_data({"type": "warning", "delta": image_warning}) def _simple_from_openai_chunk(chunk: dict) -> Optional[dict]: """把单个 OpenAI chunk 转为 simple 事件。""" choices = chunk.get("choices") or [] if not choices: return None choice = choices[0] delta = choice.get("delta") or {} content = delta.get("content") finish_reason = choice.get("finish_reason") if content is not None: return {"type": "text", "delta": content, "finish_reason": finish_reason} if finish_reason: return {"type": "done", "delta": "", "finish_reason": finish_reason} return None async def _parse_multipart(request: Request) -> dict: """解析 multipart/form-data。 兼容前端 api.js 的 callUpload 路径,该路径发送: - 简单字段:access_key / provider / model / input / stream / max_tokens / image_fix / file_names(JSON 字符串)/ images(图片 URL 字符串,可多条) - 文件字段:name="images" 的图片 UploadFile,name="files" 的文件 UploadFile 图片 UploadFile 转 data URL 放进 out["images"]; 文本类文件 UploadFile 读 utf-8 文本放进 out["files"],文件名进 out["file_names"]。 最终由 normalize_chat_body 统一组装为 messages。 """ import base64 from starlette.datastructures import UploadFile form = await request.form() out: dict[str, Any] = {} # 简单字段(含 input) for key in ("model", "provider", "input", "stream", "stream_mode", "image_fix", "image_camera_facing", "max_tokens", "temperature", "top_p", "seed", "stop", "n", "thoughts"): if key not in form: continue val = form[key] if isinstance(val, UploadFile): continue val = str(val) if val in ("true", "false"): out[key] = val == "true" elif val.lstrip("-").isdigit(): out[key] = int(val) else: try: out[key] = float(val) except (TypeError, ValueError): out[key] = val # messages(JSON 字符串) if "messages" in form: raw = form["messages"] if not isinstance(raw, UploadFile): try: out["messages"] = json.loads(str(raw)) except Exception: out["messages"] = [{"role": "user", "content": str(raw)}] # file_names(前端会发 JSON.stringify(names)) file_names: list[str] = [] if "file_names" in form: raw = form["file_names"] if not isinstance(raw, UploadFile): raw = str(raw) try: parsed = json.loads(raw) if isinstance(parsed, list): file_names = [str(x) for x in parsed] else: file_names = [raw] except Exception: file_names = [raw] # 遍历所有项,收集图片 URL 字符串 + UploadFile image_urls: list[str] = [] files_text: list[str] = [] upload_file_names: list[str] = [] for key, value in form.multi_items(): if isinstance(value, UploadFile): # 检查文件大小限制(10MB) # 获取文件大小(需要先检查是否支持获取大小) try: # 读取文件内容(必须读取,不能仅依赖 value.size) content_bytes = await value.read() file_size = len(content_bytes) # 文件大小限制(10MB) if file_size > 10 * 1024 * 1024: # 10MB limit raise HttpError( f"file too large: {file_size} bytes > 10MB", status=413, code="too_large" ) filename = value.filename or key if key == "images" or key == "image" or key.startswith("image_"): # 图片文件 → data URL mime = (value.content_type or "image/jpeg").split(";")[0].strip() or "image/jpeg" b64 = base64.b64encode(content_bytes).decode("ascii") image_urls.append(f"data:{mime};base64,{b64}") elif key == "files" or key == "file" or key.startswith("file_"): # 文件 → 尝试 utf-8 文本;二进制则记空串占位 try: files_text.append(content_bytes.decode("utf-8")) except Exception: files_text.append("") upload_file_names.append(filename) except HttpError: # 重新抛出HttpError raise except Exception as e: # 处理其他异常,特别是大文件导致的内存问题 if 'memory' in str(e).lower() or 'out of memory' in str(e).lower(): raise HttpError( "Failed to upload inline file.", status=500, code="internal_error" ) raise HttpError( "Failed to upload inline file.", status=500, code="internal_error" ) continue # 字符串值 sval = str(value or "").strip() if not sval: continue if key == "images" or key == "image" or key.startswith("image_"): if sval.startswith(("http://", "https://", "data:")): image_urls.append(sval) elif key == "files" or key == "file" or key.startswith("file_"): files_text.append(sval) upload_file_names.append(key) if image_urls: out["images"] = image_urls if files_text: out["files"] = files_text out["file_names"] = (file_names + upload_file_names) or upload_file_names return out