qyyfw/qihuaiflow
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

recover: chat.py Auto routing + models.py capability-defaults route (lost in 5d7f32d)

Browse Source
This commit is contained in:
qyyfw committed 2026-07-12 07:12:59 +08:00
1 parent 405e836ee2
commit ab9cf55321
2 files changed
+1520 -415
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/python/app/api/admin/models.py
+542 -145
View File
@@ -7,9 +7,13 @@ from sqlalchemy import func
from typing import List, Optional
from pydantic import BaseModel
from ...database import get_db, SessionLocal
from ...models import AIModel, ModelProvider
from ...models import AIModel, ModelProvider, SiteSetting
from ...auth import get_current_admin
from ...middleware.log_service import LogService
from ...capability_registry import (
CAPABILITY_TYPES, CAPABILITY_COLORS, TEXT_FAMILY,
get_capability_defaults, set_capability_default, resolve_auto_model,
)
import uuid
import json
import logging
@@ -18,6 +22,36 @@ from datetime import datetime
router = APIRouter()
# ============================================================
# 请求体模型（前端以 JSON body 提交，而非 query 参数）
# ============================================================
class ModelCreateRequest(BaseModel):
provider_id: str
name: str
display_name: Optional[str] = None
api_key: Optional[str] = None
endpoint: Optional[str] = None
capabilities: Optional[List[str]] = None
timeout: int = 60
status: str = "paid"
price_per_million_tokens: float = 0
is_enabled: bool = True
class ModelUpdateRequest(BaseModel):
provider_id: Optional[str] = None
name: Optional[str] = None
display_name: Optional[str] = None
api_key: Optional[str] = None
endpoint: Optional[str] = None
capabilities: Optional[List[str]] = None
timeout: Optional[int] = None
status: Optional[str] = None
price_per_million_tokens: Optional[float] = None
is_enabled: Optional[bool] = None
# ============================================================
# 协议预设列表
# ============================================================
@@ -75,6 +109,7 @@ def model_to_dict(m: AIModel) -> dict:
"id": m.id,
"provider_id": m.provider_id,
"name": m.name,
"display_name": getattr(m, "display_name", None) or m.name,
"api_key": m.api_key,
"endpoint": m.endpoint,
"capabilities": m.capabilities,
@@ -102,6 +137,17 @@ async def get_protocols(
return PROTOCOLS
@router.get("/capabilities")
async def get_capabilities(
admin_user=Depends(get_current_admin),
):
"""获取平台支持的全部模型能力类型（含标签、测试类型、配色、描述）"""
return [
{**c, "color": CAPABILITY_COLORS.get(c["id"], "#6B7280")}
for c in CAPABILITY_TYPES
]
# ============================================================
# 模型供应商管理
# ============================================================
@@ -320,7 +366,8 @@ async def add_models(
model_info = next((m for m in FETCHABLE_MODELS if m["id"] == model_id), None)
# 使用 model_id 作为模型名称
model_name = model_info["name"] if model_info else model_id
model_name = model_id # API ID
display_name_val = model_info["name"] if model_info else model_id
# 检查是否已存在同名模型
existing = db.query(AIModel).filter(
@@ -340,6 +387,7 @@ async def add_models(
id=str(uuid.uuid4()),
provider_id=id,
name=model_name,
display_name=display_name_val,
api_key=provider.api_key or "",
endpoint=endpoint,
capabilities=model_info.get("capabilities") if model_info else ["text"],
@@ -401,35 +449,28 @@ async def get_models(
@router.post("/")
async def create_model(
provider_id: str,
name: str,
api_key: str = None,
endpoint: str = None,
capabilities: List[str] = None,
timeout: int = 60,
status: str = "paid",
price_per_million_tokens: float = 0,
is_enabled: bool = True,
data: ModelCreateRequest,
admin_user=Depends(get_current_admin),
db: Session = Depends(get_db),
):
"""添加模型"""
# 验证供应商存在
provider = db.query(ModelProvider).filter(ModelProvider.id == provider_id).first()
provider = db.query(ModelProvider).filter(ModelProvider.id == data.provider_id).first()
if not provider:
raise HTTPException(status_code=404, detail="供应商不存在")
model = AIModel(
id=str(uuid.uuid4()),
provider_id=provider_id,
name=name,
api_key=api_key,
endpoint=endpoint,
capabilities=capabilities,
timeout=timeout,
status=status,
price_per_million_tokens=price_per_million_tokens,
is_enabled=is_enabled,
provider_id=data.provider_id,
name=data.name,
display_name=data.display_name,
api_key=data.api_key,
endpoint=data.endpoint,
capabilities=data.capabilities,
timeout=data.timeout,
status=data.status,
price_per_million_tokens=data.price_per_million_tokens,
is_enabled=data.is_enabled,
)
db.add(model)
db.commit()
@@ -446,18 +487,73 @@ async def create_model(
return {"message": "模型已创建", "id": model.id}
# ============================================================
# 按能力默认模型（必须定义在 /{id} 路由之前，避免被路径参数拦截）
# ============================================================
@router.get("/capability-defaults")
async def get_capability_defaults_api(
admin_user=Depends(get_current_admin),
db: Session = Depends(get_db),
):
"""获取每个能力当前配置的默认模型，以及可选模型列表"""
defaults = get_capability_defaults(db)
models = db.query(AIModel).filter(AIModel.is_enabled == True).all()
model_list = [{
"id": m.id,
"name": m.name,
"display_name": getattr(m, "display_name", None) or m.name,
"provider_name": (
db.query(ModelProvider).filter(ModelProvider.id == m.provider_id).first().name
if m.provider_id else ""
),
"capabilities": m.capabilities or [],
} for m in models]
return {
"capabilities": [
{**c, "color": CAPABILITY_COLORS.get(c["id"], "#6B7280")}
for c in CAPABILITY_TYPES
],
"models": model_list,
"defaults": defaults,
}
class CapabilityDefaultRequest(BaseModel):
capability: str
model_id: Optional[str] = None
@router.put("/capability-defaults")
async def set_capability_default_api(
data: CapabilityDefaultRequest,
admin_user=Depends(get_current_admin),
db: Session = Depends(get_db),
):
"""为某个能力设置 / 清除默认模型"""
valid = {c["id"] for c in CAPABILITY_TYPES}
if data.capability not in valid:
raise HTTPException(status_code=400, detail=f"无效的能力类型: {data.capability}")
if data.model_id:
model = db.query(AIModel).filter(AIModel.id == data.model_id).first()
if not model:
raise HTTPException(status_code=404, detail="模型不存在")
updated = set_capability_default(db, data.capability, data.model_id)
LogService.log_audit(
admin_id=admin_user.id,
admin_name=admin_user.username,
action="set_capability_default",
target_type="model_capability",
target_id=data.capability,
detail=f"设置能力 {data.capability} 的默认模型为 {data.model_id or '（清除）'}",
)
return {"message": "已更新能力默认模型", "defaults": updated}
@router.put("/{id}")
async def update_model(
id: str,
provider_id: str = None,
name: str = None,
api_key: str = None,
endpoint: str = None,
capabilities: List[str] = None,
timeout: int = None,
status: str = None,
price_per_million_tokens: float = None,
is_enabled: bool = None,
data: ModelUpdateRequest,
admin_user=Depends(get_current_admin),
db: Session = Depends(get_db),
):
@@ -466,27 +562,29 @@ async def update_model(
if not model:
raise HTTPException(status_code=404, detail="模型不存在")
if provider_id is not None:
provider = db.query(ModelProvider).filter(ModelProvider.id == provider_id).first()
if data.provider_id is not None:
provider = db.query(ModelProvider).filter(ModelProvider.id == data.provider_id).first()
if not provider:
raise HTTPException(status_code=404, detail="供应商不存在")
model.provider_id = provider_id
if name is not None:
model.name = name
if api_key is not None:
model.api_key = api_key
if endpoint is not None:
model.endpoint = endpoint
if capabilities is not None:
model.capabilities = capabilities
if timeout is not None:
model.timeout = timeout
if status is not None:
model.status = status
if price_per_million_tokens is not None:
model.price_per_million_tokens = price_per_million_tokens
if is_enabled is not None:
model.is_enabled = is_enabled
model.provider_id = data.provider_id
if data.name is not None:
model.name = data.name
if data.display_name is not None:
model.display_name = data.display_name
if data.api_key is not None:
model.api_key = data.api_key
if data.endpoint is not None:
model.endpoint = data.endpoint
if data.capabilities is not None:
model.capabilities = data.capabilities
if data.timeout is not None:
model.timeout = data.timeout
if data.status is not None:
model.status = data.status
if data.price_per_million_tokens is not None:
model.price_per_million_tokens = data.price_per_million_tokens
if data.is_enabled is not None:
model.is_enabled = data.is_enabled
model.updated_at = datetime.now()
db.commit()
@@ -602,11 +700,134 @@ async def get_model_stats(
}
# ============================================================
# 按能力指定默认模型（供技能 / 应用 / IM Bot 自动匹配）
# ============================================================
# (按能力默认模型路由已移至 /{id} 路由之前，避免路径参数冲突)
# ============================================================
# 模型连接测试 API (Phase 2)
# ============================================================
@router.post("/models/{model_id}/test-connection")
def _detect_model_type(capabilities):
"""根据 capabilities 判断模型类型，返回测试类型。
优先级（从高到低）：image > video > audio > asr > ocr > embedding > rerank > vision > thinking > tool > text
"""
if not capabilities:
return "text"
caps = [c.lower() for c in capabilities]
if "image" in caps:
return "image"
if "video" in caps:
return "video"
if "audio" in caps:
return "audio"
if "asr" in caps or "speech" in caps:
return "asr"
if "ocr" in caps:
return "ocr"
if "embedding" in caps:
return "embedding"
if "rerank" in caps:
return "rerank"
if "vision" in caps:
return "vision"
if "thinking" in caps:
return "thinking"
if "tool" in caps:
return "tool"
return "text"
def _build_api_base(provider, ai_model):
"""构造 API base URL 和 headers"""
api_key = ai_model.api_key or (provider.api_key if provider else None)
base_url = ""
if provider and provider.api_base_url:
base_url = provider.api_base_url.rstrip("/")
elif ai_model.endpoint:
# 从 endpoint 反推 base
endpoint = ai_model.endpoint
for suffix in ["/v1/chat/completions", "/v1/images/generations", "/v1/audio/speech", "/v1/audio/transcriptions"]:
if endpoint.endswith(suffix):
base_url = endpoint[:-len(suffix)]
break
if not base_url:
base_url = endpoint.rsplit("/v1", 1)[0] if "/v1" in endpoint else ""
return api_key, base_url
# ============================================================
# 批量连接测试（必须定义在 /{model_id}/test-connection 之前，避免被路径参数拦截）
# ============================================================
@router.post("/test-all")
async def test_all_models(
admin_user=Depends(get_current_admin),
db: Session = Depends(get_db),
):
"""批量测试所有已启用模型的连接状态"""
import asyncio
models = db.query(AIModel).filter(AIModel.is_enabled == True).all()
results = []
for m in models:
provider = db.query(ModelProvider).filter(ModelProvider.id == m.provider_id).first()
if not provider:
results.append({
"model_id": m.id,
"model_name": m.name,
"display_name": m.display_name or m.name,
"success": False,
"message": "供应商不存在",
"latency_ms": None,
"model_type": "unknown",
})
continue
# 复用 test_model_connection 逻辑
try:
result = await test_model_connection(m.id, admin_user, db)
if isinstance(result, dict):
result["model_id"] = m.id
result["model_name"] = m.name
result["display_name"] = m.display_name or m.name
results.append(result)
else:
results.append({
"model_id": m.id,
"model_name": m.name,
"display_name": m.display_name or m.name,
"success": False,
"message": "测试返回异常",
"latency_ms": None,
"model_type": "unknown",
})
except Exception as e:
results.append({
"model_id": m.id,
"model_name": m.name,
"display_name": m.display_name or m.name,
"success": False,
"message": f"测试异常: {str(e)}",
"latency_ms": None,
"model_type": "unknown",
})
success_count = sum(1 for r in results if r.get("success"))
return {
"total": len(results),
"success": success_count,
"failed": len(results) - success_count,
"results": results,
}
@router.post("/{model_id}/test-connection")
async def test_model_connection(
model_id: str,
admin_user=Depends(get_current_admin),
@@ -614,12 +835,13 @@ async def test_model_connection(
):
"""
测试指定模型的连接是否正常
支持从模型记录或提供商记录读取配置
支持多模态: text/image/audio/vision 自动检测模型类型并使用对应API测试
"""
import time
logger = logging.getLogger(__name__)
logger.info(f"Testing connection for model_id: {model_id}")
# 优先从 AIModel 读取
ai_model = db.query(AIModel).filter(AIModel.id == model_id).first()
if not ai_model:
raise HTTPException(status_code=404, detail="模型不存在")
@@ -628,101 +850,298 @@ async def test_model_connection(
if not provider:
raise HTTPException(status_code=404, detail="模型供应商不存在")
# 确定最终配置
api_key = ai_model.api_key or (provider.api_key if provider else None)
api_url = ai_model.endpoint
# 如果 endpoint 为空，尝试构造
if not api_url and provider and provider.api_base_url:
base = provider.api_base_url.rstrip('/')
if provider.protocol == 'openai_compatible':
api_url = f"{base}/v1/chat/completions"
else:
api_url = f"{base}/v1/chat/completions"
api_key, base_url = _build_api_base(provider, ai_model)
if not api_key:
return {
"success": False,
"message": "未配置 API Key，请在模型或供应商设置中添加",
"latency_ms": None,
"model_type": "unknown",
}
if not api_url:
if not base_url:
return {
"success": False,
"message": "未配置 API 地址，请在模型或供应商设置中添加",
"latency_ms": None,
"model_type": "unknown",
}
model_type = _detect_model_type(ai_model.capabilities)
timeout = ai_model.timeout or 60
import time
start = time.time()
try:
async with httpx.AsyncClient(timeout=timeout) as client:
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
}
# 使用 minimal payload 测试连接
start = time.time()
try:
async with httpx.AsyncClient(timeout=timeout) as client:
# ============================================================
# 文本模型 / 视觉模型 / OCR模型 / 思考模型 → chat/completions
# ============================================================
if model_type in ("text", "vision", "ocr", "thinking", "tool"):
chat_url = f"{base_url}/v1/chat/completions"
payload = {
"model": ai_model.name,
"messages": [{"role": "user", "content": "hi"}],
"max_tokens": 5,
}
resp = await client.post(api_url, json=payload, headers=headers)
# 视觉 / OCR 模型附带图片测试（用 base64 data-URI，避免远端 URL 不可达）
if model_type in ("vision", "ocr"):
_test_png_b64 = (
"iVBORw0KGgoAAAANSUhEUgAAAEAAAABACAIAAAAlC+aJAAAAjUlEQVR4nO3PoRHAMADDwMzS/XdsJzAoUnQnYwH/eceesdv6c9uhAPShAPShAPSh3wDL0dUHoPsAdB+A7v0Ay9HVB6D7AHQfgO79AMvR1Qeg+wB0H4Du/QDL0dUHoPsAdB+A7v0Ay9HVB6D7AHQfgO79AMvR1Qeg+wB0H4Du/QDL0dUHoPsAdB+A7vWAD38HGZb2HO9/AAAAAElFTkSuQmCC"
)
payload["messages"] = [{
"role": "user",
"content": [
{"type": "text", "text": "What is in this image?"},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{_test_png_b64}"}},
],
}]
payload["max_tokens"] = 64
resp = await client.post(chat_url, json=payload, headers=headers)
latency = int((time.time() - start) * 1000)
if resp.status_code == 200:
try:
data = resp.json()
model_used = data.get("model", ai_model.name)
return {
"success": True,
"message": f"连接成功，使用模型: {model_used}",
"message": f"连接成功 ({model_type})，模型: {model_used}",
"latency_ms": latency,
"model_type": model_type,
"model_in_response": model_used,
}
except Exception:
else:
return _build_error_response(resp, latency, model_type)
# ============================================================
# 图像生成模型 → images/generations
# ============================================================
elif model_type == "image":
image_url = f"{base_url}/v1/images/generations"
payload = {
"model": ai_model.name,
"prompt": "a red circle",
"n": 1,
"size": "256x256",
}
resp = await client.post(image_url, json=payload, headers=headers)
latency = int((time.time() - start) * 1000)
if resp.status_code == 200:
data = resp.json()
img_count = len(data.get("data", []))
return {
"success": True,
"message": f"连接成功 (HTTP {resp.status_code})",
"latency_ms": latency,
}
elif resp.status_code == 401:
return {
"success": False,
"message": "认证失败：API Key 无效或已过期",
"latency_ms": latency,
}
elif resp.status_code == 403:
return {
"success": False,
"message": "访问被拒绝：API Key 权限不足",
"latency_ms": latency,
}
elif resp.status_code == 429:
return {
"success": False,
"message": "触发限流：请稍后重试",
"message": f"图像生成成功 ({model_type})，返回 {img_count} 张图片",
"latency_ms": latency,
"model_type": model_type,
}
else:
return _build_error_response(resp, latency, model_type)
# ============================================================
# 音频模型 → audio/speech (TTS)
# ============================================================
elif model_type == "audio":
audio_url = f"{base_url}/v1/audio/speech"
payload = {
"model": ai_model.name,
"input": "hello",
"voice": "alloy",
}
resp = await client.post(audio_url, json=payload, headers=headers)
latency = int((time.time() - start) * 1000)
if resp.status_code == 200:
content_type = resp.headers.get("content-type", "")
has_audio = "audio" in content_type or len(resp.content) > 100
return {
"success": True,
"message": f"音频服务连接成功 ({model_type})，响应类型: {content_type}",
"latency_ms": latency,
"model_type": model_type,
}
else:
return _build_error_response(resp, latency, model_type)
# ============================================================
# 嵌入向量模型 → embeddings
# ============================================================
elif model_type == "embedding":
embed_url = f"{base_url}/v1/embeddings"
payload = {
"model": ai_model.name,
"input": "hello world",
}
resp = await client.post(embed_url, json=payload, headers=headers)
latency = int((time.time() - start) * 1000)
if resp.status_code == 200:
data = resp.json()
emb = data.get("data") or []
dim = len(emb[0].get("embedding", [])) if emb else 0
return {
"success": True,
"message": f"嵌入服务连接成功 ({model_type})，向量维度: {dim}",
"latency_ms": latency,
"model_type": model_type,
}
else:
return _build_error_response(resp, latency, model_type)
# ============================================================
# 重排序模型 → rerank
# ============================================================
elif model_type == "rerank":
rerank_url = f"{base_url}/v1/rerank"
payload = {
"model": ai_model.name,
"query": "什么是人工智能",
"documents": ["人工智能是计算机科学的一个分支", "今天天气晴朗"],
}
resp = await client.post(rerank_url, json=payload, headers=headers)
latency = int((time.time() - start) * 1000)
if resp.status_code == 200:
data = resp.json()
results = data.get("results") or data.get("data") or []
return {
"success": True,
"message": f"重排序服务连接成功 ({model_type})，返回 {len(results)} 条结果",
"latency_ms": latency,
"model_type": model_type,
}
else:
return _build_error_response(resp, latency, model_type)
# ============================================================
# 语音识别模型 → audio/transcriptions (ASR)
# ============================================================
elif model_type == "asr":
asr_url = f"{base_url}/v1/audio/transcriptions"
# 一段极短的静音 WAV（base64），仅用于探活端点是否可达
silent_wav_b64 = (
"UklGRiQAAABXQVZFZm10IBAAAAABAAEAQB8AAIA+AAACABAAZGF0YQAAAAA="
)
files = {
"file": ("test.wav", silent_wav_b64, "audio/wav"),
"model": (None, ai_model.name),
}
asr_headers = {"Authorization": headers["Authorization"]}
try:
err_data = resp.json()
err_msg = err_data.get("error", {}).get("message", resp.text[:200])
resp = await client.post(asr_url, files=files, headers=asr_headers)
except Exception:
err_msg = resp.text[:200]
# 某些实现要求 JSON 而非表单，回退尝试 JSON
resp = await client.post(
asr_url,
json={"model": ai_model.name, "audio": silent_wav_b64},
headers=headers,
)
latency = int((time.time() - start) * 1000)
if resp.status_code == 200:
return {
"success": True,
"message": f"语音识别服务连接成功 ({model_type})",
"latency_ms": latency,
"model_type": model_type,
}
elif resp.status_code in (400, 422):
# 端点可达但音频数据无效（探活目的达成）
return {
"success": True,
"message": f"语音识别端点可达 ({model_type})，返回 {resp.status_code}（需真实音频）",
"latency_ms": latency,
"model_type": model_type,
}
elif resp.status_code in (404, 405):
# OpenAI transcriptions 路径不存在 → 可能是 MiMo 风格
# 的 /v1/chat/completions + input_audio（如 mimo-v2.5-asr）
chat_url = f"{base_url}/v1/chat/completions"
asr_payload = {
"model": ai_model.name,
"messages": [{
"role": "user",
"content": [{
"type": "input_audio",
"input_audio": {
"data": f"data:audio/wav;base64,{silent_wav_b64}"
},
}],
}],
"asr_options": {"language": "auto"},
"max_tokens": 5,
}
try:
resp2 = await client.post(chat_url, json=asr_payload, headers=headers)
except Exception:
resp2 = resp
latency2 = int((time.time() - start) * 1000)
if resp2.status_code == 200:
return {
"success": True,
"message": f"语音识别服务连接成功 ({model_type})",
"latency_ms": latency2,
"model_type": model_type,
}
elif resp2.status_code in (400, 422):
return {
"success": True,
"message": f"语音识别端点可达 ({model_type})，返回 {resp2.status_code}（需真实音频）",
"latency_ms": latency2,
"model_type": model_type,
}
else:
return _build_error_response(resp, latency, model_type)
else:
return _build_error_response(resp, latency, model_type)
# ============================================================
# 视频模型 → 尝试 chat/completions (可能不支持，回退)
# ============================================================
elif model_type == "video":
chat_url = f"{base_url}/v1/chat/completions"
payload = {
"model": ai_model.name,
"messages": [{"role": "user", "content": "hi"}],
"max_tokens": 5,
}
resp = await client.post(chat_url, json=payload, headers=headers)
latency = int((time.time() - start) * 1000)
if resp.status_code == 200:
data = resp.json()
model_used = data.get("model", ai_model.name)
return {
"success": True,
"message": f"连接成功 ({model_type})，模型: {model_used}",
"latency_ms": latency,
"model_type": model_type,
"model_in_response": model_used,
}
else:
return _build_error_response(resp, latency, model_type)
else:
return {
"success": False,
"message": f"API 错误 (HTTP {resp.status_code}): {err_msg}",
"latency_ms": latency,
"message": f"未知模型类型: {model_type}",
"latency_ms": int((time.time() - start) * 1000),
"model_type": model_type,
}
except httpx.TimeoutException:
latency = int((time.time() - start) * 1000)
return {
"success": False,
"message": f"连接超时（{timeout}s）：请检查 API 地址是否正确，或服务是否可访问",
"latency_ms": latency,
"model_type": model_type,
}
except Exception as e:
latency = int((time.time() - start) * 1000)
@@ -731,57 +1150,37 @@ async def test_model_connection(
"success": False,
"message": f"连接失败: {str(e)}",
"latency_ms": latency,
"model_type": model_type,
}
@router.post("/models/{model_id}/set-default")
async def set_default_model(
model_id: str,
admin_user=Depends(get_current_admin),
db: Session = Depends(get_db),
):
"""
设置全局默认模型（同时关闭其他模型的默认标记）
"""
ai_model = db.query(AIModel).filter(AIModel.id == model_id).first()
if not ai_model:
raise HTTPException(status_code=404, detail="模型不存在")
if not ai_model.is_enabled:
raise HTTPException(status_code=400, detail="只能将已启用的模型设为默认")
# 关闭所有其他模型的默认标记
db.query(AIModel).filter(AIModel.is_default == True).update({"is_default": False})
# 开启此模型
ai_model.is_default = True
ai_model.updated_at = datetime.now()
db.commit()
LogService.log_audit(
admin_id=admin_user.id,
admin_name=admin_user.username,
action="set_default_model",
target_type="ai_model",
target_id=model_id,
detail=f"设置默认模型为 {ai_model.name}",
)
return {"message": f"已将 {ai_model.name} 设为默认模型"}
class CreateModelRequest(BaseModel):
name: str
provider_id: str
api_key: Optional[str] = None
endpoint: Optional[str] = None
capabilities: Optional[List[str]] = ["text"]
timeout: int = 60
status: str = "paid"
price_per_million_tokens: float = 0.0
is_enabled: bool = True
def _build_error_response(resp, latency, model_type):
"""构造错误响应"""
if resp.status_code == 401:
msg = "认证失败：API Key 无效或已过期"
elif resp.status_code == 403:
msg = "访问被拒绝：API Key 权限不足"
elif resp.status_code == 429:
msg = "触发限流：请稍后重试"
elif resp.status_code == 404:
msg = f"API 地址不存在 (404)：请检查 endpoint 配置是否正确"
else:
try:
err_data = resp.json()
err_msg = err_data.get("error", {}).get("message", resp.text[:200])
except Exception:
err_msg = resp.text[:200]
msg = f"API 错误 (HTTP {resp.status_code}): {err_msg}"
return {
"success": False,
"message": msg,
"latency_ms": latency,
"model_type": model_type,
}
class UpdateModelRequest(BaseModel):
"""更新模型请求（保留兼容，实际更新使用 ModelUpdateRequest）"""
name: Optional[str] = None
api_key: Optional[str] = None
endpoint: Optional[str] = None
@@ -791,5 +1190,3 @@ class UpdateModelRequest(BaseModel):
price_per_million_tokens: Optional[float] = None
is_enabled: Optional[bool] = None
from typing import Optional
backend/python/app/api/chat.py
+978 -270
View File
File diff suppressed because it is too large. Load diff