| from __future__ import annotations |
|
|
| import os |
| from pathlib import Path |
| from typing import Any |
|
|
| import httpx |
| from fastapi import APIRouter, Header, HTTPException |
| from pydantic import BaseModel, Field |
|
|
| from api.support import require_identity |
| from services.config import config |
| from services.log_service import LoggedCall |
| from services.protocol import openai_v1_chat_complete |
|
|
| |
| |
| _PROJECT_ROOT = Path(__file__).resolve().parent.parent |
| _SYSTEM_PROMPT_PATH = _PROJECT_ROOT / "openui" / "system-prompt.txt" |
| _system_prompt_cache: str | None = None |
| _system_prompt_mtime: float | None = None |
|
|
|
|
| def load_system_prompt() -> str: |
| """读取预生成的 OpenUI 系统提示词（约 29KB）。按 mtime 缓存：文件更新后 |
| 下一次调用自动生效，无需重启后端。""" |
| global _system_prompt_cache, _system_prompt_mtime |
| try: |
| mtime = _SYSTEM_PROMPT_PATH.stat().st_mtime |
| except OSError as exc: |
| raise HTTPException( |
| status_code=500, |
| detail={"error": f"缺少 OpenUI 系统提示词文件（{_SYSTEM_PROMPT_PATH}）：{exc}"}, |
| ) from exc |
| if _system_prompt_cache is None or _system_prompt_mtime != mtime: |
| _system_prompt_cache = _SYSTEM_PROMPT_PATH.read_text(encoding="utf-8") |
| _system_prompt_mtime = mtime |
| return _system_prompt_cache |
|
|
|
|
| def _first_non_empty(*values: Any) -> str: |
| for value in values: |
| text = str(value or "").strip() |
| if text: |
| return text |
| return "" |
|
|
|
|
| def _external_target() -> tuple[str, str, str] | None: |
| """ |
| 可选的外部通道（不配置就不存在）。 |
| |
| 只有在明确配置了 OpenUI Cloud 或第三方 OpenAI 兼容接口时才启用； |
| 默认情况下生成式 UI 走本站自己的账号池，不需要任何外部 Key。 |
| """ |
| settings = config.get() |
| cloud_key = _first_non_empty(settings.get("openui_api_key"), os.environ.get("THESYS_API_KEY")) |
| if cloud_key: |
| base_url = _first_non_empty( |
| settings.get("openui_base_url"), |
| os.environ.get("THESYS_BASE_URL"), |
| "https://api.thesys.dev/v1/embed", |
| ).rstrip("/") |
| model = _first_non_empty(settings.get("openui_model"), os.environ.get("THESYS_MODEL"), "openai/gpt-5") |
| return base_url, cloud_key, model |
|
|
| base_url = _first_non_empty(settings.get("openui_llm_base_url"), os.environ.get("OPENUI_LLM_BASE_URL")).rstrip("/") |
| api_key = _first_non_empty(settings.get("openui_llm_api_key"), os.environ.get("OPENUI_LLM_API_KEY")) |
| model = _first_non_empty(settings.get("openui_llm_model"), os.environ.get("OPENUI_LLM_MODEL")) |
| if base_url and api_key and model: |
| return base_url, api_key, model |
| return None |
|
|
|
|
| def _text_of(content: Any) -> str: |
| """从 content（可能是字符串，也可能是多模态分片数组）里抽出纯文本，用于日志预览。""" |
| if isinstance(content, str): |
| return content |
| if isinstance(content, list): |
| parts: list[str] = [] |
| for item in content: |
| if isinstance(item, dict): |
| text = item.get("text") |
| if isinstance(text, str): |
| parts.append(text) |
| return " ".join(parts) |
| return str(content or "") |
|
|
|
|
| def _build_openui_messages(messages: list[OpenUiMessage]) -> list[dict[str, Any]]: |
| """把请求里的消息转成 OpenAI 兼容格式。 |
| |
| 普通消息用纯文本；带图片附件（data:image/ 的 base64）的用户/助手消息， |
| 转成多模态 content 数组（text + image_url），让模型能真正“看到”图片。 |
| """ |
| out: list[dict[str, Any]] = [] |
| for item in messages: |
| if item.role not in ("user", "assistant"): |
| continue |
| images = [url for url in (item.images or []) if isinstance(url, str) and url.startswith("data:image/")] |
| if images: |
| content: list[dict[str, Any]] = [] |
| if item.content.strip(): |
| content.append({"type": "text", "text": item.content}) |
| for url in images: |
| content.append({"type": "image_url", "image_url": {"url": url}}) |
| out.append({"role": item.role, "content": content}) |
| elif item.content.strip(): |
| out.append({"role": item.role, "content": item.content}) |
| return out |
|
|
|
|
| def _completion_text(data: Any) -> str: |
| """从 OpenAI 兼容响应里取出文本（content 可能是字符串，也可能是分片数组）。""" |
| if not isinstance(data, dict): |
| return "" |
| choices = data.get("choices") or [] |
| if not choices: |
| return "" |
| message = choices[0].get("message") or {} |
| content = message.get("content") |
| if isinstance(content, str): |
| return content |
| if isinstance(content, list): |
| parts: list[str] = [] |
| for item in content: |
| if isinstance(item, dict): |
| text = item.get("text") |
| if isinstance(text, str): |
| parts.append(text) |
| return "".join(parts) |
| return str(content or "") |
|
|
|
|
| class OpenUiMessage(BaseModel): |
| role: str |
| content: str = "" |
| images: list[str] = Field(default_factory=list) |
|
|
|
|
| class OpenUiChatRequest(BaseModel): |
| messages: list[OpenUiMessage] = Field(default_factory=list) |
|
|
|
|
| async def _call_local_pool(identity: dict[str, object], model: str, messages: list[dict[str, Any]]) -> str: |
| """走本站账号池（逆向 ChatGPT 通道），与 /v1/chat/completions 同一条链路。""" |
| settings = config.get() |
| payload: dict[str, Any] = { |
| "model": model, |
| "stream": False, |
| "messages": [{"role": "system", "content": load_system_prompt()}, *messages], |
| } |
| max_tokens = _first_non_empty(settings.get("openui_max_tokens"), os.environ.get("OPENUI_MAX_TOKENS")) |
| payload["max_tokens"] = int(max_tokens) if str(max_tokens).isdigit() else 4096 |
| last_user = next((_text_of(m["content"]) for m in reversed(messages) if m["role"] == "user"), "") |
| call = LoggedCall(identity, "/api/openui/chat", model, "生成式 UI", request_text=str(last_user)[:500]) |
| result = await call.run(openai_v1_chat_complete.handle, payload, identity) |
| text = _completion_text(result) |
| if not text.strip(): |
| raise HTTPException(status_code=502, detail={"error": "模型没有返回生成式 UI 内容"}) |
| return text |
|
|
|
|
| async def _call_external(target: tuple[str, str, str], messages: list[dict[str, Any]]) -> str: |
| base_url, api_key, model = target |
| request_body: dict[str, Any] = { |
| "model": model, |
| "stream": False, |
| "messages": [{"role": "system", "content": load_system_prompt()}, *messages], |
| } |
| try: |
| async with httpx.AsyncClient(timeout=httpx.Timeout(180.0, connect=20.0)) as client: |
| response = await client.post( |
| f"{base_url}/chat/completions", |
| headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, |
| json=request_body, |
| ) |
| except httpx.HTTPError as exc: |
| raise HTTPException(status_code=502, detail={"error": f"无法连接生成式 UI 模型服务：{exc}"}) from exc |
|
|
| if response.status_code >= 400: |
| raise HTTPException( |
| status_code=502, |
| detail={"error": f"生成式 UI 模型服务返回 {response.status_code}", "detail": response.text[:500]}, |
| ) |
| try: |
| data = response.json() |
| except ValueError as exc: |
| raise HTTPException(status_code=502, detail={"error": "生成式 UI 模型返回了非 JSON 响应"}) from exc |
| text = _completion_text(data) |
| if not text.strip(): |
| raise HTTPException(status_code=502, detail={"error": "生成式 UI 模型未返回内容"}) |
| return text |
|
|
|
|
| def create_router() -> APIRouter: |
| router = APIRouter() |
|
|
| @router.post("/api/openui/chat") |
| async def openui_chat(payload: OpenUiChatRequest, authorization: str | None = Header(default=None)): |
| identity = require_identity(authorization) |
|
|
| messages = _build_openui_messages(payload.messages) |
| if not messages: |
| raise HTTPException(status_code=400, detail={"error": "缺少可发送的消息"}) |
|
|
| |
| external = _external_target() |
| if external: |
| source = await _call_external(external, messages) |
| return {"source": source, "channel": "external"} |
|
|
| settings = config.get() |
| |
| |
| model = _first_non_empty( |
| settings.get("openui_pool_model"), |
| os.environ.get("OPENUI_POOL_MODEL"), |
| "auto", |
| ) |
| source = await _call_local_pool(identity, model, messages) |
| return {"source": source, "channel": "local"} |
|
|
| return router |
|
|
