"""
增强版 Chat API — Phase 2.5
包含: Skill ID统一(MarketApp映射), 真实Skill执行, Agent工作流注入,
      多轮上下文保持, BM25 KB检索, Function Calling循环增强
"""
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from sqlalchemy import and_, desc, asc, func
from pydantic import BaseModel, Field
from typing import Optional
import uuid
from datetime import datetime, timezone, timedelta
import re
import json
import asyncio
import math
import httpx
import os
import logging

from ..database import get_db, SessionLocal
from ..models import (
    User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog,
    KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill,
    EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation,
    gen_uuid
)
from ..auth import get_current_user
from ..platform_knowledge import get_platform_knowledge
from ..middleware.log_service import LogService
from ..capability_registry import resolve_auto_model

router = APIRouter()

logger = logging.getLogger(__name__)


def detect_media_intent(text: str) -> str | None:
    """根据用户输入判断是否为 图像/视频/语音 生成请求，返回对应能力或 None。

    用于 Auto 模式：用户未指定具体模型时，依据意图自动匹配平台各类能力的默认模型。
    疑问句（含「吗/呢/怎么/能否」或结尾问号）不触发，避免「你能生成视频吗」被误判。
    """
    if not text:
        return None
    t = text.lower().strip()
    # 疑问句不触发媒体生成
    if t.endswith(("？", "?")) or any(q in t for q in ("吗", "呢", "怎么", "如何", "能否", "是否可以", "可以吗", "会不会", "why", "how")):
        return None
    video_kw = ("视频", "短片", "影片", "vide", "动画视频", "一段视频", "做个视频", "生成视频",
                "视频生成", "来段视频", "拍个视频", "做段视频", "视频剪辑", "做视频",
                "生成一个视频", "制作一个视频", "帮我做视频", "制作视频",
                "生成短片", "做短片", "制作短片", "一段短片")
    image_kw = ("图片", "图像", "插画", "海报", "照片", "头像", "壁纸", "漫画", "表情包", "logo", "图标",
                "imag", "画一张", "画张", "画一只", "画只", "画个", "画幅",
                "来张", "来一张", "来幅", "发张图", "发个图",
                "生成图片", "生成一张", "生成一幅", "生成图", "图像生成",
                "做一张", "做张图", "做张", "出张图", "出一张图",
                "帮我画", "给我画", "请画", "绘制", "作画", "ai绘画", "ai画图",
                "画个图", "画张图", "画一幅图", "一张图", "一幅画", "生成一幅",
                "画画", "画一下", "画出来", "生成个图", "生成一张图", "生成个图片",
                "做图片", "做个图", "制图", "整张图", "整张图出来", "搞张图",
                "帮我生成图片", "帮我生成图像", "给我生成图片", "给我生成图像",
                "生成一张图片", "生成一个图片", "制作一张图", "制作一张图片",
                "来一张图片", "画一张图片", "画一幅", "画一副")
    audio_kw = ("语音", "配音", "朗读", "播报", "念一段", "有声", "tts", "音频", "生成语音",
                "读一下", "念一下", "语音合成")
    if any(k in t for k in video_kw):
        return "video"
    if any(k in t for k in image_kw):
        return "image"
    if any(k in t for k in audio_kw):
        return "audio"
    return None


def detect_ambiguous_creative_intent(text: str):
    """检测「模糊创作意图」：用户使用了创作类动词(制作/创作/做/生成/画/写/设计…)，
    但未明确产出形式(文案/图片/视频/其他)。此时应与用户确认方向，而不是直接调用大模型
    （某些模型会对这类开放式创作请求返回 "high risk" 拒答）。

    返回 None 表示无需确认；否则返回 {"options": [...]}。
    """
    if not text:
        return None
    t = text.strip()
    tl = t.lower()
    # 已有明确媒体意图(图像/视频/语音)则无需确认
    if detect_media_intent(t):
        return None
    # 疑问句不触发
    if tl.endswith(("？", "?")) or any(q in tl for q in ("吗", "呢", "怎么", "如何", "能否", "是否可以", "可以吗", "会不会", "why", "how", "什么", "哪些")):
        return None
    # 已明确给出产出形式(文字类/其他类)则无需确认
    explicit_medium = (
        "文案", "文章", "诗歌", "诗", "小说", "故事", "代码", "程序", "脚本", "总结", "翻译",
        "邮件", "报告", "段落", "正文", "文字", "写作", "ppt", "演示", "思维导图", "表格",
        "图表", "数据",
    )
    if any(m in tl for m in explicit_medium):
        return None
    # 创作类动词
    creative_verbs = (
        "制作", "创作", "做个", "做一个", "搞一个", "产出", "构思", "设想",
        "打造", "塑造", "生成", "帮我做", "帮我生成", "帮我画", "帮我写", "请制作",
        "请创作", "请生成", "请画", "请写", "来一个", "来一张", "来一幅", "做一张",
        "做张", "写一首", "写一段", "写个",
    )
    if not any(v in t for v in creative_verbs):
        return None
    # 模糊：有创作动词但无明确产出形式 → 提供方向选择
    options = [
        {"key": "text", "label": "文案", "desc": "文章 / 故事 / 诗歌等文字内容", "medium": "文案"},
        {"key": "image", "label": "图片", "desc": "插画 / 海报 / 照片等图像", "medium": "图片"},
        {"key": "video", "label": "视频", "desc": "短片 / 动画等视频", "medium": "视频"},
        {"key": "other", "label": "其他", "desc": "语音 / 代码 / 其他形式", "medium": "其他"},
    ]
    return {"options": options}

# CST timezone (UTC+8) for consistent time display
CST_TZ = timezone(timedelta(hours=8))


def cst_now():
    """Get current time in CST timezone"""
    return datetime.now(CST_TZ)


def _html_escape(s: str) -> str:
    return (str(s).replace("&", "&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"


def _proxy_for_url(url: str) -> str | None:
    """按目标域名选择是否走代理。

    实测(2026-07-15)：apihub.agnes-ai.com / platform-outputs.agnes-ai.space 为国内域名，
    容器内直连稳定返回 200；而经智能路由代理 172.18.0.1:7890 当前返回 502，导致
    httpx 报 "Server disconnected without sending a response"。
    因此对 AGNES 等国内域名返回 None(直连)，其余域名才走 _outbound_proxy()。

    注意：httpx 默认 trust_env=True 会忽略 proxy=None 而改用环境变量 HTTP_PROXY，
    故调用方必须同时传 trust_env=False 才能真正直连。
    """
    from urllib.parse import urlparse
    host = (urlparse(url).hostname or "").lower()
    if "agnes-ai.com" in host or "agnes-ai.space" in host:
        return None
    # 平台自有域名(含自建 Dify/n8n 等)必须直连，不经出网代理。
    if "qihuyun.com" in host:
        return None
    return _outbound_proxy()


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},
            ]
            body = {"model": text_model.name, "messages": messages, "max_tokens": 512, "temperature": 0.7}
            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


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（用户原始输入含多轮对话噪音 → 高质量图像生成提示词）
        optimized_prompt = await _optimize_media_prompt(prompt, "image", db)
        payload = {
            "model": ai_model.name,
            "prompt": optimized_prompt,
            "size": "1024x768",
        }
        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 != 200:
                return f"⚠️ 图像生成失败 (HTTP {resp.status_code}): {resp.text[:200]}"
            data = resp.json()
            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)}')
        except Exception as e:
            return f"⚠️ 图像生成异常：{str(e)[:200]}"

    # ============ 视频生成 (异步任务 + 轮询) ============
    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（用户输入含多轮对话噪音 → 高质量视频提示词）
        optimized_prompt = await _optimize_media_prompt(prompt, "video", db)
        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,
        }
        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 != 200:
                    return f"⚠️ 视频生成失败 (HTTP {resp.status_code}): {resp.text[:200]}"
                data = resp.json()
                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)
                    r2 = await client.get(f"{base}/v1/videos/{video_id}", headers=headers)
                    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 '生成超时，请稍后在对话中重试'}"
        except Exception as e:
            return f"⚠️ 视频生成异常：{str(e)[:200]}"

    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} 已回写内容")
    except Exception as e:
        logger.warning(f"[_generate_media_task] 后台生成异常: {e}")
    finally:
        try:
            db.close()
        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 _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,
    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)
    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_market_mcp(db, effective_mcp_ids, user_id)

                    # 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,
                        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}

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