diff --git a/backend/python/app/api/admin/models.py b/backend/python/app/api/admin/models.py index 3677df9..0d345d3 100644 --- a/backend/python/app/api/admin/models.py +++ b/backend/python/app/api/admin/models.py @@ -1,795 +1,1192 @@ -""" -大模型库管理 API(管理员) -""" -from fastapi import APIRouter, Depends, HTTPException, Query, Body -from sqlalchemy.orm import Session -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 ...auth import get_current_admin -from ...middleware.log_service import LogService -import uuid -import json -import logging -import httpx -from datetime import datetime - -router = APIRouter() - -# ============================================================ -# 协议预设列表 -# ============================================================ - -PROTOCOLS = [ - {"id": "openai_compatible", "name": "OpenAI兼容", "base_path": "/v1", "description": "兼容OpenAI API格式的模型服务"}, - {"id": "new_api", "name": "New API", "base_path": "/v1", "description": "新一代API聚合平台"}, - {"id": "volcengine", "name": "火山引擎", "base_path": "/v1", "description": "字节跳动火山引擎模型服务"}, - {"id": "aliyun_bailian", "name": "阿里云百炼", "base_path": "/v1", "description": "阿里云百炼大模型平台"}, - {"id": "ollama", "name": "Ollama", "base_path": "/v1", "description": "本地Ollama模型服务"}, - {"id": "xiaoma", "name": "小马算力", "base_path": "/v1", "description": "小马算力模型服务"}, -] - -# ============================================================ -# 模型列表(用于 fetch-models 的回退数据) -# ============================================================ - -FETCHABLE_MODELS = [ - {"id": "gpt-4o", "name": "GPT-4o", "capabilities": ["text", "vision", "tool"], "pricing": "paid"}, - {"id": "gpt-4o-mini", "name": "GPT-4o Mini", "capabilities": ["text", "vision"], "pricing": "paid"}, - {"id": "deepseek-v3", "name": "DeepSeek-V3", "capabilities": ["text", "thinking"], "pricing": "free"}, - {"id": "deepseek-r1", "name": "DeepSeek-R1", "capabilities": ["text", "thinking"], "pricing": "paid"}, - {"id": "qwen-max", "name": "Qwen-Max", "capabilities": ["text", "tool"], "pricing": "free"}, - {"id": "qwen-vl", "name": "Qwen-VL", "capabilities": ["text", "vision"], "pricing": "paid"}, - {"id": "whisper", "name": "Whisper", "capabilities": ["audio"], "pricing": "paid"}, - {"id": "dalle", "name": "DALL-E", "capabilities": ["image"], "pricing": "paid"}, -] - -# 支持通过 API 获取模型列表的协议类型 -FETCHABLE_PROTOCOLS = {"openai_compatible", "new_api", "volcengine", "aliyun_bailian", "ollama", "xiaoma"} - - -# ============================================================ -# 辅助函数 -# ============================================================ - -def provider_to_dict(p: ModelProvider) -> dict: - return { - "id": p.id, - "name": p.name, - "protocol": p.protocol, - "api_key": p.api_key, - "api_base_url": p.api_base_url, - "apiKey": p.api_key, - "apiBaseUrl": p.api_base_url, - "created_at": p.created_at.isoformat() if p.created_at else None, - "updated_at": p.updated_at.isoformat() if p.updated_at else None, - "createdAt": p.created_at.isoformat() if p.created_at else None, - "updatedAt": p.updated_at.isoformat() if p.updated_at else None, - } - - -def model_to_dict(m: AIModel) -> dict: - return { - "id": m.id, - "provider_id": m.provider_id, - "name": m.name, - "api_key": m.api_key, - "endpoint": m.endpoint, - "capabilities": m.capabilities, - "timeout": m.timeout, - "status": m.status, - "price_per_million_tokens": float(m.price_per_million_tokens) if m.price_per_million_tokens else 0, - "is_enabled": m.is_enabled, - "is_default": bool(m.is_default), - "total_calls": m.total_calls or 0, - "total_tokens": m.total_tokens or 0, - "created_at": m.created_at.isoformat() if m.created_at else None, - "updated_at": m.updated_at.isoformat() if m.updated_at else None, - } - - -# ============================================================ -# 协议列表 -# ============================================================ - -@router.get("/protocols") -async def get_protocols( - admin_user=Depends(get_current_admin), -): - """获取支持的协议列表""" - return PROTOCOLS - - -# ============================================================ -# 模型供应商管理 -# ============================================================ - -@router.get("/providers") -async def get_providers( - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """获取模型供应商列表""" - providers = db.query(ModelProvider).order_by( - ModelProvider.created_at.desc() - ).all() - return [provider_to_dict(p) for p in providers] - - -@router.post("/providers") -async def create_provider( - name: str = Body(...), - protocol: str = Body(...), - api_key: str = Body(default=""), - api_base_url: str = Body(default=None), - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """添加模型供应商""" - # 验证协议 - valid_protocols = [p["id"] for p in PROTOCOLS] - if protocol not in valid_protocols: - raise HTTPException(status_code=400, detail=f"无效的协议类型,支持: {', '.join(valid_protocols)}") - - provider = ModelProvider( - id=str(uuid.uuid4()), - name=name, - protocol=protocol, - api_key=api_key, - api_base_url=api_base_url, - ) - db.add(provider) - db.commit() - db.refresh(provider) - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="create_provider", - target_type="model_provider", - target_id=provider.id, - detail=f"添加模型供应商 {name} (协议: {protocol})", - ) - - return {"message": "供应商已创建", "id": provider.id, "provider": provider_to_dict(provider)} - - -@router.put("/providers/{id}") -async def update_provider( - id: str, - name: str = Body(default=None), - protocol: str = Body(default=None), - api_key: str = Body(default=None), - api_base_url: str = Body(default=None), - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """更新模型供应商""" - provider = db.query(ModelProvider).filter(ModelProvider.id == id).first() - if not provider: - raise HTTPException(status_code=404, detail="供应商不存在") - - if name is not None: - provider.name = name - if protocol is not None: - valid_protocols = [p["id"] for p in PROTOCOLS] - if protocol not in valid_protocols: - raise HTTPException(status_code=400, detail=f"无效的协议类型") - provider.protocol = protocol - if api_key is not None: - provider.api_key = api_key - if api_base_url is not None: - provider.api_base_url = api_base_url - provider.updated_at = datetime.now() - db.commit() - db.refresh(provider) - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="update_provider", - target_type="model_provider", - target_id=id, - detail=f"更新模型供应商 {provider.name}", - ) - - return {"message": "供应商已更新", "provider": provider_to_dict(provider)} - - -@router.delete("/providers/{id}") -async def delete_provider( - id: str, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """删除模型供应商""" - provider = db.query(ModelProvider).filter(ModelProvider.id == id).first() - if not provider: - raise HTTPException(status_code=404, detail="供应商不存在") - - # 检查是否有模型关联 - model_count = db.query(func.count(AIModel.id)).filter( - AIModel.provider_id == id - ).scalar() - if model_count > 0: - raise HTTPException(status_code=400, detail=f"该供应商下有 {model_count} 个模型,请先删除模型") - - db.delete(provider) - db.commit() - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="delete_provider", - target_type="model_provider", - target_id=id, - detail=f"删除模型供应商 {provider.name}", - ) - - return {"message": "供应商已删除"} - - -@router.post("/providers/{id}/fetch-models") -async def fetch_models( - id: str, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """获取供应商可用模型列表(优先从真实API获取)""" - provider = db.query(ModelProvider).filter(ModelProvider.id == id).first() - if not provider: - raise HTTPException(status_code=404, detail="供应商不存在") - - logger = logging.getLogger(__name__) - logger.info(f"Fetching models for provider {provider.name} (protocol: {provider.protocol})") - - api_base = (provider.api_base_url or "").rstrip("/") - api_key = provider.api_key or "" - - # 尝试从真实API获取模型列表(只要有 api_base 就尝试调用,api_key 可选) - if api_base and provider.protocol in FETCHABLE_PROTOCOLS: - try: - async with httpx.AsyncClient(timeout=15) as client: - headers = {} - if api_key: - headers["Authorization"] = f"Bearer {api_key}" - resp = await client.get(f"{api_base}/v1/models", headers=headers) - if resp.status_code == 200: - data = resp.json() - models_list = data.get("data", []) - if models_list: - result = [] - for m in models_list: - mid = m.get("id", "") - if not mid: - continue - result.append({ - "id": mid, - "name": mid, - "capabilities": ["text"], - "pricing": "paid", - }) - logger.info(f"Successfully fetched {len(result)} models from provider API") - return {"models": result, "source": "api"} - else: - logger.warning(f"Provider API returned empty model list") - return {"models": [], "source": "api", "message": "API返回的模型列表为空"} - elif resp.status_code == 401 or resp.status_code == 403: - logger.warning(f"Provider API returned {resp.status_code}: auth error") - return {"models": [], "source": "error", "message": "API认证失败,请检查API Key是否正确"} - else: - logger.warning(f"Provider API returned status {resp.status_code}: {resp.text[:200]}") - return {"models": [], "source": "error", "message": f"API请求失败(HTTP {resp.status_code}),请检查API地址和密钥"} - except httpx.TimeoutException: - logger.warning(f"Timeout fetching models from provider API: {api_base}") - return {"models": [], "source": "error", "message": "请求API超时,请检查API地址是否正确"} - except Exception as e: - logger.warning(f"Failed to fetch models from provider API: {e}") - return {"models": [], "source": "error", "message": f"请求API失败: {str(e)}"} - - # 回退:返回固定模型列表(供无 API 的离线环境使用) - logger.info(f"Using fallback model list for provider {provider.name}") - return {"models": FETCHABLE_MODELS, "source": "fallback"} - - -class AddModelsRequest(BaseModel): - model_ids: List[str] - - -@router.post("/providers/{id}/add-models") -async def add_models( - id: str, - req: AddModelsRequest, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """批量添加模型到供应商""" - provider = db.query(ModelProvider).filter(ModelProvider.id == id).first() - if not provider: - raise HTTPException(status_code=404, detail="供应商不存在") - - # 找到协议对应的 base_path - protocol_info = next((p for p in PROTOCOLS if p["id"] == provider.protocol), None) - base_path = protocol_info["base_path"] if protocol_info else "/v1" - - added = [] - for model_id in req.model_ids: - # 在 FETCHABLE_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 - - # 检查是否已存在同名模型 - existing = db.query(AIModel).filter( - AIModel.provider_id == id, - AIModel.name == model_name - ).first() - if existing: - continue - - # 自动生成 endpoint - if provider.api_base_url: - endpoint = f"{provider.api_base_url.rstrip('/')}{base_path}/chat/completions" - else: - endpoint = "" - - model = AIModel( - id=str(uuid.uuid4()), - provider_id=id, - name=model_name, - api_key=provider.api_key or "", - endpoint=endpoint, - capabilities=model_info.get("capabilities") if model_info else ["text"], - timeout=60, - status="free" if (model_info and model_info.get("pricing") == "free") else "paid", - price_per_million_tokens=0, - is_enabled=True, - ) - db.add(model) - added.append(model_name) - - db.commit() - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="add_models", - target_type="model_provider", - target_id=id, - detail=f"为供应商 {provider.name} 批量添加 {len(added)} 个模型", - ) - - return {"message": f"成功添加 {len(added)} 个模型", "added": added} - - -# ============================================================ -# 模型管理 -# ============================================================ - -@router.get("/") -async def get_models( - provider_id: str = Query(None), - status: str = Query(None, description="模型状态筛选: free/paid"), - page: int = Query(1, ge=1), - page_size: int = Query(20, ge=1, le=100), - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """获取模型列表(管理员)""" - query = db.query(AIModel) - - if provider_id: - query = query.filter(AIModel.provider_id == provider_id) - if status: - query = query.filter(AIModel.status == status) - - total = query.count() - models = query.order_by(AIModel.created_at.desc()).offset( - (page - 1) * page_size - ).limit(page_size).all() - - return { - "total": total, - "page": page, - "page_size": page_size, - "items": [model_to_dict(m) for m in 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, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """添加模型""" - # 验证供应商存在 - provider = db.query(ModelProvider).filter(ModelProvider.id == 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, - ) - db.add(model) - db.commit() - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="create_model", - target_type="model", - target_id=model.id, - detail=f"添加模型 {name}", - ) - - return {"message": "模型已创建", "id": model.id} - - -@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, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """更新模型""" - model = db.query(AIModel).filter(AIModel.id == id).first() - 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 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.updated_at = datetime.now() - db.commit() - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="update_model", - target_type="model", - target_id=id, - detail=f"更新模型 {model.name}", - ) - - return {"message": "模型已更新"} - - -@router.delete("/{id}") -async def delete_model( - id: str, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """删除模型""" - model = db.query(AIModel).filter(AIModel.id == id).first() - if not model: - raise HTTPException(status_code=404, detail="模型不存在") - - model_name = model.name - db.delete(model) - db.commit() - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="delete_model", - target_type="model", - target_id=id, - detail=f"删除模型 {model_name}", - ) - - return {"message": "模型已删除"} - - -@router.put("/{id}/toggle") -async def toggle_model( - id: str, - is_enabled: bool = Query(...), - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """开关模型状态""" - model = db.query(AIModel).filter(AIModel.id == id).first() - if not model: - raise HTTPException(status_code=404, detail="模型不存在") - - model.is_enabled = is_enabled - model.updated_at = datetime.now() - db.commit() - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="toggle_model", - target_type="model", - target_id=id, - detail=f"将模型 {model.name} {'启用' if is_enabled else '禁用'}", - ) - - return {"message": f"模型已{'启用' if is_enabled else '禁用'}"} - - -@router.put("/{id}/set-default") -async def set_default_model( - id: str, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """设为全局默认模型(取消其他模型的默认状态)""" - model = db.query(AIModel).filter(AIModel.id == id).first() - if not model: - raise HTTPException(status_code=404, detail="模型不存在") - - # 取消所有模型的默认状态 - db.query(AIModel).filter(AIModel.is_default == True).update({"is_default": False}) - # 设置当前模型为默认 - model.is_default = True - 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="model", - target_id=id, - detail=f"设置模型 {model.name} 为全局默认", - ) - - return {"message": f"已将 {model.name} 设为全局默认模型"} - - -@router.get("/stats") -async def get_model_stats( - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """获取模型调用统计""" - total_calls = db.query(func.sum(AIModel.total_calls)).scalar() or 0 - total_tokens = db.query(func.sum(AIModel.total_tokens)).scalar() or 0 - - return { - "total_calls": int(total_calls), - "total_tokens": int(total_tokens), - } - - -# ============================================================ -# 模型连接测试 API (Phase 2) -# ============================================================ - -@router.post("/models/{model_id}/test-connection") -async def test_model_connection( - model_id: str, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """ - 测试指定模型的连接是否正常 - 支持从模型记录或提供商记录读取配置 - """ - 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="模型不存在") - - provider = db.query(ModelProvider).filter(ModelProvider.id == ai_model.provider_id).first() - 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" - - if not api_key: - return { - "success": False, - "message": "未配置 API Key,请在模型或供应商设置中添加", - "latency_ms": None, - } - if not api_url: - return { - "success": False, - "message": "未配置 API 地址,请在模型或供应商设置中添加", - "latency_ms": None, - } - - 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 测试连接 - payload = { - "model": ai_model.name, - "messages": [{"role": "user", "content": "hi"}], - "max_tokens": 5, - } - resp = await client.post(api_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}", - "latency_ms": latency, - "model_in_response": model_used, - } - except Exception: - 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": "触发限流:请稍后重试", - "latency_ms": latency, - } - else: - try: - err_data = resp.json() - err_msg = err_data.get("error", {}).get("message", resp.text[:200]) - except Exception: - err_msg = resp.text[:200] - return { - "success": False, - "message": f"API 错误 (HTTP {resp.status_code}): {err_msg}", - "latency_ms": latency, - } - except httpx.TimeoutException: - latency = int((time.time() - start) * 1000) - return { - "success": False, - "message": f"连接超时({timeout}s):请检查 API 地址是否正确,或服务是否可访问", - "latency_ms": latency, - } - except Exception as e: - latency = int((time.time() - start) * 1000) - logger.error(f"Model connection test failed: {e}") - return { - "success": False, - "message": f"连接失败: {str(e)}", - "latency_ms": latency, - } - - -@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 - - -class UpdateModelRequest(BaseModel): - 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 - - -from typing import Optional +""" +大模型库管理 API(管理员) +""" +from fastapi import APIRouter, Depends, HTTPException, Query, Body +from sqlalchemy.orm import Session +from sqlalchemy import func +from typing import List, Optional +from pydantic import BaseModel +from ...database import get_db, SessionLocal +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 +import httpx +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 + + +# ============================================================ +# 协议预设列表 +# ============================================================ + +PROTOCOLS = [ + {"id": "openai_compatible", "name": "OpenAI兼容", "base_path": "/v1", "description": "兼容OpenAI API格式的模型服务"}, + {"id": "new_api", "name": "New API", "base_path": "/v1", "description": "新一代API聚合平台"}, + {"id": "volcengine", "name": "火山引擎", "base_path": "/v1", "description": "字节跳动火山引擎模型服务"}, + {"id": "aliyun_bailian", "name": "阿里云百炼", "base_path": "/v1", "description": "阿里云百炼大模型平台"}, + {"id": "ollama", "name": "Ollama", "base_path": "/v1", "description": "本地Ollama模型服务"}, + {"id": "xiaoma", "name": "小马算力", "base_path": "/v1", "description": "小马算力模型服务"}, +] + +# ============================================================ +# 模型列表(用于 fetch-models 的回退数据) +# ============================================================ + +FETCHABLE_MODELS = [ + {"id": "gpt-4o", "name": "GPT-4o", "capabilities": ["text", "vision", "tool"], "pricing": "paid"}, + {"id": "gpt-4o-mini", "name": "GPT-4o Mini", "capabilities": ["text", "vision"], "pricing": "paid"}, + {"id": "deepseek-v3", "name": "DeepSeek-V3", "capabilities": ["text", "thinking"], "pricing": "free"}, + {"id": "deepseek-r1", "name": "DeepSeek-R1", "capabilities": ["text", "thinking"], "pricing": "paid"}, + {"id": "qwen-max", "name": "Qwen-Max", "capabilities": ["text", "tool"], "pricing": "free"}, + {"id": "qwen-vl", "name": "Qwen-VL", "capabilities": ["text", "vision"], "pricing": "paid"}, + {"id": "whisper", "name": "Whisper", "capabilities": ["audio"], "pricing": "paid"}, + {"id": "dalle", "name": "DALL-E", "capabilities": ["image"], "pricing": "paid"}, +] + +# 支持通过 API 获取模型列表的协议类型 +FETCHABLE_PROTOCOLS = {"openai_compatible", "new_api", "volcengine", "aliyun_bailian", "ollama", "xiaoma"} + + +# ============================================================ +# 辅助函数 +# ============================================================ + +def provider_to_dict(p: ModelProvider) -> dict: + return { + "id": p.id, + "name": p.name, + "protocol": p.protocol, + "api_key": p.api_key, + "api_base_url": p.api_base_url, + "apiKey": p.api_key, + "apiBaseUrl": p.api_base_url, + "created_at": p.created_at.isoformat() if p.created_at else None, + "updated_at": p.updated_at.isoformat() if p.updated_at else None, + "createdAt": p.created_at.isoformat() if p.created_at else None, + "updatedAt": p.updated_at.isoformat() if p.updated_at else None, + } + + +def model_to_dict(m: AIModel) -> dict: + return { + "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, + "timeout": m.timeout, + "status": m.status, + "price_per_million_tokens": float(m.price_per_million_tokens) if m.price_per_million_tokens else 0, + "is_enabled": m.is_enabled, + "is_default": bool(m.is_default), + "total_calls": m.total_calls or 0, + "total_tokens": m.total_tokens or 0, + "created_at": m.created_at.isoformat() if m.created_at else None, + "updated_at": m.updated_at.isoformat() if m.updated_at else None, + } + + +# ============================================================ +# 协议列表 +# ============================================================ + +@router.get("/protocols") +async def get_protocols( + admin_user=Depends(get_current_admin), +): + """获取支持的协议列表""" + 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 + ] + + +# ============================================================ +# 模型供应商管理 +# ============================================================ + +@router.get("/providers") +async def get_providers( + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取模型供应商列表""" + providers = db.query(ModelProvider).order_by( + ModelProvider.created_at.desc() + ).all() + return [provider_to_dict(p) for p in providers] + + +@router.post("/providers") +async def create_provider( + name: str = Body(...), + protocol: str = Body(...), + api_key: str = Body(default=""), + api_base_url: str = Body(default=None), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """添加模型供应商""" + # 验证协议 + valid_protocols = [p["id"] for p in PROTOCOLS] + if protocol not in valid_protocols: + raise HTTPException(status_code=400, detail=f"无效的协议类型,支持: {', '.join(valid_protocols)}") + + provider = ModelProvider( + id=str(uuid.uuid4()), + name=name, + protocol=protocol, + api_key=api_key, + api_base_url=api_base_url, + ) + db.add(provider) + db.commit() + db.refresh(provider) + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="create_provider", + target_type="model_provider", + target_id=provider.id, + detail=f"添加模型供应商 {name} (协议: {protocol})", + ) + + return {"message": "供应商已创建", "id": provider.id, "provider": provider_to_dict(provider)} + + +@router.put("/providers/{id}") +async def update_provider( + id: str, + name: str = Body(default=None), + protocol: str = Body(default=None), + api_key: str = Body(default=None), + api_base_url: str = Body(default=None), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """更新模型供应商""" + provider = db.query(ModelProvider).filter(ModelProvider.id == id).first() + if not provider: + raise HTTPException(status_code=404, detail="供应商不存在") + + if name is not None: + provider.name = name + if protocol is not None: + valid_protocols = [p["id"] for p in PROTOCOLS] + if protocol not in valid_protocols: + raise HTTPException(status_code=400, detail=f"无效的协议类型") + provider.protocol = protocol + if api_key is not None: + provider.api_key = api_key + if api_base_url is not None: + provider.api_base_url = api_base_url + provider.updated_at = datetime.now() + db.commit() + db.refresh(provider) + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="update_provider", + target_type="model_provider", + target_id=id, + detail=f"更新模型供应商 {provider.name}", + ) + + return {"message": "供应商已更新", "provider": provider_to_dict(provider)} + + +@router.delete("/providers/{id}") +async def delete_provider( + id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """删除模型供应商""" + provider = db.query(ModelProvider).filter(ModelProvider.id == id).first() + if not provider: + raise HTTPException(status_code=404, detail="供应商不存在") + + # 检查是否有模型关联 + model_count = db.query(func.count(AIModel.id)).filter( + AIModel.provider_id == id + ).scalar() + if model_count > 0: + raise HTTPException(status_code=400, detail=f"该供应商下有 {model_count} 个模型,请先删除模型") + + db.delete(provider) + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="delete_provider", + target_type="model_provider", + target_id=id, + detail=f"删除模型供应商 {provider.name}", + ) + + return {"message": "供应商已删除"} + + +@router.post("/providers/{id}/fetch-models") +async def fetch_models( + id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取供应商可用模型列表(优先从真实API获取)""" + provider = db.query(ModelProvider).filter(ModelProvider.id == id).first() + if not provider: + raise HTTPException(status_code=404, detail="供应商不存在") + + logger = logging.getLogger(__name__) + logger.info(f"Fetching models for provider {provider.name} (protocol: {provider.protocol})") + + api_base = (provider.api_base_url or "").rstrip("/") + api_key = provider.api_key or "" + + # 尝试从真实API获取模型列表(只要有 api_base 就尝试调用,api_key 可选) + if api_base and provider.protocol in FETCHABLE_PROTOCOLS: + try: + async with httpx.AsyncClient(timeout=15) as client: + headers = {} + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + resp = await client.get(f"{api_base}/v1/models", headers=headers) + if resp.status_code == 200: + data = resp.json() + models_list = data.get("data", []) + if models_list: + result = [] + for m in models_list: + mid = m.get("id", "") + if not mid: + continue + result.append({ + "id": mid, + "name": mid, + "capabilities": ["text"], + "pricing": "paid", + }) + logger.info(f"Successfully fetched {len(result)} models from provider API") + return {"models": result, "source": "api"} + else: + logger.warning(f"Provider API returned empty model list") + return {"models": [], "source": "api", "message": "API返回的模型列表为空"} + elif resp.status_code == 401 or resp.status_code == 403: + logger.warning(f"Provider API returned {resp.status_code}: auth error") + return {"models": [], "source": "error", "message": "API认证失败,请检查API Key是否正确"} + else: + logger.warning(f"Provider API returned status {resp.status_code}: {resp.text[:200]}") + return {"models": [], "source": "error", "message": f"API请求失败(HTTP {resp.status_code}),请检查API地址和密钥"} + except httpx.TimeoutException: + logger.warning(f"Timeout fetching models from provider API: {api_base}") + return {"models": [], "source": "error", "message": "请求API超时,请检查API地址是否正确"} + except Exception as e: + logger.warning(f"Failed to fetch models from provider API: {e}") + return {"models": [], "source": "error", "message": f"请求API失败: {str(e)}"} + + # 回退:返回固定模型列表(供无 API 的离线环境使用) + logger.info(f"Using fallback model list for provider {provider.name}") + return {"models": FETCHABLE_MODELS, "source": "fallback"} + + +class AddModelsRequest(BaseModel): + model_ids: List[str] + + +@router.post("/providers/{id}/add-models") +async def add_models( + id: str, + req: AddModelsRequest, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """批量添加模型到供应商""" + provider = db.query(ModelProvider).filter(ModelProvider.id == id).first() + if not provider: + raise HTTPException(status_code=404, detail="供应商不存在") + + # 找到协议对应的 base_path + protocol_info = next((p for p in PROTOCOLS if p["id"] == provider.protocol), None) + base_path = protocol_info["base_path"] if protocol_info else "/v1" + + added = [] + for model_id in req.model_ids: + # 在 FETCHABLE_MODELS 中查找额外信息 + model_info = next((m for m in FETCHABLE_MODELS if m["id"] == model_id), None) + + # 使用 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( + AIModel.provider_id == id, + AIModel.name == model_name + ).first() + if existing: + continue + + # 自动生成 endpoint + if provider.api_base_url: + endpoint = f"{provider.api_base_url.rstrip('/')}{base_path}/chat/completions" + else: + endpoint = "" + + model = AIModel( + 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"], + timeout=60, + status="free" if (model_info and model_info.get("pricing") == "free") else "paid", + price_per_million_tokens=0, + is_enabled=True, + ) + db.add(model) + added.append(model_name) + + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="add_models", + target_type="model_provider", + target_id=id, + detail=f"为供应商 {provider.name} 批量添加 {len(added)} 个模型", + ) + + return {"message": f"成功添加 {len(added)} 个模型", "added": added} + + +# ============================================================ +# 模型管理 +# ============================================================ + +@router.get("/") +async def get_models( + provider_id: str = Query(None), + status: str = Query(None, description="模型状态筛选: free/paid"), + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取模型列表(管理员)""" + query = db.query(AIModel) + + if provider_id: + query = query.filter(AIModel.provider_id == provider_id) + if status: + query = query.filter(AIModel.status == status) + + total = query.count() + models = query.order_by(AIModel.created_at.desc()).offset( + (page - 1) * page_size + ).limit(page_size).all() + + return { + "total": total, + "page": page, + "page_size": page_size, + "items": [model_to_dict(m) for m in models], + } + + +@router.post("/") +async def create_model( + data: ModelCreateRequest, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """添加模型""" + # 验证供应商存在 + 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=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() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="create_model", + target_type="model", + target_id=model.id, + detail=f"添加模型 {name}", + ) + + 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, + data: ModelUpdateRequest, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """更新模型""" + model = db.query(AIModel).filter(AIModel.id == id).first() + if not model: + raise HTTPException(status_code=404, detail="模型不存在") + + 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 = 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() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="update_model", + target_type="model", + target_id=id, + detail=f"更新模型 {model.name}", + ) + + return {"message": "模型已更新"} + + +@router.delete("/{id}") +async def delete_model( + id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """删除模型""" + model = db.query(AIModel).filter(AIModel.id == id).first() + if not model: + raise HTTPException(status_code=404, detail="模型不存在") + + model_name = model.name + db.delete(model) + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="delete_model", + target_type="model", + target_id=id, + detail=f"删除模型 {model_name}", + ) + + return {"message": "模型已删除"} + + +@router.put("/{id}/toggle") +async def toggle_model( + id: str, + is_enabled: bool = Query(...), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """开关模型状态""" + model = db.query(AIModel).filter(AIModel.id == id).first() + if not model: + raise HTTPException(status_code=404, detail="模型不存在") + + model.is_enabled = is_enabled + model.updated_at = datetime.now() + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="toggle_model", + target_type="model", + target_id=id, + detail=f"将模型 {model.name} {'启用' if is_enabled else '禁用'}", + ) + + return {"message": f"模型已{'启用' if is_enabled else '禁用'}"} + + +@router.put("/{id}/set-default") +async def set_default_model( + id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """设为全局默认模型(取消其他模型的默认状态)""" + model = db.query(AIModel).filter(AIModel.id == id).first() + if not model: + raise HTTPException(status_code=404, detail="模型不存在") + + # 取消所有模型的默认状态 + db.query(AIModel).filter(AIModel.is_default == True).update({"is_default": False}) + # 设置当前模型为默认 + model.is_default = True + 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="model", + target_id=id, + detail=f"设置模型 {model.name} 为全局默认", + ) + + return {"message": f"已将 {model.name} 设为全局默认模型"} + + +@router.get("/stats") +async def get_model_stats( + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取模型调用统计""" + total_calls = db.query(func.sum(AIModel.total_calls)).scalar() or 0 + total_tokens = db.query(func.sum(AIModel.total_tokens)).scalar() or 0 + + return { + "total_calls": int(total_calls), + "total_tokens": int(total_tokens), + } + + +# ============================================================ +# 按能力指定默认模型(供技能 / 应用 / IM Bot 自动匹配) +# ============================================================ +# (按能力默认模型路由已移至 /{id} 路由之前,避免路径参数冲突) + + + + +# ============================================================ +# 模型连接测试 API (Phase 2) +# ============================================================ + +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), + db: Session = Depends(get_db), +): + """ + 测试指定模型的连接是否正常 + 支持多模态: text/image/audio/vision 自动检测模型类型并使用对应API测试 + """ + import time + + logger = logging.getLogger(__name__) + logger.info(f"Testing connection for model_id: {model_id}") + + ai_model = db.query(AIModel).filter(AIModel.id == model_id).first() + if not ai_model: + raise HTTPException(status_code=404, detail="模型不存在") + + provider = db.query(ModelProvider).filter(ModelProvider.id == ai_model.provider_id).first() + if not provider: + raise HTTPException(status_code=404, detail="模型供应商不存在") + + 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 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 + headers = { + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + } + 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, + } + # 视觉 / 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: + 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) + + # ============================================================ + # 图像生成模型 → 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"图像生成成功 ({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: + resp = await client.post(asr_url, files=files, headers=asr_headers) + except Exception: + # 某些实现要求 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"未知模型类型: {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) + logger.error(f"Model connection test failed: {e}") + return { + "success": False, + "message": f"连接失败: {str(e)}", + "latency_ms": latency, + "model_type": model_type, + } + + +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 + 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 + diff --git a/backend/python/app/api/chat.py b/backend/python/app/api/chat.py index 9405a6b..5df16c5 100644 --- a/backend/python/app/api/chat.py +++ b/backend/python/app/api/chat.py @@ -1,781 +1,1489 @@ -from fastapi import APIRouter, Depends, HTTPException, Query -from sqlalchemy.orm import Session -from sqlalchemy import and_, desc, asc -from pydantic import BaseModel, Field -from typing import Optional -import uuid -from datetime import datetime -import re -import random -import json -import httpx -import os -import logging - -from ..database import get_db, SessionLocal -from ..models import ( - User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog, - KnowledgeBase, KnowledgeDocument, EvolutionSkill, - EvolutionTrajectory, - gen_uuid -) -from ..auth import get_current_user -from ..middleware.log_service import LogService - -router = APIRouter() - -logger = logging.getLogger(__name__) - - -class CreateConversationRequest(BaseModel): - title: str = Field(default="新对话") - model_id: Optional[str] = None - - -class UpdateConversationRequest(BaseModel): - title: Optional[str] = None - status: Optional[str] = None # active, archived - pinned: Optional[bool] = None - - -class SendMessageRequest(BaseModel): - content: str - model_id: Optional[str] = None - skill_ids: Optional[list] = None - agent_ids: Optional[list] = None - knowledge_base_ids: Optional[list] = None - use_tools: Optional[bool] = False - - -@router.get("/conversations") -def list_conversations( - status: Optional[str] = Query(None), - search: Optional[str] = Query(None), - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - query = db.query(Conversation).filter(Conversation.user_id == current_user.id) - - if status: - query = query.filter(Conversation.status == status) - else: - # 默认不返回已归档的 - query = query.filter(Conversation.status != "archived") - - if search: - query = query.filter(Conversation.title.ilike(f"%{search}%")) - - # pinned优先,然后按更新时间倒序 - query = query.order_by(desc(Conversation.pinned), desc(Conversation.updated_at)) - - conversations = query.all() - result = [] - for conv in conversations: - result.append({ - "id": conv.id, - "title": conv.title, - "model_id": conv.model_id, - "status": conv.status, - "pinned": conv.pinned, - "knowledge_graph_count": conv.knowledge_graph_count or 0, - "created_at": str(conv.created_at) if conv.created_at else None, - "updated_at": str(conv.updated_at) if conv.updated_at else None, - }) - - return {"code": 0, "data": result, "message": "ok"} - - -@router.post("/conversations") -def create_conversation( - req: CreateConversationRequest, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - conv = Conversation( - id=gen_uuid(), - user_id=current_user.id, - title=req.title, - model_id=req.model_id, - status="active", - pinned=False, - knowledge_graph_count=0, - created_at=datetime.now(), - updated_at=datetime.now(), - ) - db.add(conv) - db.commit() - db.refresh(conv) - - return { - "code": 0, - "data": { - "id": conv.id, - "title": conv.title, - "model_id": conv.model_id, - "status": conv.status, - "pinned": conv.pinned, - "knowledge_graph_count": conv.knowledge_graph_count, - "created_at": str(conv.created_at) if conv.created_at else None, - "updated_at": str(conv.updated_at) if conv.updated_at else None, - }, - "message": "创建成功", - } - - -@router.put("/conversations/{conv_id}") -def update_conversation( - conv_id: str, - req: UpdateConversationRequest, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - conv = db.query(Conversation).filter( - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - ).first() - if not conv: - raise HTTPException(status_code=404, detail="对话不存在") - - if req.title is not None: - conv.title = req.title - if req.status is not None: - conv.status = req.status - if req.pinned is not None: - conv.pinned = req.pinned - conv.updated_at = datetime.now() - - db.commit() - db.refresh(conv) - - return {"code": 0, "data": { - "id": conv.id, "title": conv.title, "status": conv.status, "pinned": conv.pinned, - "updated_at": str(conv.updated_at), - }, "message": "更新成功"} - - -@router.delete("/conversations/{conv_id}") -def delete_conversation( - conv_id: str, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - conv = db.query(Conversation).filter( - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - ).first() - if not conv: - raise HTTPException(status_code=404, detail="对话不存在") - - # 软删除 - conv.status = "archived" - conv.updated_at = datetime.now() - db.commit() - - return {"code": 0, "message": "删除成功"} - - -@router.get("/conversations/{conv_id}/messages") -def get_messages( - conv_id: str, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - conv = db.query(Conversation).filter( - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - ).first() - if not conv: - raise HTTPException(status_code=404, detail="对话不存在") - - messages = ( - db.query(Message) - .filter(Message.conversation_id == conv_id) - .order_by(asc(Message.created_at)) - .all() - ) - - result = [] - for msg in messages: - result.append({ - "id": msg.id, - "conversation_id": msg.conversation_id, - "role": msg.role, - "content": msg.content, - "token_used": msg.token_used or 0, - "model_id": msg.model_id, - "used_skills": msg.used_skills or [], - "created_at": str(msg.created_at) if msg.created_at else None, - }) - - return {"code": 0, "data": result, "message": "ok"} - - -@router.post("/conversations/{conv_id}/messages") -async def send_message( - conv_id: str, - req: SendMessageRequest, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - conv = db.query(Conversation).filter( - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - ).first() - if not conv: - raise HTTPException(status_code=404, detail="对话不存在") - - # 保存用户消息 - user_msg = Message( - id=gen_uuid(), - conversation_id=conv_id, - role="user", - content=req.content, - token_used=len(req.content) // 2, # 粗略估算 - model_id=req.model_id or conv.model_id, - created_at=datetime.now(), - ) - db.add(user_msg) - - # 调用 AI 模型 API - user_id = current_user.id - model_id = req.model_id or (conv.model_id if hasattr(conv, 'model_id') else None) - prompt_tokens = 0 - completion_tokens = 0 - total_tokens = 0 - - ai_model = None - if model_id: - ai_model = db.query(AIModel).filter(AIModel.id == model_id).first() - - if ai_model and ai_model.endpoint: - provider = db.query(ModelProvider).filter(ModelProvider.id == ai_model.provider_id).first() - api_key = ai_model.api_key or (provider.api_key if provider else None) - api_url = ai_model.endpoint - - if not api_url and provider and provider.api_base_url: - api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions" - - if api_key and api_url: - try: - # 构建消息列表:先system prompt(含skills/agents上下文),再历史消息,再用户消息 - messages = [] - system_parts = ["你是AIFlow平台的AI助手。"] - - # 加载skills信息用于tool-call - skills_list = [] - function_definitions = [] - if req.skill_ids and len(req.skill_ids) > 0: - skills_list = db.query(EvolutionSkill).filter(EvolutionSkill.id.in_(req.skill_ids)).all() - if skills_list: - skill_descs = ", ".join([f"{s.skill_name}({s.category or 'general'})" for s in skills_list]) - system_parts.append(f"可用技能: {skill_descs}") - system_parts.append("重要: 当用户的请求可以使用上述技能处理时,你必须优先调用对应的工具函数(tool-call)来完成任务,而不是直接回复文字。") - - # 构建function definitions用于tool-call (OpenAI function calling协议) - if req.use_tools: - for s in skills_list: - func_name = s.skill_name.replace(" ", "_").replace("-", "_").lower() - _desc = s.description or f"{s.skill_name} - {s.category or 'general'}" - function_definitions.append({ - "name": func_name, - "description": f"调用技能「{s.skill_name}」处理用户请求。使用场景: {_desc}。当用户请求涉及{s.category or '通用'}类任务时优先调用此工具。", - "parameters": { - "type": "object", - "properties": { - "input": { - "type": "string", - "description": f"需要传递给{s.skill_name}的完整用户输入内容", - } - }, - "required": ["input"], - }, - }) - - if req.agent_ids and len(req.agent_ids) > 0: - system_parts.append(f"已选择Agent辅助处理任务") - if req.knowledge_base_ids and len(req.knowledge_base_ids) > 0: - kbs = db.query(KnowledgeBase).filter(KnowledgeBase.id.in_(req.knowledge_base_ids)).all() - if kbs: - kb_names = ", ".join([k.name for k in kbs]) - system_parts.append(f"参考知识库: {kb_names}") - - messages.append({"role": "system", "content": "; ".join(system_parts)}) - - # 加载最近的对话历史(最多10条) - history = db.query(Message).filter( - Message.conversation_id == conv_id - ).order_by(asc(Message.created_at)).limit(10).all() - for h in history: - messages.append({"role": h.role, "content": h.content}) - - messages.append({"role": "user", "content": req.content}) - - # Tool-call循环(最多3轮) - max_tool_rounds = 3 - ai_content = "" - for round_num in range(max_tool_rounds): - request_body = { - "model": ai_model.name, - "messages": messages, - "max_tokens": 2048, - } - if function_definitions and req.use_tools: - request_body["tools"] = [ - {"type": "function", "function": fd} - for fd in function_definitions - ] - request_body["tool_choice"] = "auto" - - async with httpx.AsyncClient(timeout=ai_model.timeout or 60) as client: - resp = await client.post( - api_url, - headers={ - "Authorization": f"Bearer {api_key}", - "Content-Type": "application/json" - }, - json=request_body - ) - if resp.status_code != 200: - if round_num == 0: - ai_content = f"AI模型调用失败 (HTTP {resp.status_code}),请检查模型配置。" - break - - data = resp.json() - choice = data.get("choices", [{}])[0] - message = choice.get("message", {}) - - # 记录token使用量 - usage = data.get("usage", {}) - prompt_tokens += usage.get("prompt_tokens", 0) - completion_tokens += usage.get("completion_tokens", 0) - total_tokens += usage.get("total_tokens", 0) - - # 检查是否有tool_calls - tool_calls = message.get("tool_calls", []) - if tool_calls and req.use_tools: - # 处理tool call:将assistant消息(含tool_calls)加入上下文 - messages.append({ - "role": "assistant", - "content": message.get("content") or "", - "tool_calls": tool_calls, - }) - - # 执行每个tool call - for tc in tool_calls: - func_name = tc.get("function", {}).get("name", "") - func_args_str = tc.get("function", {}).get("arguments", "{}") - try: - func_args = json.loads(func_args_str) - except Exception: - func_args = {"input": func_args_str} - - # 查找匹配的skill并生成tool结果 - tool_result = _execute_skill_tool(db, func_name, func_args, skills_list) - messages.append({ - "role": "tool", - "tool_call_id": tc.get("id", f"call_{round_num}"), - "content": tool_result, - }) - - # 继续下一轮,让模型处理tool结果 - continue - else: - # 没有tool_calls,正常回复 - ai_content = message.get("content", "") - if not ai_content and round_num > 0: - ai_content = "AI模型已处理工具调用结果,但返回了空响应。" - break - - if not ai_content: - ai_content = "AI模型返回了空响应,请稍后重试。" - except Exception as e: - import logging - logging.getLogger(__name__).warning(f"AI API调用异常: {e}") - ai_content = "AI服务暂时不可用,请稍后重试或检查模型API配置。" - else: - ai_content = "该模型缺少API密钥或端点配置,请在后台管理-大模型库中配置API Key和Endpoint。" - else: - ai_content = "请选择一个已配置的AI模型进行对话。" - else: - ai_content = "请先在聊天设置中选择一个AI模型。" - - ai_token_count = len(ai_content) // 2 - ai_msg = Message( - id=gen_uuid(), - conversation_id=conv_id, - role="assistant", - content=ai_content, - token_used=ai_token_count, - model_id=req.model_id or conv.model_id, - created_at=datetime.now(), - ) - db.add(ai_msg) - - # 基于请求中skill_ids关联用户真实技能作为used_skills - used_skills = [] - if req.skill_ids and len(req.skill_ids) > 0: - selected_skills = db.query(EvolutionSkill).filter( - EvolutionSkill.id.in_(req.skill_ids), - EvolutionSkill.user_id == current_user.id - ).all() - if selected_skills: - used_skills = [s.skill_name for s in selected_skills] - ai_msg.used_skills = used_skills - # 更新每个被使用技能的use_count和last_used_at - now = datetime.now() - for s in selected_skills: - s.use_count = (s.use_count or 0) + 1 - s.last_used_at = now - - # 更新对话统计 - conv.knowledge_graph_count = (conv.knowledge_graph_count or 0) + 1 - conv.updated_at = datetime.now() - - # 记录Token使用 - token_usage = TokenUsageLog( - id=gen_uuid(), - user_id=current_user.id, - model_id=req.model_id or conv.model_id, - model_name=ai_model.name if model_id and ai_model else None, - conversation_id=conv_id, - prompt_tokens=prompt_tokens or len(req.content) // 2, - completion_tokens=completion_tokens or ai_token_count, - total_tokens=total_tokens or (len(req.content) // 2 + ai_token_count), - created_at=datetime.now(), - ) - db.add(token_usage) - - # 记录 AI 进化轨迹 (每次对话产生一条轨迹,供PipelineScheduler处理) - # 去重:检查该对话是否已有trajectory记录,避免重复创建 - try: - existing = db.query(EvolutionTrajectory).filter( - EvolutionTrajectory.conversation_id == conv_id - ).first() - if not existing: - trajectory = EvolutionTrajectory( - id=gen_uuid(), - user_id=current_user.id, - conversation_id=conv_id, - status="discovered", - token_used=ai_token_count, - created_at=datetime.now(), - ) - db.add(trajectory) - db.flush() # 确保trajectory先写入,再启动pipeline - - # 自动启动用户进化Pipeline(如果未运行) - try: - from app.services.evolution_service import EvolutionWatcherManager - watcher = EvolutionWatcherManager() - status = watcher.get_user_status(current_user.id, lambda: SessionLocal()) - if not status.get("scheduler_running"): - watcher.start_user_evolution(current_user.id, lambda: SessionLocal()) - logger.info(f"[Evolution] 自动启动用户 {current_user.id} 的进化Pipeline") - except Exception as evolution_err: - logger.warning(f"[Evolution] 自动启动Pipeline失败: {evolution_err}") - except Exception: - pass # 轨迹记录失败不影响主流程 - - db.commit() - db.refresh(user_msg) - db.refresh(ai_msg) - - return { - "code": 0, - "data": { - "user_message": { - "id": user_msg.id, "role": user_msg.role, "content": user_msg.content, - "token_used": user_msg.token_used, "created_at": str(user_msg.created_at), - }, - "assistant_message": { - "id": ai_msg.id, "role": ai_msg.role, "content": ai_msg.content, - "token_used": ai_msg.token_used, "created_at": str(ai_msg.created_at), - "used_skills": used_skills, - }, - }, - "message": "发送成功", - } - - -@router.post("/conversations/{conv_id}/clear") -def clear_conversation( - conv_id: str, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - conv = db.query(Conversation).filter( - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - ).first() - if not conv: - raise HTTPException(status_code=404, detail="对话不存在") - - # 清空消息(物理删除消息记录) - db.query(Message).filter( - Message.conversation_id == conv_id - ).delete() - - conv.knowledge_graph_count = 0 - conv.updated_at = datetime.now() - db.commit() - - return {"code": 0, "message": "对话上下文已清空"} - - -@router.post("/conversations/{conv_id}/to-knowledge-graph") -def convert_to_knowledge_graph( - conv_id: str, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - """将对话内容转换为知识图谱:提取消息内容、创建知识库、构建知识图谱""" - conv = db.query(Conversation).filter( - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - ).first() - if not conv: - raise HTTPException(status_code=404, detail="对话不存在") - - # 获取该对话的所有消息 - messages = ( - db.query(Message) - .filter(Message.conversation_id == conv_id) - .order_by(asc(Message.created_at)) - .all() - ) - - if not messages: - raise HTTPException(status_code=400, detail="对话中没有消息,无法生成知识图谱") - - # 1. 自动创建一个知识库(名为"对话图谱-{对话标题}") - kb_name = f"对话图谱-{conv.title}" - kb = KnowledgeBase( - id=gen_uuid(), - user_id=current_user.id, - name=kb_name, - description=f"从对话「{conv.title}」自动生成的知识图谱", - directory_path="/", - document_count=0, - created_at=datetime.now(), - updated_at=datetime.now(), - ) - db.add(kb) - db.flush() - - # 2. 将每条消息内容作为文档存入知识库 - doc_count = 0 - for idx, msg in enumerate(messages): - # 提取消息中的简明摘要作为文档文件名 - role_label = "用户" if msg.role == "user" else "AI" - preview = re.sub(r'[\\n\\r\\t]+', ' ', msg.content[:40]).strip() - filename = f"msg_{idx+1:03d}_{role_label}_{preview}.txt" - - doc = KnowledgeDocument( - id=gen_uuid(), - knowledge_base_id=kb.id, - filename=filename, - file_type="txt", - file_size=len(msg.content.encode('utf-8')), - status="completed", - chunk_count=max(1, len(msg.content) // 500 + 1), - created_at=datetime.now(), - updated_at=datetime.now(), - ) - db.add(doc) - doc_count += 1 - - kb.document_count = doc_count - db.flush() - - # 3. 调用知识图谱构建逻辑(类似 knowledge.py 中的 build-graph) - documents = db.query(KnowledgeDocument).filter( - and_( - KnowledgeDocument.knowledge_base_id == kb.id, - KnowledgeDocument.status == "completed", - ) - ).all() - - graph_nodes = [] - graph_edges = [] - - doc_entity_types = [ - {"type": "concept", "label": "核心概念"}, - {"type": "entity", "label": "关键实体"}, - {"type": "attribute", "label": "属性特征"}, - {"type": "relation", "label": "关联关系"}, - ] - - for idx, doc in enumerate(documents): - doc_node_id = f"doc-{doc.id[:8]}" - graph_nodes.append({ - "id": doc_node_id, - "label": doc.filename, - "type": "document", - "size": 18, - "group": "document", - }) - - filename_base = doc.filename.rsplit(".", 1)[0] - words = filename_base.replace("_", " ").replace("-", " ").split() - - for wi, word in enumerate(words): - if len(word) >= 2: - entity_type = doc_entity_types[wi % len(doc_entity_types)] - node_id = f"ent-{doc.id[:8]}-{wi}" - graph_nodes.append({ - "id": node_id, - "label": word, - "type": entity_type["type"], - "size": 10 + (wi % 8), - "group": entity_type["type"], - }) - graph_edges.append({ - "source": doc_node_id, - "target": node_id, - "label": entity_type["label"], - "weight": 0.5 + (0.5 / (wi + 1)), - }) - - # 在文档之间建立关联 - for i in range(len(documents)): - for j in range(i + 1, len(documents)): - fi = documents[i].filename.lower().rsplit(".", 1)[0] - fj = documents[j].filename.lower().rsplit(".", 1)[0] - words_i = set(fi.replace("_", " ").replace("-", " ").split()) - words_j = set(fj.replace("_", " ").replace("-", " ").split()) - common = words_i & words_j - if common and len(common) > 0: - graph_edges.append({ - "source": f"doc-{documents[i].id[:8]}", - "target": f"doc-{documents[j].id[:8]}", - "label": f"共有关键词: {', '.join(list(common)[:3])}", - "weight": min(0.9, 0.3 * len(common)), - }) - - graph_data = { - "nodes": graph_nodes, - "edges": graph_edges, - "stats": { - "total_nodes": len(graph_nodes), - "total_edges": len(graph_edges), - "document_count": len(documents), - "entity_types": {et["type"]: sum(1 for n in graph_nodes if n.get("group") == et["type"]) - for et in doc_entity_types}, - }, - } - - # 更新知识库的知识图谱计数 - kb.knowledge_graph_count = len(graph_nodes) - kb.updated_at = datetime.now() - - # 更新对话的知识图谱计数 - conv.knowledge_graph_count = (conv.knowledge_graph_count or 0) + 1 - conv.updated_at = datetime.now() - - db.commit() - - LogService.log_operation( - user_id=current_user.id, username=current_user.username, action="convert_to_graph", - target="conversation", target_id=conv_id, - detail=f"将对话 {conv.title} 转换为知识图谱 {kb.name},共 {len(graph_nodes)} 节点、{len(graph_edges)} 边", - ) - - return { - "code": 0, - "data": { - "knowledge_base_id": kb.id, - "knowledge_base_name": kb.name, - "graph": graph_data, - }, - "message": f"知识图谱构建完成,共 {len(graph_nodes)} 个节点、{len(graph_edges)} 条边", - } - - -@router.get("/models") -def get_models(db: Session = Depends(get_db)): - models = db.query(AIModel).filter(AIModel.is_enabled == True).all() - result = [] - for m in models: - provider = db.query(ModelProvider).filter(ModelProvider.id == m.provider_id).first() - result.append({ - "id": m.id, - "name": m.name, - "provider_id": m.provider_id, - "provider_name": provider.name if provider else "", - "status": m.status, - "price_per_million_tokens": float(m.price_per_million_tokens or 0), - "is_enabled": m.is_enabled, - "is_default": bool(m.is_default) if hasattr(m, 'is_default') else False, - "total_calls": m.total_calls or 0, - "total_tokens": m.total_tokens or 0, - "created_at": str(m.created_at) if m.created_at else None, - }) - # 默认模型排在最前面 - result.sort(key=lambda x: (0 if x["is_default"] else 1, x["name"])) - return {"code": 0, "data": result, "message": "ok"} - - -def _execute_skill_tool(db: Session, func_name: str, args: dict, skills_list: list) -> str: - """执行tool-call中的skill工具调用,返回工具执行结果""" - # 根据func_name匹配对应的skill(支持新旧两种命名方式) - matched_skill = None - for s in skills_list: - expected_name = s.skill_name.replace(" ", "_").replace("-", "_").lower() - if func_name == expected_name or func_name == f"skill_{s.id[:8]}": - matched_skill = s - break - - skill_name = matched_skill.skill_name if matched_skill else func_name - input_text = args.get("input", "") - - # 根据skill类型生成模拟执行结果 - skill_type = matched_skill.skill_type if matched_skill else "unknown" - - if skill_type == "writer": - result = ( - f"技能 '{skill_name}' (写作类) 执行结果:\n" - f"根据输入 '{input_text[:100]}' 生成以下内容:\n" - f"---\n" - f"# 生成文本\n\n这是一篇关于 {input_text[:50]} 的专业文章。" - f"通过深入分析和研究,我们得出以下核心观点:首先,该主题在当前领域具有重要的实践价值;" - f"其次,从技术发展趋势来看,该方向有望在未来3-5年内实现重大突破。" - f"\n\n## 关键建议\n1. 加强基础研究投入\n2. 建立标准化流程\n3. 推动产业协同创新" - f"\n\n*由技能 '{skill_name}' 自动生成*" - ) - elif skill_type == "analyzer": - result = ( - f"技能 '{skill_name}' (分析类) 执行结果:\n" - f"对输入内容进行了以下分析:\n" - f"---\n" - f"分析主题: {input_text[:80]}\n" - f"情感倾向: 中性偏积极\n" - f"关键词提取: {', '.join([w for w in input_text[:80].replace(',', ',').split(',')[:5]])}\n" - f"关键实体: 识别到 {max(1, len(input_text) // 50)} 个实体\n" - f"语义密度: {'高' if len(input_text) > 50 else '中'}\n" - f"建议行动: 进一步细化分析维度,增加对比数据" - ) - elif skill_type == "translator": - result = ( - f"技能 '{skill_name}' (翻译类) 执行结果:\n" - f"输入: {input_text[:100]}\n" - f"英文翻译: [This is the English translation of: {input_text[:80]}...]" - f"\n日文翻译: [これは次の内容の日本語訳です: {input_text[:50]}...]" - ) - elif skill_type == "summarizer": - result = ( - f"技能 '{skill_name}' (摘要类) 执行结果:\n" - f"原文长度: {len(input_text)} 字符\n" - f"摘要:\n" - f"本文围绕 {input_text[:50]} 展开讨论,核心观点包括:技术革新驱动效率提升、" - f"数据标准化是行业发展的关键瓶颈、以及跨平台协作将成为未来趋势。" - f"\n关键数字: 预计增长率为15-20%,实施周期为6-12个月。" - ) - else: - result = ( - f"技能 '{skill_name}' ({skill_type}) 执行结果:\n" - f"已处理输入: {input_text[:200]}\n" - f"处理状态: 成功\n" - f"输出: 基于技能 '{skill_name}' 的处理逻辑,已完成对输入内容的分析处理。" - f"建议结合具体业务场景进一步优化结果。" - ) - - # 更新skill使用计数 - if matched_skill: - matched_skill.use_count = (matched_skill.use_count or 0) + 1 - matched_skill.last_used_at = datetime.now() - - return result +""" +增强版 Chat API — Phase 2.5 +包含: Skill ID统一(MarketApp映射), 真实Skill执行, Agent工作流注入, + 多轮上下文保持, BM25 KB检索, Function Calling循环增强 +""" +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy.orm import Session +from sqlalchemy import and_, desc, asc, func +from pydantic import BaseModel, Field +from typing import Optional +import uuid +from datetime import datetime, timezone, timedelta +import re +import json +import asyncio +import math +import httpx +import os +import logging + +from ..database import get_db, SessionLocal +from ..models import ( + User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog, + KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill, + EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation, + gen_uuid +) +from ..auth import get_current_user +from ..middleware.log_service import LogService +from ..capability_registry import resolve_auto_model + +router = APIRouter() + +logger = logging.getLogger(__name__) + + +def detect_media_intent(text: str) -> str | None: + """根据用户输入判断是否为 图像/视频/语音 生成请求,返回对应能力或 None。 + + 用于 Auto 模式:用户未指定具体模型时,依据意图自动匹配平台各类能力的默认模型。 + 疑问句(含「吗/呢/怎么/能否」或结尾问号)不触发,避免「你能生成视频吗」被误判。 + """ + if not text: + return None + t = text.lower().strip() + # 疑问句不触发媒体生成 + if t.endswith(("?", "?")) or any(q in t for q in ("吗", "呢", "怎么", "如何", "能否", "是否可以", "可以吗", "会不会", "why", "how")): + return None + video_kw = ("视频", "短片", "影片", "vide", "动画视频", "一段视频", "做个视频", "生成视频") + image_kw = ("图片", "图像", "插画", "海报", "照片", "imag", "画一张", "画个", "生成图片", "一张图", "一幅画", "生成一幅") + audio_kw = ("语音", "配音", "朗读", "播报", "念一段", "有声", "tts", "音频", "生成语音") + if any(k in t for k in video_kw): + return "video" + if any(k in t for k in image_kw): + return "image" + if any(k in t for k in audio_kw): + return "audio" + return None + +# CST timezone (UTC+8) for consistent time display +CST_TZ = timezone(timedelta(hours=8)) + + +def cst_now(): + """Get current time in CST timezone""" + return datetime.now(CST_TZ) + + +def _html_escape(s: str) -> str: + return (str(s).replace("&", "&").replace("<", "<").replace(">", ">") + .replace('"', """).replace("'", "'")) + + +async def _generate_media(db, ai_model, prompt: str, user_id: str = None) -> str: + """图像/视频生成模型统一入口:按 capabilities 调用 AGNES 等媒体生成 API。 + 返回可直接在 AI 聊天中渲染的富文本(含 /