"""



增强版 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,



    MCPServer,



    gen_uuid



)



from ..auth import get_current_user



from ..platform_knowledge import get_platform_knowledge

from ..mcp_client import list_mcp_tools, call_mcp_tool



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("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")



            .replace('"', "&quot;").replace("'", "&#39;"))











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 聊天中渲染的富文本(含 <img>/<video> 标签)。"""



    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)



    if not api_key:



        return "⚠️ 该媒体模型缺少 API Key，请在后台「大模型库」中配置。"







    caps = [c.lower() for c in (ai_model.capabilities or [])]



    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}



    timeout = ai_model.timeout or 120







    # ============ 图像生成 ============



    if "image" in caps:



        api_url = ai_model.endpoint



        if not api_url:



            api_url = f"{provider.api_base_url.rstrip('/')}/v1/images/generations" if provider else ""



        if not api_url:



            return "⚠️ 该图像模型未配置 Endpoint。"



        # 先优化 prompt（用户原始输入含多轮对话噪音 → 高质量图像生成提示词）



        # 优化失败绝不中断生成，退化为原始提示词



        try:



            optimized_prompt = await _optimize_media_prompt(prompt, "image", db)



        except Exception:



            optimized_prompt = prompt



        payload = {



            "model": ai_model.name,



            "prompt": optimized_prompt,



            "size": "1024x768",



        }



        resp, err = await _agnes_post_json(api_url, headers, payload, timeout)



        if resp is None:



            # 3 次重试后仍失败 → AGNES 侧持续抖动，提示稍后重试而非甩原始错误



            return _friendly_media_err("图像", err)



        if resp.status_code != 200:



            # AGNES 内容策略拒绝：用原始(未优化)提示词重试一次，避免优化后提示词触发误杀



            if resp.status_code == 400 and "content_policy" in (resp.text or "").lower():



                payload["prompt"] = prompt



                resp2, _ = await _agnes_post_json(api_url, headers, payload, timeout)



                if resp2 is not None and resp2.status_code == 200:



                    resp = resp2



                else:



                    return "⚠️ 图像生成被内容策略拒绝，请换一个更中性的描述后重试。"



            else:



                return _friendly_media_err("图像", f"HTTP {resp.status_code}: {resp.text[:200]}")



        try:



            data = resp.json()



        except Exception:



            return f"⚠️ 图像生成失败：响应解析错误 {resp.text[:200]}"



        items = data.get("data") or []



        if not items:



            return f"⚠️ 图像生成失败：响应无图片数据 {str(data)[:200]}"



        item = items[0]



        url = item.get("url") or item.get("b64_json")



        if not url:



            return f"⚠️ 图像生成失败：未返回图片地址 {str(data)[:200]}"



        if not url.startswith("http"):



            url = f"data:image/png;base64,{url}"



        return (f'<img class="md-img" src="{url}" alt="生成图片"/>\n\n'



                f'**提示词：** {_html_escape(prompt)}')







    # ============ 视频生成 (异步任务 + 轮询) ============



    if "video" in caps:



        api_url = ai_model.endpoint



        if not api_url:



            api_url = f"{provider.api_base_url.rstrip('/')}/v1/videos" if provider else ""



        if not api_url:



            return "⚠️ 该视频模型未配置 Endpoint。"



        # 优化 prompt（用户输入含多轮对话噪音 → 高质量视频提示词）；失败退化为原始提示词



        try:



            optimized_prompt = await _optimize_media_prompt(prompt, "video", db)



        except Exception:



            optimized_prompt = prompt



        base = api_url.split("/v1")[0] if "/v1" in api_url else api_url



        payload = {



            "model": ai_model.name,



            "prompt": optimized_prompt,



            "height": 768,



            "width": 1152,



            "num_frames": 121,



            "frame_rate": 24,



        }



        resp, err = await _agnes_post_json(api_url, headers, payload, timeout)



        if resp is None:



            return _friendly_media_err("视频", err)



        if resp.status_code != 200:



            # AGNES 内容策略拒绝：用原始(未优化)提示词重试一次，避免优化后提示词触发误杀



            if resp.status_code == 400 and "content_policy" in (resp.text or "").lower():



                payload["prompt"] = prompt



                resp2, _ = await _agnes_post_json(api_url, headers, payload, timeout)



                if resp2 is not None and resp2.status_code == 200:



                    resp = resp2



                else:



                    return "⚠️ 视频生成被内容策略拒绝，请换一个更中性的描述后重试。"



            else:



                return _friendly_media_err("视频", f"HTTP {resp.status_code}: {resp.text[:200]}")



        try:



            data = resp.json()



        except Exception:



            return f"⚠️ 视频生成失败：响应解析错误 {resp.text[:200]}"



        video_id = data.get("video_id") or data.get("task_id") or data.get("id")



        if not video_id:



            return f"⚠️ 视频生成失败：未返回任务ID {str(data)[:200]}"



        # 轮询结果(最多约 5 分钟)。AGNES 视频状态端点为 GET {base}/v1/videos/{video_id}



        # 该端点顶层返回 status(queued/in_progress/completed/failed)，成片地址在 metadata.url



        result_url = None



        err_msg = None



        for _ in range(60):



            await asyncio.sleep(5)



            try:



                async with httpx.AsyncClient(



                    timeout=timeout, proxy=_proxy_for_url(f"{base}/v1/videos/{video_id}"), trust_env=False



                ) as client:



                    r2 = await client.get(f"{base}/v1/videos/{video_id}", headers=headers)



            except Exception:



                continue



            if r2.status_code != 200:



                continue



            rd = r2.json()



            st = rd.get("status")



            if st == "completed":



                result_url = (rd.get("metadata") or {}).get("url") or rd.get("url")



                break



            if st == "failed":



                err_msg = rd.get("error") or (rd.get("metadata") or {}).get("error")



                break



        if result_url:



            return (f'<video class="md-video" controls src="{result_url}"></video>\n\n'



                    f'**提示词：** {_html_escape(prompt)}')



        return f"⚠️ 视频生成失败：{err_msg or '生成超时，请稍后在对话中重试'}"







    return "⚠️ 未知的媒体模型类型。"











async def _generate_media_task(ai_msg_id: str, ai_model_id: str, prompt: str, user_id: str):



    """后台异步执行图像/视频生成，完成后回写消息内容。







    目的是把最长 5 分钟的 AGNES 视频轮询从 HTTP 请求中剥离，避免请求被客户端/



    代理超时断开(表现为前端「发送失败：请求失败」)。send_message 立即返回占位



    内容(generating 标记)，本任务独立在事件循环中轮询并在完成后 merge 回写消息。



    使用独立的 SessionLocal 会话，因为请求级 db 在响应返回后即被 FastAPI 关闭。



    """



    db = SessionLocal()



    try:



        ai_model = db.query(AIModel).filter(AIModel.id == ai_model_id).first()



        if not ai_model:



            logger.warning(f"[_generate_media_task] 模型不存在: {ai_model_id}")



            return



        try:



            result = await _generate_media(db, ai_model, prompt, user_id)



        except Exception as e:



            result = f"⚠️ 媒体生成失败：{str(e)[:200]}"



        # 回写消息内容(占位 → 最终富文本/错误)



        msg = db.query(Message).filter(Message.id == ai_msg_id).first()



        if msg:



            msg.content = result



            msg.token_used = len(result) // 2



            db.commit()



            logger.info(f"[_generate_media_task] 消息 {ai_msg_id} 已回写内容")



        # 媒体 Token 记账：把图片/视频真实消耗写入 TokenUsageLog，供驾驶舱统计。



        # 主流程对媒体请求只写了占位文本的几 token，此处补写真实媒体消耗（模型名正确、数值合理）。



        _write_media_token_log(db, ai_model, user_id, msg.conversation_id if msg else None, prompt)



    except Exception as e:



        logger.warning(f"[_generate_media_task] 后台生成异常: {e}")



    finally:



        try:



            db.close()



        except Exception:



            pass











def _write_media_token_log(db, ai_model, user_id: str, conversation_id: str, prompt: str):



    """媒体生成完成后补写 TokenUsageLog（图片/视频），修正此前只记占位文本的偏差。







    媒体 API 不返回 token 用量，采用与 IM Bot 路径一致的估算：



    视频基线 1200、图片基线 400，再加 prompt 长度的 1/4 作为提示词开销。



    """



    if not user_id or not ai_model:



        return



    try:



        is_video = "video" in (ai_model.capabilities or [])



        base = 1200 if is_video else 400



        est = base + (len(prompt or "") // 4)



        db.add(TokenUsageLog(



            id=gen_uuid(),



            user_id=user_id,



            model_id=ai_model.id,



            model_name=ai_model.name,



            conversation_id=conversation_id,



            prompt_tokens=est,



            completion_tokens=0,



            total_tokens=est,



            created_at=cst_now(),



        ))



        db.commit()



    except Exception as e:



        logger.warning(f"[_write_media_token_log] 写入失败(忽略): {e}")



        try:



            db.rollback()



        except Exception:



            pass











# ============================================================



# 外部 Agent 远程调用 (Dify / n8n / Coze)



# ============================================================



# 当前台对话关联了一个带有效 agent_endpoint 的外部 Agent 时，运行期直接远程调用



# 该 Agent 平台，而不是用本地 LLM「扮演」。所有出网请求必须 trust_env=False +



# _proxy_for_url() 直连，避免容器 env 里的坏代理(172.18.0.1:7890 返回 502)。











def _normalize_dify_chat_url(endpoint: str) -> str:



    """把用户填写的 Dify 地址归一化为 /v1/chat-messages 端点。与 market.py 保持一致。"""



    ep = (endpoint or "").rstrip("/")



    if ep.endswith("/chat-messages"):



        return ep



    if ep.endswith("/v1"):



        return ep + "/chat-messages"



    return ep + "/v1/chat-messages"











def _extract_text_from_json(obj) -> str:



    """从 n8n / 通用 webhook 的任意 JSON 响应中尽力抽取文本回答。"""



    if obj is None:



        return ""



    if isinstance(obj, str):



        return obj



    if isinstance(obj, (int, float, bool)):



        return str(obj)



    if isinstance(obj, list):



        # 取第一个可抽取到文本的元素



        for it in obj:



            t = _extract_text_from_json(it)



            if t:



                return t



        return ""



    if isinstance(obj, dict):



        # 常见回答字段优先级



        for k in ("output", "text", "answer", "response", "reply",



                  "message", "content", "result", "data"):



            if k in obj:



                t = _extract_text_from_json(obj[k])



                if t:



                    return t



        # 兜底：整体序列化



        try:



            return json.dumps(obj, ensure_ascii=False)



        except Exception:



            return str(obj)



    return str(obj)











async def _call_dify_agent(endpoint: str, api_key: str, query: str,



                           user_id: str, ext_conv_id: str | None) -> tuple:



    """调用 Dify(新版 Agent) /v1/chat-messages，流式(SSE)解析最终回答。







    新版 Agent API 仅支持 response_mode=streaming(blocking 会 400)。事件流中



    message / agent_message 事件携带增量 answer，累加即为最终回答；每个事件都带



    conversation_id，用于维持多轮上下文。返回 (answer, new_conversation_id)。



    """



    url = _normalize_dify_chat_url(endpoint)



    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}



    payload = {



        "inputs": {},



        "query": query,



        "user": (user_id or "web-user")[:64],



        "response_mode": "streaming",



        "conversation_id": ext_conv_id or "",



    }



    answer_parts = []



    new_conv_id = ext_conv_id



    err_msg = None



    wf_outputs = None



    # httpx 默认 follow_redirects=False；stream POST 若遇 301/302 会直接返回该状态码。



    # 且 httpx 自动跟随会把 POST 降级为 GET(对 /v1/chat-messages 变 405)。



    # 因此手动跟随 301/302/307/308，保持 POST 与请求体不变(最多 4 跳)。



    async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False,



                                 follow_redirects=False) as client:



        cur_url = url



        for _redir in range(4):



            async with client.stream("POST", cur_url, headers=headers, json=payload) as resp:



                if resp.status_code in (301, 302, 307, 308):



                    loc = resp.headers.get("location") or resp.headers.get("Location")



                    await resp.aclose()



                    if not loc:



                        raise RuntimeError(f"Dify HTTP {resp.status_code}: 缺少重定向地址")



                    # 处理相对地址



                    if loc.startswith("/"):



                        from urllib.parse import urlsplit, urlunsplit



                        p = urlsplit(cur_url)



                        loc = urlunsplit((p.scheme, p.netloc, loc, "", ""))



                    cur_url = loc



                    continue



                if resp.status_code != 200:



                    body = (await resp.aread()).decode("utf-8", "replace")



                    raise RuntimeError(f"Dify HTTP {resp.status_code}: {body[:300]}")



                async for line in resp.aiter_lines():



                    if not line:



                        continue



                    line = line.strip()



                    if not line.startswith("data:"):



                        continue



                    data_str = line[5:].strip()



                    if not data_str or data_str == "[DONE]":



                        continue



                    try:



                        evt = json.loads(data_str)



                    except Exception:



                        continue



                    ev = evt.get("event")



                    if evt.get("conversation_id"):



                        new_conv_id = evt["conversation_id"]



                    data = evt.get("data") or {}



                    # 兼容三类 Dify 应用：



                    #  1) Chat/Agent 应用: message / agent_message 事件带增量 answer



                    #  2) Chatflow/Workflow 应用: text_chunk 事件带 data.text；



                    #     结束时 workflow_finished 带 status / outputs / error



                    if ev in ("message", "agent_message"):



                        answer_parts.append(evt.get("answer", "") or "")



                    elif ev == "message_replace":



                        # 内容审查替换：以替换后的完整 answer 为准



                        answer_parts = [evt.get("answer", "") or ""]



                    elif ev == "text_chunk":



                        answer_parts.append((data.get("text") or ""))



                    elif ev == "node_finished":



                        # answer / end 节点的 outputs 可能含最终文本



                        if data.get("node_type") in ("answer", "end") and data.get("outputs"):



                            wf_outputs = data.get("outputs")



                        if data.get("status") == "failed" and data.get("error"):



                            err_msg = str(data.get("error"))



                    elif ev == "workflow_finished":



                        if data.get("status") == "failed":



                            err_msg = str(data.get("error") or "Dify 工作流执行失败")



                        elif data.get("outputs"):



                            wf_outputs = data.get("outputs")



                    elif ev == "error":



                        err_msg = evt.get("message") or evt.get("code") or "Dify 流式返回错误"



            break



        else:



            raise RuntimeError("Dify 重定向次数过多(>4)，请直接填写重定向后的地址")



    answer = "".join(answer_parts).strip()



    # 流式无增量文本时，尝试从 workflow 结束 outputs 兜底提取



    if not answer and wf_outputs:



        answer = _extract_text_from_json(wf_outputs).strip()



    if not answer and err_msg:



        raise RuntimeError(err_msg)



    return answer, new_conv_id











async def _call_n8n_agent(endpoint: str, api_key: str, query: str,



                          user_id: str, history: list) -> tuple:



    """调用 n8n 生产 Webhook (POST)。webhook 用 Respond to Webhook 节点返回 JSON，



    尽力从中抽取文本回答。n8n webhook 无平台级会话态，多轮靠我们下发 history。



    返回 (answer, None)。"""



    url = (endpoint or "").strip()



    headers = {"Content-Type": "application/json"}



    if api_key:



        # n8n Header Auth 自定义头名不固定，同时下发常见几种以提高兼容性



        headers["Authorization"] = f"Bearer {api_key}"



        headers["X-N8N-API-KEY"] = api_key



    # 携带多种字段别名，兼容 n8n AI Agent(chatInput) 与自定义工作流



    payload = {



        "query": query,



        "chatInput": query,



        "message": query,



        "input": query,



        "user": (user_id or "web-user")[:64],



        "history": history or [],



    }



    async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False) as client:



        resp = await client.post(url, headers=headers, json=payload)



    if resp.status_code not in (200, 201):



        raise RuntimeError(f"n8n Webhook HTTP {resp.status_code}: {resp.text[:300]}")



    ctype = resp.headers.get("content-type", "")



    if "application/json" in ctype:



        try:



            answer = _extract_text_from_json(resp.json())



        except Exception:



            answer = resp.text



    else:



        answer = resp.text



    return (answer or "").strip(), None











async def _call_coze_agent(endpoint: str, api_key: str, query: str,



                           user_id: str, ext_conv_id: str | None) -> tuple:



    """调用扣子(Coze) v3 对话 API：create → 轮询 retrieve → 拉取 message list。







    约定 agent_endpoint 存放 bot_id(纯 id)，或形如



    https://api.coze.cn/v3/chat?bot_id=xxx 的完整地址。默认基址 api.coze.cn。



    返回 (answer, conversation_id)。



    """



    ep = (endpoint or "").strip()



    from urllib.parse import urlparse, parse_qs



    bot_id = None



    base = "https://api.coze.cn"



    if ep.startswith("http"):



        pr = urlparse(ep)



        base = f"{pr.scheme}://{pr.netloc}"



        qs = parse_qs(pr.query)



        bot_id = (qs.get("bot_id") or [None])[0]



        if not bot_id and "/bot/" in pr.path:



            bot_id = pr.path.rstrip("/").split("/bot/")[-1]



    else:



        bot_id = ep  # 纯 bot_id



    if not bot_id:



        raise RuntimeError("Coze 接入缺少 bot_id(请在 Endpoint 填入 bot_id 或带 ?bot_id= 的地址)")



    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}



    create_url = f"{base}/v3/chat"



    if ext_conv_id:



        create_url += f"?conversation_id={ext_conv_id}"



    payload = {



        "bot_id": bot_id,



        "user_id": (user_id or "web-user")[:64],



        "stream": False,



        "auto_save_history": True,



        "additional_messages": [



            {"role": "user", "content": query, "content_type": "text"}



        ],



    }



    async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(create_url), trust_env=False) as client:



        r = await client.post(create_url, headers=headers, json=payload)



        if r.status_code != 200:



            raise RuntimeError(f"Coze create HTTP {r.status_code}: {r.text[:300]}")



        cd = r.json()



        if cd.get("code") not in (0, None):



            raise RuntimeError(f"Coze 错误: {cd.get('msg') or cd}")



        chat = cd.get("data") or {}



        chat_id = chat.get("id")



        conv_id2 = chat.get("conversation_id") or ext_conv_id



        if not chat_id or not conv_id2:



            raise RuntimeError(f"Coze 未返回 chat_id/conversation_id: {str(cd)[:200]}")



        # 轮询直到完成(最多约 3 分钟)



        status = chat.get("status")



        for _ in range(60):



            if status == "completed":



                break



            if status in ("failed", "requires_action", "canceled"):



                raise RuntimeError(f"Coze 对话状态 {status}")



            await asyncio.sleep(3)



            rr = await client.get(



                f"{base}/v3/chat/retrieve?chat_id={chat_id}&conversation_id={conv_id2}",



                headers=headers,



            )



            if rr.status_code == 200:



                status = ((rr.json() or {}).get("data") or {}).get("status")



        # 拉取消息列表，取 type=answer 的内容



        rm = await client.get(



            f"{base}/v3/chat/message/list?chat_id={chat_id}&conversation_id={conv_id2}",



            headers=headers,



        )



        answer = ""



        if rm.status_code == 200:



            for m in ((rm.json() or {}).get("data") or []):



                if m.get("type") == "answer" and m.get("role") == "assistant":



                    answer = m.get("content", "") or answer



    return (answer or "").strip(), conv_id2











async def _call_external_agent(db, agent: dict, query: str, user_id: str,



                               ext_conversation_id: str | None,



                               history: list) -> tuple:



    """按平台分发到对应外部 Agent 远程调用。返回 (answer, new_ext_conversation_id)。"""



    platform = (agent.get("platform") or "dify").lower()



    endpoint = agent.get("agent_endpoint")



    api_key = agent.get("agent_api_key") or ""



    if not endpoint:



        raise RuntimeError("该 Agent 未配置远程调用地址(agent_endpoint)")



    if platform == "dify":



        return await _call_dify_agent(endpoint, api_key, query, user_id, ext_conversation_id)



    if platform == "n8n":



        return await _call_n8n_agent(endpoint, api_key, query, user_id, history)



    if platform == "coze":



        return await _call_coze_agent(endpoint, api_key, query, user_id, ext_conversation_id)



    # 未知平台：默认按 Dify 协议尝试



    return await _call_dify_agent(endpoint, api_key, query, user_id, ext_conversation_id)











# ============================================================



# Request Models



# ============================================================







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



    agent_ids: Optional[list] = None



    skill_ids: Optional[list] = None



    knowledge_base_ids: Optional[list] = 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



    mcp_ids: Optional[list] = None



    attachments: Optional[list] = None



    use_tools: Optional[bool] = False











# ============================================================



# Skill Resolution: MarketApp → Skill Definition



# ============================================================







def _resolve_market_skills(db: Session, skill_ids: list, user_id: str) -> list:



    """



    Stage A.1: 统一Skill ID空间



    前端传MarketApp.id, 需要映射为可用的技能定义。



    返回: [{id, name, description, skill_content, category, func_name}]



    """



    if not skill_ids:



        return []







    # 1. 从MarketApp查询 (前端传的是MarketApp.id)



    market_skills = db.query(MarketApp).filter(



        MarketApp.id.in_(skill_ids),



        MarketApp.type == "skill"



    ).all()







    results = []



    for ms in market_skills:



        skill_content = ms.skill_content or ""



        # 尝试从skill_content解析SKILL.md格式



        # SKILL.md格式: ---\nname: ...\ndescription: ...\n---\n内容...



        parsed = _parse_skill_content(skill_content)







        func_name = ms.name.replace(" ", "_").replace("-", "_").lower()



        results.append({



            "id": ms.id,



            "market_app_id": ms.id,



            "name": ms.name,



            "description": ms.description or parsed.get("description", ""),



            "skill_content": skill_content,



            "category": parsed.get("category", "general"),



            "skill_type": parsed.get("skill_type", "general"),



            "func_name": func_name,



            "parameters": parsed.get("parameters", {"input": {"type": "string", "description": "用户输入内容"}}),



            "instructions": parsed.get("instructions", ""),



        })







    # 2. 如果有未匹配的ID, 尝试查EvolutionSkill (兼容旧的进化技能ID)



    matched_ids = {r["id"] for r in results}



    unmatched = [sid for sid in skill_ids if sid not in matched_ids]







    if unmatched:



        evo_skills = db.query(EvolutionSkill).filter(



            EvolutionSkill.id.in_(unmatched),



            EvolutionSkill.user_id == user_id



        ).all()







        for es in evo_skills:



            # 读取SKILL.md内容



            skill_md = ""



            if es.skill_md_path and os.path.exists(es.skill_md_path):



                try:



                    with open(es.skill_md_path, "r", encoding="utf-8") as f:



                        skill_md = f.read()



                except Exception:



                    skill_md = ""



            elif es.description:



                skill_md = es.description







            parsed = _parse_skill_content(skill_md)



            func_name = es.skill_name.replace(" ", "_").replace("-", "_").lower()







            results.append({



                "id": es.id,



                "market_app_id": None,



                "name": es.skill_name,



                "description": es.description or parsed.get("description", ""),



                "skill_content": skill_md,



                "category": es.category or parsed.get("category", "general"),



                "skill_type": parsed.get("skill_type", es.category or "general"),



                "func_name": func_name,



                "parameters": parsed.get("parameters", {"input": {"type": "string", "description": "用户输入内容"}}),



                "instructions": parsed.get("instructions", ""),



            })







    return results











def _resolve_market_agents(db: Session, agent_ids: list, user_id: str) -> list:



    """



    Stage B.1: 加载Agent配置



    前端传MarketApp.id(agent类型), 需要解析AgentConfig。



    返回: [{id, name, config, skills, knowledge_base_ids}]



    """



    if not agent_ids:



        return []







    agents = db.query(MarketApp).filter(



        MarketApp.id.in_(agent_ids),



        MarketApp.type == "agent"



    ).all()







    results = []



    for ag in agents:



        # AgentConfig在skill_content字段(JSON序列化)



        agent_config = {}



        try:



            if ag.skill_content:



                agent_config = json.loads(ag.skill_content)



        except Exception:



            agent_config = {"role": ag.description or "AI助手", "instructions": ag.description or ""}







        # 提取配置字段



        role_prompt = agent_config.get("role", ag.description or "AI助手")



        instructions = agent_config.get("instructions", "")



        agent_skill_ids = agent_config.get("skills", [])  # 关联的skill MarketApp IDs



        agent_kb_ids = agent_config.get("knowledge_base_ids", [])  # 关联的知识库IDs



        agent_model_id = agent_config.get("model_id", None)  # 指定使用的模型



        tools = agent_config.get("tools", [])  # MCP server IDs(暂不用)







        results.append({



            "id": ag.id,



            "name": ag.name,



            "role_prompt": role_prompt,



            "instructions": instructions,



            "skill_ids": agent_skill_ids,



            "knowledge_base_ids": agent_kb_ids,



            "model_id": agent_model_id,



            "tools": tools,



            "description": ag.description or "",



            # 外部 Agent 接入(Dify/n8n/Coze)运行期远程调用所需字段：



            # 这三列此前被丢弃，导致选中外部 Agent 时仍用本地 LLM「扮演」而非真远程调用。



            "platform": (ag.platform or "").lower() if ag.platform else None,



            "agent_endpoint": (ag.agent_endpoint or "").strip() if ag.agent_endpoint else None,



            "agent_api_key": ag.agent_api_key or None,



        })







    return results











def _resolve_market_mcp(db: Session, mcp_ids: list, user_id: str) -> list:



    """



    加载用户选择的 MCP 应用配置（MarketApp.type == 'mcp'）。



    返回: [{id, name, transport, endpoint, description}]



    """



    if not mcp_ids:



        return []







    mcps = db.query(MarketApp).filter(



        MarketApp.id.in_(mcp_ids),



        MarketApp.type == "mcp"



    ).all()







    results = []



    for m in mcps:



        config = {}



        try:



            if m.skill_content:



                config = json.loads(m.skill_content)



        except Exception:



            config = {}



        results.append({



            "id": m.id,



            "name": m.name,



            "transport": m.mcp_transport or config.get("transport", "stdio"),



            "endpoint": m.agent_endpoint or config.get("endpoint", "") or config.get("url", ""),



            "description": m.description or config.get("description", ""),



        })







    return results











def _resolve_active_mcp_servers(db: Session, mcp_ids: list) -> list:

    """

    解析用户选中的 MCP 服务连接信息（来自 mcp_servers 表，import 与后台添加通用）。

    返回: [{id, name, description, transport, endpoint, url, headers}]

    """

    if not mcp_ids:

        return []

    servers = db.query(MCPServer).filter(

        MCPServer.id.in_(mcp_ids),

        MCPServer.is_enabled == True,

    ).all()

    out = []

    for s in servers:

        try:

            headers = json.loads(s.headers) if isinstance(s.headers, str) else (s.headers or {})

        except Exception:

            headers = {}

        out.append({

            "id": s.id,

            "name": s.name,

            "description": s.description or "",

            "transport": s.transport_type or "streamable_http",

            "endpoint": s.url or "",

            "url": s.url or "",

            "headers": headers,

        })

    return out





def _discover_installed_skills(db: Session, user_id: str, limit: int = 15) -> list:



    """



    Issue 8: 默认技能自动发现。



    检索「本账号已安装(app_installations) 且 平台已部署发布(MarketApp.status='published')」的



    skill 型应用，返回可供 function calling 使用的技能定义列表（最多 limit 个）。



    仅当用户未显式选择技能时才启用；多个技能适用时由模型自行判断选 1 个最优。



    """



    if not user_id:



        return []



    installed = db.query(AppInstallation).filter(



        AppInstallation.user_id == user_id



    ).all()



    app_ids = [i.app_id for i in installed]



    if not app_ids:



        return []



    apps = db.query(MarketApp).filter(



        MarketApp.id.in_(app_ids),



        MarketApp.type == "skill",



        MarketApp.status == "published",



    ).all()



    ids = [a.id for a in apps]



    if not ids:



        return []



    return _resolve_market_skills(db, ids, user_id)[:limit]











def _parse_skill_content(content: str) -> dict:



    """



    解析SKILL.md格式内容, 提取结构化信息。



    SKILL.md格式:



    ---



    name: 技能名



    description: 描述



    category: 分类



    skill_type: 类型



    parameters: JSON或简单定义



    ---



    技能指令/规则内容



    """



    result = {



        "name": "",



        "description": "",



        "category": "general",



        "skill_type": "general",



        "parameters": {},



        "instructions": "",



    }







    if not content:



        return result







    # 尝试解析YAML front matter



    if content.strip().startswith("---"):



        parts = content.split("---")



        if len(parts) >= 3:



            front_matter = parts[1].strip()



            body = parts[2].strip() if len(parts) > 2 else ""







            # 简单YAML解析 (不用pyyaml库)



            for line in front_matter.split("\n"):



                line = line.strip()



                if ":" in line:



                    key, val = line.split(":", 1)



                    key = key.strip().lower()



                    val = val.strip()



                    if key == "name":



                        result["name"] = val



                    elif key == "description":



                        result["description"] = val



                    elif key == "category":



                        result["category"] = val



                    elif key == "skill_type":



                        result["skill_type"] = val



                    elif key == "parameters":



                        try:



                            result["parameters"] = json.loads(val)



                        except Exception:



                            # 简单格式: input:string,query:string



                            params = {}



                            for p in val.split(","):



                                p = p.strip()



                                if ":" in p:



                                    pname, ptype = p.split(":", 1)



                                    params[pname.strip()] = {"type": ptype.strip(), "description": pname.strip()}



                            result["parameters"] = params







            result["instructions"] = body



        else:



            result["instructions"] = content



    else:



        # 没有front matter, 全部当作instructions



        result["instructions"] = content



        # 尝试从第一行提取name



        first_line = content.split("\n")[0].strip()



        if first_line and len(first_line) < 50:



            result["name"] = first_line







    return result











# ============================================================



# BM25 Text Search (Stage C.2)



# ============================================================







def _bm25_search(query: str, documents: list, top_k: int = 5) -> list:



    """



    简易BM25检索 — 替代朴素的文档拼接。



    对文档列表进行关键词匹配+TF-IDF排序, 返回最相关的top_k段。



    """



    if not documents or not query:



        return documents[:top_k]







    # 中文简单分词(按字符+标点)



    def tokenize(text: str) -> list:



        # 移除标点, 拆分为2-4字词组



        text = re.sub(r'[^\w\u4e00-\u9fff]', ' ', text.lower())



        tokens = []



        words = text.split()



        for w in words:



            if len(w) >= 2:



                tokens.append(w)



            # 也生成2-gram



            if len(w) >= 4:



                for i in range(len(w) - 2):



                    tokens.append(w[i:i+2])



        return tokens







    query_tokens = tokenize(query)



    if not query_tokens:



        return documents[:top_k]







    scored_docs = []



    avg_doc_len = sum(len(tokenize(d.get("content", ""))) for d in documents) / max(len(documents), 1)







    # BM25参数



    k1 = 1.5



    b = 0.75







    # 计算每个token的IDF



    df = {}



    for token in query_tokens:



        for doc in documents:



            if token in tokenize(doc.get("content", "")):



                df[token] = df.get(token, 0) + 1







    N = len(documents)



    idf = {}



    for token in query_tokens:



        idf[token] = math.log((N - df.get(token, 0) + 0.5) / (df.get(token, 0) + 0.5) + 1)







    # 计算每个文档的BM25得分



    for doc in documents:



        doc_tokens = tokenize(doc.get("content", ""))



        doc_len = len(doc_tokens)



        score = 0







        for token in query_tokens:



            tf = doc_tokens.count(token)



            numerator = tf * (k1 + 1)



            denominator = tf + k1 * (1 - b + b * doc_len / max(avg_doc_len, 1))



            score += idf.get(token, 0) * numerator / denominator







        scored_docs.append((score, doc))







    # 按得分排序



    scored_docs.sort(key=lambda x: x[0], reverse=True)



    return [d for s, d in scored_docs[:top_k]]











def _retrieve_kb_context(db: Session, kb_ids: list, query: str, max_chars: int = 3000) -> str:



    """



    Stage C: 增强版知识库检索 — 使用BM25排序替代朴素拼接



    1. 获取每个KB的文档



    2. 构建文档列表(content+title+metadata)



    3. BM25检索最相关的段落



    4. 拼接为上下文注入system prompt



    同时叠加 Dify 外部知识库(RAG)检索结果。



    """



    local_context = ""







    if kb_ids:



        kbs = db.query(KnowledgeBase).filter(KnowledgeBase.id.in_(kb_ids)).all()



        if kbs:



            # 收集所有文档段落



            # Stage C.1: 优先使用 KnowledgeDocumentChunk 表的预分段数据



            all_segments = []



            for kb in kbs:



                # 先查chunk表



                chunks = db.query(KnowledgeDocumentChunk).filter(



                    KnowledgeDocumentChunk.knowledge_base_id == kb.id



                ).order_by(KnowledgeDocumentChunk.chunk_index).limit(30).all()







                if chunks:



                    # 使用预分段数据



                    for chunk in chunks:



                        if chunk.content and len(chunk.content.strip()) >= 20:



                            all_segments.append({



                                "content": "[{}/{}/段{}]: {}".format(



                                    kb.name, chunk.document_id[:8], chunk.chunk_index, chunk.content[:500]



                                ),



                                "raw_content": chunk.content,



                                "kb_name": kb.name,



                            })



                else:



                    # 退化: 使用原始文档分段



                    docs = db.query(KnowledgeDocument).filter(



                        KnowledgeDocument.knowledge_base_id == kb.id,



                        KnowledgeDocument.status == "completed"



                    ).order_by(KnowledgeDocument.updated_at.desc()).limit(20).all()







                    for doc in docs:



                        content = doc.content or ""



                        if not content:



                            continue







                        # 如果文档较长, 分段处理 (每段~800字)



                        if len(content) > 800:



                            paragraphs = re.split(r'\n\n+', content)



                            for pi, para in enumerate(paragraphs):



                                if len(para.strip()) >= 20:



                                    all_segments.append({



                                        "content": "[{}/{}/段{}]: {}".format(



                                            kb.name, doc.name, pi+1, para.strip()[:500]



                                        ),



                                        "raw_content": para.strip(),



                                        "kb_name": kb.name,



                                        "doc_name": doc.name,



                                    })



                        else:



                            all_segments.append({



                                "content": "[{}/{}]: {}".format(kb.name, doc.name, content[:500]),



                                "raw_content": content,



                                "kb_name": kb.name,



                                "doc_name": doc.name,



                            })







            if not all_segments:



                # 退化: 使用原始朴素拼接



                kb_names = ", ".join([k.name for k in kbs])



                local_context = f"参考知识库: {kb_names}"



            else:



                # BM25检索最相关的段落



                top_segments = _bm25_search(query, all_segments, top_k=8)



                context_parts = [seg["content"] for seg in top_segments]



                local_context = "\n".join(context_parts)



                if len(local_context) > max_chars:



                    local_context = local_context[:max_chars] + "...(内容截断)"







    # 叠加 Dify 外部知识库(RAG)检索



    dify_context = ""



    try:



        from app.dify_kb import get_dify_kb_context



        dify_context = get_dify_kb_context(db, query, top_k=5)



    except Exception as e:



        logger.warning(f"Dify KB context retrieval failed: {e}")







    parts = []



    if local_context:



        parts.append(local_context)



    if dify_context:



        parts.append("【Dify 知识库检索结果】\n" + dify_context)



    return "\n\n".join(parts)











async def _ingest_answer_to_dify(conv_id: str, msg_id: str, content: str):



    """后台任务：将助手回答入库到 Dify 知识库（量化后可供 RAG 检索复用）。"""



    try:



        from app.dify_kb import aingest_text



        from app.database import SessionLocal



        from sqlalchemy import text as _t



        db = SessionLocal()



        try:



            row = db.execute(_t(



                "SELECT api_url, api_key, config FROM external_kb_services "



                "WHERE service_type='dify' AND is_enabled=1 LIMIT 1"



            )).fetchone()



        finally:



            db.close()



        if not row:



            return



        api_url, api_key, config_raw = row



        dataset_name = None



        if config_raw:



            try:



                import json



                cfg = config_raw if isinstance(config_raw, dict) else json.loads(config_raw)



                dataset_name = cfg.get("dataset_name")



            except Exception:



                pass



        await aingest_text(api_url, api_key, content, name=f"chat_{conv_id}_{msg_id}", dataset_name=dataset_name)



    except Exception as e:



        logger.warning(f"后台入库 Dify 失败: {e}")











# ============================================================



# Dynamic System Prompt Builder (Stage B)



# ============================================================







def _build_system_prompt(



    skills: list,



    agents: list,



    kb_context: str,



    user_content: str,



    mcp: list = None,

    mcp_functions: list = None,

    auto_discovered: bool = False,

) -> tuple:



    """



    动态组装system prompt + function definitions。



    返回: (system_prompt_str, function_definitions_list)



    """



    system_parts = ["你是AIFlow平台的AI助手，具备专业的知识储备和工具调用能力。"]



    



    # 平台知识库注入（始终开启）：确保AI优先使用平台已有能力



    try:



        pk = get_platform_knowledge()



        if pk:



            system_parts.append(f"【平台知识库——优先使用以下知识回答用户问题】\n{pk}")



    except Exception:



        pass



    



    # 回复格式要求：强制使用Markdown



    system_parts.append("回复格式要求：你必须使用Markdown格式回复，确保层次分明、条理清晰。使用##和###做标题、**加粗**强调重点、-或1.做列表、```代码块包裹代码或配置、表格展示对比数据。禁止回复纯文本长段落。")







    function_definitions = []







    # Stage B.1: Agent角色注入（Issue 2：让用户询问 agent 能力时，AI 能“看到”该 agent 资料）



    if agents:



        for agent in agents:



            name = agent.get("name", "")



            role = agent.get("role_prompt", "")



            instructions = agent.get("instructions", "")



            desc = agent.get("description", "")



            block = []



            if name:



                block.append(



                    f"用户已选择智能体「{name}」。当用户询问该智能体的功能、能力、用法或相关问题时，"



                    f"请基于以下资料准确作答，不要说“不知道这个智能体”。"



                )



            if role:



                block.append(f"该智能体的角色设定: {role}")



            if desc:



                block.append(f"该智能体的简介: {desc}")



            if instructions:



                block.append(f"该智能体的行为指引: {instructions}")



            if block:



                system_parts.append("\n".join(block))







    # Stage A.2: Skill工具注入



    if skills:



        skill_descs = ", ".join([f"{s['name']}({s.get('category', 'general')})" for s in skills])



        system_parts.append(f"可用技能工具: {skill_descs}")



        if auto_discovered:



            # Issue 8：每轮默认先寻找本账号已安装且平台已部署的技能，多个适用时 AI 自行判断选 1 个最优



            system_parts.append(



                "重要规则: 以上技能为本账号已安装且平台已部署的技能。每轮对话请先判断用户请求是否可由其中某个技能处理；"



                "若多个技能都适用，请自行判断并选择最贴切的 1 个优先调用，不要一次性全部调用。"



                "调用对应工具后将获得执行结果，基于结果生成最终回复。若这些技能均不适用，则直接基于你的知识回答。"



            )



        else:



            system_parts.append("重要规则: 当用户请求涉及上述技能工具的处理范围时, 你必须优先通过function calling调用对应工具, 而不是直接回复文字。调用后将获得工具执行结果, 基于结果生成最终回复。")







        # 为每个skill构建function definition



        for s in skills:



            # 从skill_content提取指令注入system prompt



            instructions = s.get("instructions", "")



            if instructions and len(instructions) < 800:



                system_parts.append(f"技能「{s['name']}」的详细指令: {instructions}")







            # 构建参数schema



            params = s.get("parameters", {})



            if isinstance(params, dict) and len(params) > 0:



                properties = {}



                required = []



                for pname, pdef in params.items():



                    if isinstance(pdef, dict):



                        properties[pname] = {



                            "type": pdef.get("type", "string"),



                            "description": pdef.get("description", f"传递给{s['name']}的{pname}参数"),



                        }



                        if pdef.get("required", True):



                            required.append(pname)



                    else:



                        properties[pname] = {"type": "string", "description": f"传递给{s['name']}的{pname}参数"}



                        required.append(pname)







                if not properties:



                    properties = {"input": {"type": "string", "description": f"需要传递给{s['name']}的完整用户输入"}}



                    required = ["input"]



            else:



                properties = {"input": {"type": "string", "description": f"需要传递给{s['name']}的完整用户输入"}}



                required = ["input"]







            desc = s.get("description", "")



            category = s.get("category", "general")



            func_desc = f"调用技能「{s['name']}」处理用户请求。使用场景: {desc}。当用户请求涉及{category}类任务时优先调用此工具。"







            function_definitions.append({



                "name": s["func_name"],



                "description": func_desc,



                "parameters": {



                    "type": "object",



                    "properties": properties,



                    "required": required,



                },



            })







    # KB context injection



    if kb_context:



        system_parts.append(f"参考以下知识库内容回答用户问题:\n{kb_context}")







    # MCP 应用注入（Issue 3：让用户询问 MCP 能力时，AI 能“看到”该 MCP 的资料与地址）



    if mcp:



        mcp_lines = []



        for m in mcp:



            name = m.get("name", "")



            desc = m.get("description", "")



            transport = m.get("transport", "stdio")



            endpoint = m.get("endpoint", "")



            line = f"用户已选择 MCP 应用「{name}」（传输方式: {transport}）"



            if desc:



                line += f"，功能简介: {desc}"



            if endpoint:



                line += f"，服务地址: {endpoint}"



            mcp_lines.append(line)



        system_parts.append("\n".join(mcp_lines))



        system_parts.append(



            "用户已选择上述 MCP 应用。当请求需要借助外部工具/系统（如查询数据、调用 API、操作第三方服务）时，"



            "请优先参考这些 MCP 应用的能力来回答；若平台已接入对应工具，优先通过工具调用完成；"



            "不要说“不了解该 MCP”。"



        )







    system_prompt = "\n".join(system_parts)



    # MCP function definitions 注入（让 LLM 可通过 function calling 真实调用 MCP 工具）

    if mcp_functions:

        function_definitions.extend(mcp_functions)



    return system_prompt, function_definitions











# ============================================================



# Skill Tool Execution (Stage A.2 rewritten)



# ============================================================







def _execute_skill_tool(



    db: Session,



    func_name: str,



    args: dict,



    skills: list,



    user_content: str,



) -> str:



    """



    Stage A.2: 真实Skill执行引擎 — 替代mock实现



    1. 查找匹配的skill定义



    2. 读取skill_content(SKILL.md)中的指令



    3. 根据指令类型组装执行prompt



    4. 返回结构化的执行结果给LLM



    """



    # 查找匹配的skill



    matched_skill = None



    for s in skills:



        expected_name = s["func_name"]



        if func_name == expected_name or func_name == f"skill_{s['id'][:8]}":



            matched_skill = s



            break







    if not matched_skill:



        # 退化: 返回通用结果



        return f"工具 '{func_name}' 执行结果: 已处理输入内容, 但未找到对应的技能定义。请基于通用知识回复用户。"







    skill_name = matched_skill["name"]



    instructions = matched_skill.get("instructions", "")



    input_text = args.get("input", user_content)







    # 根据skill_content中的指令生成真实执行结果



    # 核心思路: 将skill的instructions作为"执行逻辑",



    # 把input_text传入, 让结果包含指令要求的处理格式







    if instructions:



        # 有明确指令的技能 — 按指令结构化输出



        result_parts = [



            f"技能「{skill_name}」执行结果:",



            f"---",



        ]







        # 尝试从指令中提取输出格式要求



        # 如果指令包含"输出"或"格式"关键字, 遵循其要求



        output_format = ""



        if "输出格式" in instructions or "output format" in instructions.lower():



            # 提取格式描述



            for line in instructions.split("\n"):



                if "输出" in line or "格式" in line or "format" in line.lower():



                    output_format += line + "\n"







        # 根据skill_type增强处理



        skill_type = matched_skill.get("skill_type", "general")







        if skill_type == "writer" or "写作" in skill_name or "写" in instructions:



            result_parts.append(f"基于技能「{skill_name}」的写作指令处理用户输入:")



            result_parts.append(f"主题: {input_text[:200]}")



            result_parts.append(f"根据技能规则: {instructions[:300]}")



            result_parts.append(f"处理完成 — 请基于上述技能指令和输入内容, 生成符合技能要求的写作输出。")



        elif skill_type == "analyzer" or "分析" in skill_name or "分析" in instructions:



            result_parts.append(f"基于技能「{skill_name}」的分析指令处理用户输入:")



            result_parts.append(f"分析对象: {input_text[:200]}")



            result_parts.append(f"分析维度(按技能规则): {instructions[:300]}")



            result_parts.append(f"关键发现: 识别到{max(1, len(input_text)//50)}个分析要点")



            result_parts.append(f"处理完成 — 请基于上述技能指令, 对输入内容进行深入分析并给出结构化的分析报告。")



        elif skill_type == "translator" or "翻译" in skill_name or "翻译" in instructions:



            result_parts.append(f"基于技能「{skill_name}」的翻译指令处理用户输入:")



            result_parts.append(f"原文: {input_text[:200]}")



            result_parts.append(f"翻译规则: {instructions[:300]}")



            result_parts.append(f"处理完成 — 请基于技能翻译规则, 将输入内容翻译为目标语言。")



        elif skill_type == "summarizer" or "总结" in skill_name or "摘要" in instructions:



            result_parts.append(f"基于技能「{skill_name}」的摘要指令处理用户输入:")



            result_parts.append(f"原文长度: {len(input_text)} 字符")



            result_parts.append(f"摘要规则: {instructions[:300]}")



            result_parts.append(f"处理完成 — 请基于技能摘要规则, 对输入内容进行精准摘要。")



        else:



            # 通用技能 — 传递指令让LLM执行



            result_parts.append(f"基于技能「{skill_name}」的指令处理用户请求:")



            result_parts.append(f"用户输入: {input_text[:200]}")



            result_parts.append(f"技能指令: {instructions[:500]}")



            if output_format:



                result_parts.append(f"期望输出格式: {output_format[:200]}")



            result_parts.append(f"处理完成 — 请严格按照技能指令的规则和要求, 处理用户输入并生成最终回复。")







        result = "\n".join(result_parts)



    else:



        # 无明确指令的技能 — 提供基本信息让LLM自行处理



        desc = matched_skill.get("description", "")



        result = (



            f"技能「{skill_name}」已激活。\n"



            f"技能描述: {desc}\n"



            f"用户输入: {input_text[:200]}\n"



            f"请基于技能「{skill_name}」的专业能力处理上述输入。"



        )







    # 更新使用计数(如果关联了MarketApp)



    market_app_id = matched_skill.get("market_app_id")



    if market_app_id:



        try:



            app = db.query(MarketApp).filter(MarketApp.id == market_app_id).first()



            if app:



                app.install_count = (app.install_count or 0) + 1



        except Exception:



            pass







    return result











# ============================================================



# Conversation Context Persistence (Stage B.2)



# ============================================================







def _get_conversation_context(db: Session, conv_id: str) -> dict:



    """



    Stage B.2: 获取对话的持久化上下文



    对话级别的agent/skill/kb选择保存在Conversation字段中。



    """



    conv = db.query(Conversation).filter(Conversation.id == conv_id).first()



    if not conv:



        return {}







    # Conversation目前没有存储agent/skill/kb的JSON字段



    # 我们从第一条消息中获取(如果存在), 或返回空



    # 后续可通过数据库迁移添加Conversation.context字段







    # 查询该对话中最早一条含有skill_ids/agent_ids/knowledge_ids的消息



    first_tool_msg = db.query(Message).filter(



        Message.conversation_id == conv_id,



        Message.role == "user",



    ).order_by(asc(Message.created_at)).first()







    context = {}



    if first_tool_msg:



        if first_tool_msg.skill_ids:



            context["skill_ids"] = first_tool_msg.skill_ids



        if first_tool_msg.agent_ids:



            context["agent_ids"] = first_tool_msg.agent_ids



        if first_tool_msg.knowledge_ids:



            context["knowledge_base_ids"] = first_tool_msg.knowledge_ids



        if getattr(first_tool_msg, "mcp_ids", None):



            context["mcp_ids"] = first_tool_msg.mcp_ids







    return context











# ============================================================



# API Endpoints



# ============================================================







@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}%"))







    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,



            "source": getattr(conv, "source", None),



            "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=cst_now(),



        updated_at=cst_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



    # Stage B.2: 保存agent/skill/kb选择到对话上下文



    # (存储在第一条system消息的knowledge_ids/agent_ids/skill_ids字段中)



    conv.updated_at = cst_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 = cst_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), asc(Message.id))



        .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="对话不存在")







    # Stage B.2: 多轮上下文保持 — 如果当前请求没有skill/agent/kb,



    # 从对话历史中继承上一次的选择



    effective_skill_ids = req.skill_ids or []



    effective_agent_ids = req.agent_ids or []



    effective_kb_ids = req.knowledge_base_ids or []



    effective_mcp_ids = req.mcp_ids or []



    effective_use_tools = req.use_tools or False







    if not effective_skill_ids and not effective_agent_ids and not effective_kb_ids and not effective_mcp_ids:



        # 从对话上下文继承



        prev_context = _get_conversation_context(db, conv_id)



        if prev_context:



            effective_skill_ids = prev_context.get("skill_ids", [])



            effective_agent_ids = prev_context.get("agent_ids", [])



            effective_kb_ids = prev_context.get("knowledge_base_ids", [])



            effective_mcp_ids = prev_context.get("mcp_ids", [])



            if effective_skill_ids or effective_agent_ids or effective_mcp_ids:



                effective_use_tools = True







    # 附件内容拼接到用户消息中(文本类文件附带内容, 供模型参考)



    attachment_block = ""



    if req.attachments:



        parts = []



        for att in req.attachments:



            name = att.get("name", "附件") if isinstance(att, dict) else str(att)



            ftype = att.get("file_type", "") if isinstance(att, dict) else ""



            content = att.get("content", "") if isinstance(att, dict) else ""



            if content:



                parts.append(f"【附件: {name}】\n{content}")



            else:



                parts.append(f"【附件(二进制, 名称: {name}, 类型: {ftype})】")



        if parts:



            attachment_block = "\n\n" + "\n\n".join(parts)







    # 保存用户消息(含skill/agent/kb/mcp选择用于上下文保持)



    user_msg = Message(



        id=gen_uuid(),



        conversation_id=conv_id,



        role="user",



        content=req.content + attachment_block,



        token_used=len(req.content) // 2,



        model_id=req.model_id or conv.model_id,



        skill_ids=req.skill_ids if req.skill_ids else None,



        agent_ids=req.agent_ids if req.agent_ids else None,



        knowledge_ids=req.knowledge_base_ids if req.knowledge_base_ids else None,



        mcp_ids=req.mcp_ids if req.mcp_ids else None,



        attachments=req.attachments if req.attachments else None,



        created_at=cst_now(),



    )



    db.add(user_msg)







    # ===== 模糊创作意图拦截 =====



    # 用户使用创作类动词(制作/创作/做/生成…)但未明确产出形式时，先与用户确认方向，



    # 避免直接调用大模型被内容安全策略拒答(如返回 "high risk")。



    clarify = detect_ambiguous_creative_intent(req.content)



    if clarify:



        clarify_content = "<<CLARIFY>>" + json.dumps(



            {"prompt": req.content, "options": clarify["options"]}, ensure_ascii=False



        ) + "<<CLARIFY>>"



        ai_msg = Message(



            id=gen_uuid(),



            conversation_id=conv_id,



            role="assistant",



            content=clarify_content,



            token_used=0,



            model_id=req.model_id or conv.model_id,



            created_at=cst_now(),



        )



        db.add(ai_msg)



        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": clarify_content,



                    "created_at": str(ai_msg.created_at), "clarify": clarify,



                },



            },



            "message": "请选择创作方向",



        }







    # ===== 核心聊天引擎 =====



    user_id = current_user.id



    model_id = req.model_id or conv.model_id



    prompt_tokens = 0



    completion_tokens = 0



    total_tokens = 0



    ai_content = ""







    ai_model = None



    skills = []  # 初始化避免UnboundLocalError



    agents = []  # 初始化避免UnboundLocalError



    media_generating = False  # 提前初始化：外部Agent分支与末尾均会引用



    external_handled = False   # 是否已由外部Agent(Dify/n8n/Coze)远程处理







    # ===== 外部 Agent 远程调用分支 (Dify / n8n / Coze) =====



    # 若本轮对话关联了带有效 agent_endpoint 的外部 Agent，则直接远程调用其平台，



    # 跳过本地 LLM「扮演」逻辑。effective_agent_ids 已在前面从请求/上下文继承。



    external_agents = _resolve_market_agents(db, effective_agent_ids, user_id)



    external_agent = next((a for a in external_agents if a.get("agent_endpoint")), None)



    if external_agent:



        agents = external_agents



        # 组装最近历史(供 n8n 等无平台会话态的服务维持多轮上下文)



        _hist_rows = db.query(Message).filter(



            Message.conversation_id == conv_id



        ).order_by(asc(Message.created_at)).limit(10).all()



        _history = [{"role": h.role, "content": h.content}



                    for h in _hist_rows if h.role in ("user", "assistant")]



        try:



            answer, new_ext_id = await _call_external_agent(



                db, external_agent,



                query=req.content + attachment_block,



                user_id=user_id,



                ext_conversation_id=conv.ext_agent_conversation_id,



                history=_history,



            )



            ai_content = answer or "外部 Agent 未返回内容。"



            if new_ext_id:



                conv.ext_agent_conversation_id = new_ext_id



        except Exception as e:



            logger.warning(



                f"[external_agent] 远程调用失败 platform={external_agent.get('platform')} "



                f"endpoint={external_agent.get('agent_endpoint')}: {e}"



            )



            ai_content = f"⚠️ 外部 Agent 调用失败：{str(e)[:200]}"



        external_handled = True







    # 解析模型：支持 Auto 模式（model_id='auto' 或未指定时，按能力自动匹配平台默认模型）



    if not external_handled:



        if model_id and model_id != "auto":



            ai_model = db.query(AIModel).filter(AIModel.id == model_id).first()



        if not ai_model:



            # Auto / 未指定 → 按用户输入意图匹配对应能力的默认模型，否则回退文本对话



            intent = detect_media_intent(req.content)



            resolved = resolve_auto_model(intent, db) if intent else None



            ai_model = resolved or resolve_auto_model("text", db)







    if ai_model:



        # ====== 图像/视频生成模型分支 ======



        caps = [c.lower() for c in (ai_model.capabilities or [])] if ai_model else []



        media_generating = False



        if ai_model and ("image" in caps or "video" in caps):



            # 异步生成：立即返回占位内容，后台任务轮询 AGNES 并回写，



            # 避免最长 5 分钟的视频轮询阻塞 HTTP 请求导致客户端超时断开。



            media_generating = True



            if "video" in caps:



                ai_content = "⏳ 视频生成中…（通常 1~3 分钟，完成后自动刷新）"



            else:



                ai_content = "⏳ 图像生成中…（完成后自动刷新）"



        elif 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:



                    # Stage A+B: 解析Skills和Agents



                    skills = _resolve_market_skills(db, effective_skill_ids, user_id)



                    agents = _resolve_market_agents(db, effective_agent_ids, user_id)



                    mcps = _resolve_active_mcp_servers(db, effective_mcp_ids)



                    # MCP 工具调用链路：拉取各 MCP 工具并转为 function definitions

                    mcp_servers_conn = mcps

                    mcp_function_definitions = []

                    mcp_tool_map = {}

                    mcp_idx = 0

                    for _srv in mcp_servers_conn:

                        try:

                            _tools = await list_mcp_tools(_srv)

                        except Exception as _e:

                            logger.warning(f"MCP 拉取工具失败 [{_srv.get('name')}]: {_e}")

                            _tools = []

                        if _tools:

                            try:

                                _upd = db.query(MCPServer).filter(MCPServer.id == _srv["id"]).first()

                                if _upd:

                                    _upd.tools_count = len(_tools)

                                    db.commit()

                            except Exception:

                                pass

                        for _t in _tools:

                            _fname = f"mcp_tool_{mcp_idx}"

                            mcp_tool_map[_fname] = {

                                "server_id": _srv["id"],

                                "name": _t.get("name"),

                                "server_name": _srv["name"],

                            }

                            _schema = _t.get("inputSchema") or {"type": "object", "properties": {}}

                            if not isinstance(_schema, dict) or _schema.get("type") != "object":

                                _schema = {"type": "object", "properties": {}}

                            mcp_function_definitions.append({

                                "name": _fname,

                                "description": f"[MCP·{_srv['name']}] {_t.get('description', '')}",

                                "parameters": _schema,

                            })

                            mcp_idx += 1

                    if mcp_function_definitions:

                        effective_use_tools = True







                    # Issue 8: 默认技能自动发现（仅当用户未显式选择技能时启用）



                    # 每轮先寻找本账号已安装且平台已部署的技能，多个适用时由模型自行判断选 1 个最优。



                    used_auto_skills = False



                    if not effective_skill_ids:



                        try:



                            discovered = _discover_installed_skills(db, user_id)



                        except Exception as e:



                            logger.warning(f"自动发现已安装技能失败: {e}")



                            discovered = []



                        if discovered:



                            skills = discovered



                            used_auto_skills = True



                            effective_use_tools = True







                    # Stage B.1: Agent可能自带skill_ids和kb_ids



                    agent_extra_skill_ids = []



                    agent_extra_kb_ids = []



                    agent_model_override = None



                    for agent in agents:



                        if agent.get("skill_ids"):



                            agent_extra_skill_ids.extend(agent["skill_ids"])



                        if agent.get("knowledge_base_ids"):



                            agent_extra_kb_ids.extend(agent["knowledge_base_ids"])



                        if agent.get("model_id"):



                            agent_model_override = agent["model_id"]







                    # 合并Agent自带的skills和KBs



                    all_skill_ids = list(set(effective_skill_ids + agent_extra_skill_ids))



                    all_kb_ids = list(set(effective_kb_ids + agent_extra_kb_ids))







                    if agent_extra_skill_ids or agent_extra_kb_ids:



                        # 重新解析合并后的skills



                        skills = _resolve_market_skills(db, all_skill_ids, user_id)



                        effective_use_tools = True







                    # Stage C: 增强版KB检索



                    kb_context = _retrieve_kb_context(db, all_kb_ids, req.content)







                    # Stage B: 动态组装system prompt + function definitions



                    system_prompt, function_definitions = _build_system_prompt(



                        skills=skills,



                        agents=agents,



                        kb_context=kb_context,



                        user_content=req.content,



                        mcp=mcps,

                        mcp_functions=mcp_function_definitions,

                        auto_discovered=used_auto_skills,

                    )







                    # 构建消息列表



                    messages = [{"role": "system", "content": system_prompt}]







                    # 加载对话历史(最多10条)



                    history = db.query(Message).filter(



                        Message.conversation_id == conv_id



                    ).order_by(asc(Message.created_at)).limit(10).all()



                    for h in history:



                        if h.role == "user" or h.role == "assistant":



                            messages.append({"role": h.role, "content": h.content})







                    # Issue 1: 必须把附件内容(attachment_block)一并发给模型，



                    # 否则用户在对话窗上传文件后模型“看不到”文件内容。



                    messages.append({"role": "user", "content": req.content + attachment_block})







                    # Stage B.3: Function Calling循环(最多5轮)



                    max_tool_rounds = 5



                    for round_num in range(max_tool_rounds):



                        request_body = {



                            "model": ai_model.name,



                            "messages": messages,



                            "max_tokens": 2048,



                        }



                        if function_definitions and effective_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 120,



                            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=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 effective_use_tools:



                                # 将assistant消息(含tool_calls)加入上下文



                                messages.append({



                                    "role": "assistant",



                                    "content": message.get("content") or "",



                                    "tool_calls": tool_calls,



                                })







                                # Stage A.2: 真实Skill执行



                                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}







                                    if func_name.startswith("mcp_tool_"):

                                        # MCP 工具调用：代理到对应 MCP 服务执行 tools/call

                                        _info = mcp_tool_map.get(func_name)

                                        tool_result = "[MCP 工具未找到或映射缺失]"

                                        if _info:

                                            _srv = next(

                                                (s for s in mcp_servers_conn if s["id"] == _info["server_id"]),

                                                None,

                                            )

                                            if _srv:

                                                tool_result = await call_mcp_tool(

                                                    _srv, _info["name"], func_args

                                                )

                                    else:

                                        tool_result = _execute_skill_tool(

                                            db, func_name, func_args, skills, req.content

                                        )



                                    messages.append({



                                        "role": "tool",



                                        "tool_call_id": tc.get("id", f"call_{round_num}"),



                                        "content": tool_result,



                                    })







                                # 继续下一轮让模型处理tool结果



                                continue



                            else:



                                # 普通回复



                                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:



                    logger.warning(f"AI API调用异常: {e}")



                    ai_content = "AI服务暂时不可用, 请稍后重试或检查模型API配置。"



            else:



                ai_content = "该模型缺少API密钥或端点配置, 请在后台管理-大模型库中配置API Key和Endpoint。"



        else:



            ai_content = "请选择一个已配置的AI模型进行对话。"



    elif not external_handled:



        ai_content = "请先在后台管理-大模型库配置默认对话模型（或聊天中选择模型），Auto 模式下将自动使用平台默认模型。"







    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=ai_model.id if ai_model else (req.model_id or conv.model_id),



        skill_ids=req.skill_ids if req.skill_ids else None,



        agent_ids=req.agent_ids if req.agent_ids else None,



        knowledge_ids=req.knowledge_base_ids if req.knowledge_base_ids else None,



        created_at=cst_now(),



    )



    db.add(ai_msg)







    # 后台异步：将实质性文本回答入库到 Dify 知识库（量化后可供 RAG 复用）



    _ingest_text = (ai_content or "").strip()



    if len(_ingest_text) > 120 and not (



        _ingest_text.startswith("[MEDIA")



        or _ingest_text.startswith("data:")



        or _ingest_text.startswith("http")



        or "<img" in _ingest_text



        or "![]( " in _ingest_text



    ):



        try:



            asyncio.create_task(_ingest_answer_to_dify(conv_id, ai_msg.id, _ingest_text))



        except Exception:



            pass







    # 更新used_skills



    used_skills = []



    if skills:



        used_skills = [s["name"] for s in skills]



        ai_msg.used_skills = used_skills







    # 更新对话统计



    conv.knowledge_graph_count = (conv.knowledge_graph_count or 0) + 1



    conv.updated_at = cst_now()







    # 记录Token使用（媒体生成走 _generate_media_task 异步补写真实消耗，此处跳过避免占位文本重复记账）



    if not media_generating:



        token_usage = TokenUsageLog(



            id=gen_uuid(),



            user_id=current_user.id,



            model_id=ai_model.id if ai_model else (req.model_id or conv.model_id),



            model_name=ai_model.name if 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=cst_now(),



        )



        db.add(token_usage)







    # 进化轨迹记录



    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=cst_now(),



            )



            db.add(trajectory)



            db.flush()







            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)







    # 媒体生成改为后台异步：立即返回占位，后台任务完成后回写消息内容



    if media_generating:



        try:



            asyncio.create_task(



                _generate_media_task(ai_msg.id, ai_model.id, req.content, current_user.id)



            )



        except Exception as e:



            logger.warning(f"[_generate_media_task] 启动后台任务失败: {e}")







    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,



                "generating": media_generating,



            },



        },



        "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 = cst_now()



    db.commit()







    return {"code": 0, "message": "对话上下文已清空"}











def _build_knowledge_graph(db: Session, kb: KnowledgeBase) -> dict:



    """从知识库文档构建知识图谱节点和边，返回 graph_data。"""



    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),



        },



    }







    kb.knowledge_graph_count = len(graph_nodes)



    kb.updated_at = cst_now()



    return graph_data











@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), asc(Message.id))



        .all()



    )







    if not messages:



        raise HTTPException(status_code=400, detail="对话中没有消息, 无法生成知识图谱")







    # 幂等：若该对话已生成过知识图谱且知识库仍存在，则直接刷新返回



    if conv.knowledge_graph_base_id:



        existing_kb = db.query(KnowledgeBase).filter(



            KnowledgeBase.id == conv.knowledge_graph_base_id



        ).first()



        if existing_kb:



            graph_data = _build_knowledge_graph(db, existing_kb)



            conv.knowledge_graph_count = 1



            conv.updated_at = cst_now()



            db.commit()



            return {



                "code": 0,



                "data": {



                    "knowledge_base_id": existing_kb.id,



                    "knowledge_base_name": existing_kb.name,



                    "graph": graph_data,



                    "knowledge_graph_count": 1,



                    "already_exists": True,



                },



                "message": "知识图谱已存在, 已刷新展示",



            }







    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=cst_now(),



        updated_at=cst_now(),



    )



    db.add(kb)



    db.flush()







    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')),



            content=msg.content,



            status="completed",



            chunk_count=max(1, len(msg.content) // 500 + 1),



            created_at=cst_now(),



            updated_at=cst_now(),



        )



        db.add(doc)



        doc_count += 1







    kb.document_count = doc_count



    db.flush()







    # 构建知识图谱节点和边



    graph_data = _build_knowledge_graph(db, kb)



    conv.knowledge_graph_base_id = kb.id



    conv.knowledge_graph_count = 1



    conv.updated_at = cst_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}",



    )







    return {



        "code": 0,



        "data": {



            "knowledge_base_id": kb.id,



            "knowledge_base_name": kb.name,



            "graph": graph_data,



            "knowledge_graph_count": 1,



            "already_exists": False,



        },



        "message": f"知识图谱构建完成",



    }











@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,



            "display_name": getattr(m, "display_name", None) or m.name,



            "provider_id": m.provider_id,



            "provider_name": provider.name if provider else "",



            "capabilities": m.capabilities or [],



            "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"}