From 24ed907ba51e5eb80af97a91dcffd5a9ad7a882b Mon Sep 17 00:00:00 2001 From: QihuAI Design Date: Thu, 16 Jul 2026 20:59:42 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0MCP=E5=B7=A5=E5=85=B7?= =?UTF-8?q?=E8=B0=83=E7=94=A8=E9=93=BE=E8=B7=AF(streamable=5Fhttp)=20-=20c?= =?UTF-8?q?hat=E4=B8=ADMCP=E4=BD=9C=E4=B8=BAfunction=20calling=E7=9C=9F?= =?UTF-8?q?=E5=AE=9E=E8=B0=83=E7=94=A8tools/call?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/python/app/api/chat.py | 12343 +++++++++++++++++++++++------ backend/python/app/mcp_client.py | 187 + 2 files changed, 10095 insertions(+), 2435 deletions(-) create mode 100644 backend/python/app/mcp_client.py diff --git a/backend/python/app/api/chat.py b/backend/python/app/api/chat.py index 238725e..1624b4f 100644 --- a/backend/python/app/api/chat.py +++ b/backend/python/app/api/chat.py @@ -1,2435 +1,9908 @@ -""" -增强版 Chat API — Phase 2.5 -包含: Skill ID统一(MarketApp映射), 真实Skill执行, Agent工作流注入, - 多轮上下文保持, BM25 KB检索, Function Calling循环增强 -""" -from fastapi import APIRouter, Depends, HTTPException, Query, Response -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 ..platform_knowledge import get_platform_knowledge -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 = ("图片", "图像", "插画", "海报", "照片", "头像", "壁纸", "漫画", "表情包", "logo", "图标", - "imag", "画一张", "画张", "画一只", "画只", "画个", "画幅", - "来张", "来一张", "来幅", "发张图", "发个图", - "生成图片", "生成一张", "生成一幅", "生成图", "图像生成", - "做一张", "做张图", "做张", "出张图", "出一张图", - "帮我画", "给我画", "请画", "绘制", "作画", "ai绘画", "ai画图", - "画个图", "画张图", "画一幅图", "一张图", "一幅画", "生成一幅", - "画画", "画一下", "画出来", "生成个图", "生成一张图", "生成个图片", - "做图片", "做个图", "制图", "整张图", "整张图出来", "搞张图", - "帮我生成图片", "帮我生成图像", "给我生成图片", "给我生成图像", - "生成一张图片", "生成一个图片", "制作一张图", "制作一张图片", - "来一张图片", "画一张图片", "画一幅", "画一副") - 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 - - -def detect_ambiguous_creative_intent(text: str): - """检测「模糊创作意图」:用户使用了创作类动词(制作/创作/做/生成/画/写/设计…), - 但未明确产出形式(文案/图片/视频/其他)。此时应与用户确认方向,而不是直接调用大模型 - (某些模型会对这类开放式创作请求返回 "high risk" 拒答)。 - - 返回 None 表示无需确认;否则返回 {"options": [...]}。 - """ - if not text: - return None - t = text.strip() - tl = t.lower() - # 已有明确媒体意图(图像/视频/语音)则无需确认 - if detect_media_intent(t): - return None - # 疑问句不触发 - if tl.endswith(("?", "?")) or any(q in tl for q in ("吗", "呢", "怎么", "如何", "能否", "是否可以", "可以吗", "会不会", "why", "how", "什么", "哪些")): - return None - # 已明确给出产出形式(文字类/其他类)则无需确认 - explicit_medium = ( - "文案", "文章", "诗歌", "诗", "小说", "故事", "代码", "程序", "脚本", "总结", "翻译", - "邮件", "报告", "段落", "正文", "文字", "写作", "ppt", "演示", "思维导图", "表格", - "图表", "数据", - ) - if any(m in tl for m in explicit_medium): - return None - # 创作类动词 - creative_verbs = ( - "制作", "创作", "做个", "做一个", "搞一个", "产出", "构思", "设想", - "打造", "塑造", "生成", "帮我做", "帮我生成", "帮我画", "帮我写", "请制作", - "请创作", "请生成", "请画", "请写", "来一个", "来一张", "来一幅", "做一张", - "做张", "写一首", "写一段", "写个", - ) - if not any(v in t for v in creative_verbs): - return None - # 模糊:有创作动词但无明确产出形式 → 提供方向选择 - options = [ - {"key": "text", "label": "文案", "desc": "文章 / 故事 / 诗歌等文字内容", "medium": "文案"}, - {"key": "image", "label": "图片", "desc": "插画 / 海报 / 照片等图像", "medium": "图片"}, - {"key": "video", "label": "视频", "desc": "短片 / 动画等视频", "medium": "视频"}, - {"key": "other", "label": "其他", "desc": "语音 / 代码 / 其他形式", "medium": "其他"}, - ] - return {"options": options} - -# 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("'", "'")) - - -def _outbound_proxy() -> str: - """出网代理地址(兜底)。默认经 172.18.0.1:7890 智能路由代理; - 运行时若未注入代理环境变量则回退到该地址。""" - import os - return os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY") or "http://172.18.0.1:7890" - - -# 香港出网代理(实测对 AGNES 延迟更低、更稳定,2026-07-15 用户实测)。 -# 含 Basic 鉴权,与服务器 /opt/proxy-router/proxy.py 的 UPSTREAM_USER/PASS 一致。 -# 注意:智能路由代理 172.18.0.1:7890 对 AGNES 按 DOMESTIC_OVERRIDE 强制直连(曾返 502), -# 故 AGNES 直接打 HK 代理,绕开智能路由代理。 -_HK_PROXY = "http://qyyfw:XHR521yhf@103.242.14.208:11495" - - -def _proxy_for_url(url: str) -> str | None: - """按目标域名选择是否走代理。 - - 2026-07-15 调整:AGNES(apihub.agnes-ai.com / platform-outputs.agnes-ai.space) - 经用户实测走香港代理 103.242.14.208:11495 延迟更低、更稳定,故改为走 HK 代理 - (带鉴权),不再直连,也不经智能路由代理(172.18.0.1:7890 对 AGNES 返 502)。 - 其余国内/平台域名直连,真正的境外域名走 _outbound_proxy()。 - - 注意:httpx 默认 trust_env=True 会忽略 proxy 参数而改用环境变量 HTTP_PROXY, - 故调用方必须同时传 trust_env=False 才能真正走此处指定的代理。 - """ - import os - from urllib.parse import urlparse - host = (urlparse(url).hostname or "").lower() - if "agnes-ai.com" in host or "agnes-ai.space" in host: - return _HK_PROXY - # 平台自有域名(含自建 Dify/n8n 等)必须直连,不经出网代理。 - if "qihuyun.com" in host: - return None - # 尊重 NO_PROXY:国内域名(硅基流动/DeepSeek/小米MoE 等)直连更稳定, - # 避免经易波动的智能路由代理 172.18.0.1:7890(对国内域名偶发连接重置)。 - _no_proxy = [h.strip().lower() for h in os.environ.get("NO_PROXY", "").split(",") if h.strip()] - for _np in _no_proxy: - if host == _np or host.endswith("." + _np) or _np in host: - return None - return _outbound_proxy() - - -# ============================================================ -# 媒体中转(relay)端点:让微信/iLink 能内联渲染 AGNES 生成的图片/视频 -# 背景:AGNES 媒体 URL 位于境外(agnes-ai.com / agnes-ai.space), -# iLink 服务器(中国大陆)直连常被墙/超时,导致微信里只看到"链接"无法内联播放。 -# 此端点由平台服务器(经香港代理可达 AGNES)代取媒体并以平台域名回传, -# 微信即可内联显示。仅允许白名单域名,杜绝 SSRF。 -# ============================================================ -_MEDIA_RELAY_ALLOWED = ("agnes-ai.com", "agnes-ai.space", "qihuyun.com") - - -@router.get("/media/relay") -async def media_relay(url: str = Query(...)): - """中转 AGNES 媒体(图片/视频)到平台域名,供微信/iLink 内联渲染。 - - 仅允许白名单域名;否则 400。返回的媒体流保持原 content-type。 - """ - import urllib.parse as _up - host = (_up.urlparse(url).hostname or "").lower() - if not any(allowed in host for allowed in _MEDIA_RELAY_ALLOWED): - raise HTTPException(status_code=400, detail="unsupported media host") - try: - async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False) as client: - r = await client.get(url) - if r.status_code != 200: - raise HTTPException(status_code=r.status_code, detail="upstream error") - content_type = r.headers.get("content-type", "application/octet-stream") - data = await r.aread() - return Response(content=data, media_type=content_type) - except HTTPException: - raise - except Exception as e: - raise HTTPException(status_code=502, detail=f"relay failed: {e}") - - -async def _optimize_media_prompt(raw_prompt: str, media_type: str, db=None) -> str: - """使用平台文本模型对用户原始输入做「媒体提示词增强优化」。 - - AGNES 图像/视频模型对 prompt 质量高度敏感: - - 用户在多轮对话中说"帮我生成一张刚才讨论的产品图"包含大量会话噪音 - - AGNES 需要的是**详细的、描述性的、英文优先或中英混合**的视觉提示词 - - 缺乏构图/风格/光影/质量等指导词时,生成结果与预期差异巨大 - - 本函数把用户原文交给文本 LLM 精炼为符合 AGNES 预期的优质 prompt。 - 失败时回退到原始输入(不做额外引导,保证必达)。 - """ - if not raw_prompt or len(raw_prompt.strip()) < 2: - return raw_prompt - # 如果原始输入已经很长(>200 字)且英文为主,可能本身就是一个不错的 prompt,直接返回 - stripped = raw_prompt.strip() - ascii_chars = sum(1 for c in stripped if ord(c) < 128) - if len(stripped) > 200 and ascii_chars / max(len(stripped), 1) > 0.6: - return stripped - # 用文本 LLM 做翻译增强 - try: - from ..capability_registry import resolve_auto_model as _resolve_auto - from ..models import ModelProvider as MP - from ..database import SessionLocal as SL - import httpx as _httpx - - # 独立数据库会话,避免与主流程竞争 - _db = db - _should_close = False - if _db is None: - _db = SL() - _should_close = True - try: - text_model = _resolve_auto("text", _db) - if not text_model: - return stripped - provider = _db.query(MP).filter(MP.id == text_model.provider_id).first() - api_key = text_model.api_key or (provider.api_key if provider else None) - api_url = text_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 not (api_key and api_url): - return stripped - - prompt_zh = ( - "你是一个专业的AI图像/视频提示词工程师。请将以下用户需求提炼成" - f"一段高质量的{'图像' if media_type == 'image' else '视频'}生成" - "提示词。要求:\n" - "1. 用英文书写(AGNES模型对英文响应最佳),核心关键词可保留中文;\n" - "2. 添加构图/风格/光影/色彩/质量描述(如 highly detailed, 8K, " - "professional lighting, masterpiece, cinematic等);\n" - "3. 去掉与视觉无关的对话噪音;\n" - "4. 只输出最终prompt文本,不要任何解释、不要引号包裹、不要\"Prompt:\"前缀。\n\n" - f"用户需求:{stripped}" - ) - messages = [ - {"role": "system", "content": "You are a professional prompt engineer. Output ONLY the final prompt text, no explanations."}, - {"role": "user", "content": prompt_zh}, - ] - # max_tokens 调大到 1024:mimo 等推理模型会把 token 消耗在思考链(reasoning_content) - # 上,512 时 content 常被截断为空;并关闭推理(enable_thinking=False),让模型直接输出提示词。 - # 对不支持该参数的模型至多优化失败 → 下方 except 兜底退化为原始提示词,不阻断生成。 - body = {"model": text_model.name, "messages": messages, "max_tokens": 1024, - "temperature": 0.7, "enable_thinking": False} - async with _httpx.AsyncClient( - timeout=30.0, proxy=_proxy_for_url(api_url), trust_env=False - ) as client: - resp = await client.post( - api_url, - headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, - json=body, - ) - if resp.status_code == 200: - data = resp.json() - opt = data.get("choices", [{}])[0].get("message", {}).get("content", "").strip() - if opt and len(opt) > 5: - return opt - finally: - if _should_close: - _db.close() - except Exception: - pass - return stripped - - -def _friendly_media_err(kind: str, raw: str) -> str: - """把 AGNES 原始错误转译为对用户友好的中文。 - - AGNES 常见错误:429 限频(每分钟1次视频) / Server disconnected(连接重置) / - publish_video_queue_failed + context deadline exceeded(队列超时)。这些都不是 - 平台代码问题,统一给出可操作的提示,不把原始 HTTP/JSON 甩给用户。 - """ - r = (raw or "").lower() - if "429" in raw or "rate limit" in r: - return (f"⚠️ {kind}生成请求过于频繁(AGNES 限制每分钟 1 次)," - f"请稍等 1 分钟后再试。") - if "server disconnected" in r or "连接" in raw or "reset" in r: - return f"⚠️ {kind}生成服务暂时连接失败,请稍后重试。" - if "publish_video" in r or "queue" in r or "deadline" in r or "context" in r: - return f"⚠️ {kind}生成队列繁忙(AGNES 处理超时),请稍后重试。" - return f"⚠️ {kind}生成失败(请稍后重试):{(raw or '')[:120]}" - - -async def _agnes_post_json(api_url: str, headers: dict, payload: dict, timeout) -> tuple: - """直连 AGNES 发 POST,对 5xx / 连接中断做指数退避重试。 - - AGNES 视频/图像队列极不稳定:经常返回 500 publish_video_queue_failed / - context deadline exceeded,或连接被重置(Server disconnected)。这类错误是 AGNES - 侧瞬时抖动,重试通常立刻成功。4xx(参数错误)不重试,直接返回。 - 返回 (resp, err):成功(resp.status_code<500) 时 err=None;耗尽重试时 resp=None。 - """ - last_err = None - for attempt in range(3): - try: - async with httpx.AsyncClient( - timeout=timeout, proxy=_proxy_for_url(api_url), trust_env=False - ) as client: - resp = await client.post(api_url, headers=headers, json=payload) - if resp.status_code < 500: - return resp, None - last_err = f"HTTP {resp.status_code}: {resp.text[:200]}" - except Exception as e: - last_err = str(e)[:200] - if attempt < 2: - await asyncio.sleep(2 * (attempt + 1)) - return None, last_err - - -async def _generate_media(db, ai_model, prompt: str, user_id: str = None) -> str: - """图像/视频生成模型统一入口:按 capabilities 调用 AGNES 等媒体生成 API。 - 返回可直接在 AI 聊天中渲染的富文本(含 /