chore: normalize tsconfig.json line endings (file missed by prior commit)

The earlier renormalize commit (771c845) missed frontend/tsconfig.json; this
completes the LF normalization so all tracked text files are LF. .gitattributes
(* text=auto eol=lf) prevents future CRLF.

.gitattributes

@@ -0,0 +1,33 @@
+# Normalize all text files to LF on checkout/commit (root-cause fix for mixed line endings)
+* text=auto eol=lf
+
+# Binary assets: never normalize line endings or diff as text
+*.png binary
+*.jpg binary
+*.jpeg binary
+*.gif binary
+*.ico binary
+*.webp binary
+*.woff binary
+*.woff2 binary
+*.ttf binary
+*.eot binary
+*.otf binary
+*.pdf binary
+*.zip binary
+*.tar binary
+*.gz binary
+*.bz2 binary
+*.7z binary
+*.pyc binary
+*.so binary
+*.dll binary
+*.exe binary
+*.db binary
+*.sqlite binary
+*.sqlite3 binary
+*.wasm binary
+*.mp4 binary
+*.mp3 binary
+*.avi binary
+*.mov binary

backend/python/app/api/admin/mcp_admin.py

@@ -842,110 +842,29 @@ def _test_mcp_server(server, db) -> dict:
 
 
         if transport in ("streamable_http", "streamablehttp"):
-
-
-
             url = server.url
-
-
-
             if not url:
-
-
-
                 result["message"] = "Streamable HTTP endpoint URL is required"
-
-
-
             else:
-
-
-
-                import httpx
-
-
-
                 try:
-
+                    # Use the real MCP SDK to verify connectivity (same as chat flow).
-
+                    # A naive HTTP GET/POST can give false negatives for endpoints
-
+                    # that only accept proper MCP JSON-RPC over POST (e.g., AnySearch).
-                    # A real streamable-http MCP endpoint accepts POST JSON-RPC.
+                    from ...mcp_client import list_mcp_tools as _lmt
-
+                    hdrs = json.loads(server.headers) if isinstance(server.headers, str) else (server.headers or {})
-
+                    sd = {"url": url, "transport_type": "streamable_http", "headers": hdrs}
-
+                    import asyncio as _aio
-                    # Some implementations also accept GET for SSE events.
+                    tools = _aio.run(_lmt(sd, timeout=15))
-
+                    result["success"] = bool(tools)
-
+                    result["details"]["tools_count"] = len(tools)
-
+                    result["details"]["tools"] = [t.get("name") for t in tools[:5]]
-                    # Try GET first (cheapest), fall back to a minimal POST.
+                    result["message"] = f"Connected — {len(tools)} tools available" if tools else "Connected but no tools found"
-
+                    result["details"]["status_code"] = 200
-
+                    if tools:
-
+                        server.tools_count = len(tools)
-                    resp = httpx.get(url, headers=req_headers, timeout=10.0, follow_redirects=True)
+                        server.status = "connected"
-
-
-
-                    if resp.status_code == 405:
-
-
-
-                        resp = httpx.post(
-                            url,
-                            headers=req_headers,
-                            json={"jsonrpc": "2.0", "method": "initialize", "id": 1},
-                            timeout=10.0,
-                            follow_redirects=True,
-                        )
-
-
-
-                    # Any HTTP response (including 4xx) means the host is reachable.
-
-
-
-                    result["success"] = resp.status_code < 500
-
-
-
-                    result["message"] = f"HTTP responded with {resp.status_code}"
-
-
-
-                    result["details"]["status_code"] = resp.status_code
-
-
-
-                    result["details"]["content_type"] = resp.headers.get("content-type", "")
-
-
-
-                except httpx.ConnectError as e:
-
-
-
-                    result["message"] = f"Connection refused - server may be down: {str(e)}"
-
-
-
-                except httpx.TimeoutException:
-
-
-
-                    result["message"] = "Connection timed out"
-
-
-
                 except Exception as e:
-
+                    result["message"] = f"Connection failed: {str(e)}"
-
-
-                    result["message"] = f"HTTP request failed: {str(e)}"
-
-
-
-
-
-
 
         elif transport == "sse":
 
@@ -3156,7 +3075,7 @@ async def test_mcp_connection(
 
 
 
-    result = _test_mcp_server(server, db)
+    result = await asyncio.get_running_loop().run_in_executor(None, _test_mcp_server, server, db)
 
 
 

backend/python/app/api/chat_skill.py

@@ -1593,3 +1593,150 @@ def _execute_skill_tool(
 
 
     return result
+
+
+def _generate_install_fallback_skill(name: str, description: str) -> str:
+    """install_app 兜底：无法识别为 SKILL.md 时，生成一份可用技能定义。"""
+    desc = (description or "").strip() or f"{name} 相关能力"
+    return (
+        f"---\n"
+        f"name: {name}\n"
+        f"description: {desc}\n"
+        f"category: other\n"
+        f"---\n\n"
+        f"# {name}\n\n"
+        f"{desc}\n\n"
+        f"## 使用建议\n"
+        f"- 明确用户意图后再执行，必要时先向用户确认方向。\n"
+        f"- 输出结构清晰、可直接使用。\n"
+        f"- 如任务超出能力范围，主动说明并给出替代方案。\n"
+    )
+
+
+def _execute_install_app(db: Session, url: str, user_id, app_type: str = "auto") -> str:
+    """前台 AI 聊天：通过 URL 安装 skill / mcp / agent（根治问题 #3）。
+
+    逻辑对齐 market.upload_app，但本函数不引用 market 私有函数以避免循环 import：
+    - 同步 httpx.Client(trust_env=False, 显式 proxy) 抓取 URL 内容（与 _fetch_skill_content 同款）
+    - 自动识别类型：mcp / agent / skill
+    - 建 MarketApp（+ mcp 时 MCPServer + AppInstallation），自动标记为已安装
+
+    关键约束：本函数在聊天主事件循环中【同步】执行，禁止调用 asyncio.run
+    （会抛 "cannot be called from a running event loop"）。因此 MCP 连接测试交由后台
+    「测试」按钮在 worker 线程异步执行；新建 MCPServer 直接 is_enabled=True 即可被
+    聊天链路 _resolve_active_mcp_servers 使用（该解析仅过滤 is_enabled）。
+    """
+    if not url or not str(url).strip().lower().startswith("http"):
+        return "请提供有效的安装地址(URL)，例如 GitHub 仓库、mcp.so 或 modelscope 链接。"
+    url = str(url).strip()
+    proxy = os.environ.get("HTTP_PROXY") or os.environ.get("http_proxy") or None
+    try:
+        with httpx.Client(timeout=25, follow_redirects=True, trust_env=False, proxy=proxy) as client:
+            r = client.get(url, headers={"User-Agent": "Mozilla/5.0 (compatible; QihuAI/1.0)"})
+            content = r.text if r.status_code == 200 else ""
+    except Exception as e:
+        return f"无法访问该地址：{str(e)[:200]}"
+
+    _u = url.lower()
+    _c = (content or "").lower()
+    detected = "skill"
+    if app_type and app_type != "auto":
+        detected = app_type
+    else:
+        if "mcpservers" in _c or '"mcpservers"' in (content or "").lower() or (
+            "url" in _c and ("transport" in _c or '"type"' in _c)
+        ):
+            detected = "mcp"
+        elif any(k in _u for k in ["dify", "n8n", "coze"]) or "agent_endpoint" in _c:
+            detected = "agent"
+        else:
+            detected = "skill"
+
+    import re as _re
+    _name = _re.sub(r"^https?://", "", url).strip("/").split("/")[-1] or "imported_app"
+    _name = _name[:120]
+    _desc = f"通过链接导入：{url}"
+
+    _dup = db.query(MarketApp).filter(
+        func.lower(MarketApp.name) == _name.lower(),
+        MarketApp.type == detected,
+    ).first()
+    if _dup:
+        return f"「{_name}」({detected}) 已存在，无需重复安装（ID: {_dup.id}）。可在应用市场直接使用。"
+
+    app = MarketApp(
+        id=gen_uuid(),
+        name=_name,
+        description=_desc,
+        type=detected,
+        source="user",
+        category="other",
+        tags=None,
+        price=0,
+        install_count=0,
+        rating_avg=0,
+        review_count=0,
+        publisher_id=user_id,
+        status="published",
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+    )
+
+    if detected == "mcp":
+        from ..api.admin.mcp_admin import parse_mcp_config
+        parsed = parse_mcp_config(content)
+        _murl = parsed.get("url") or url
+        app.skill_content = content
+        app.mcp_transport = parsed.get("transport_type", "streamable_http")
+        db.add(app)
+        db.commit()
+        db.refresh(app)
+        _hdr = json.dumps(parsed.get("headers") or {}, ensure_ascii=False)
+        server = MCPServer(
+            id=app.id,
+            name=app.name,
+            description=app.description or "",
+            transport_type=parsed.get("transport_type", "streamable_http"),
+            command="",
+            args="{}",
+            env_vars="{}",
+            url=_murl,
+            headers=_hdr,
+            is_enabled=True,
+            status="disconnected",
+            tools_count=0,
+            resources_count=0,
+            created_at=app.created_at,
+            updated_at=app.updated_at,
+        )
+        db.add(server)
+        db.commit()
+    elif detected == "agent":
+        app.agent_endpoint = url
+        if "dify" in _u:
+            app.platform = "dify"
+        elif "n8n" in _u:
+            app.platform = "n8n"
+        elif "coze" in _u:
+            app.platform = "coze"
+        app.skill_content = content if "{" in content else None
+        db.add(app)
+        db.commit()
+        db.refresh(app)
+    else:
+        _is_skillmd = ("name:" in _c and "description:" in _c) or _u.endswith(".md")
+        app.skill_content = content if _is_skillmd else _generate_install_fallback_skill(_name, _desc)
+        db.add(app)
+        db.commit()
+        db.refresh(app)
+
+    if user_id:
+        _inst = db.query(AppInstallation).filter(
+            AppInstallation.user_id == user_id,
+            AppInstallation.app_id == app.id,
+        ).first()
+        if not _inst:
+            db.add(AppInstallation(user_id=user_id, app_id=app.id))
+            db.commit()
+
+    return f"已成功安装「{_name}」({detected})。你现在可以直接在对话中使用它，或在应用市场查看。"

backend/python/app/api/im_bot_user.py

@@ -458,9 +458,8 @@ def _wechat_receiver_loop(account_id: int):
                     except Exception:
                         pass
                     if ret in (-2, -14):
-                        # 会话/鉴权失效：标记需重扫并停止接收线程
+                        # 会话/鉴权失效：标记需重扫但继续循环，让下一轮 poll 自然检测过期
                         _mark_wechat_expired(d["id"])
-                        return
             time.sleep(1)
         except Exception:
             time.sleep(5)

backend/python/app/api/im_llm.py

@@ -4,13 +4,16 @@ IM Bot 真实 LLM 调用。
 原实现位于 channels.py（通用频道管理模块），该模块已移除。
 为保持 IM Bot（微信机器人等）正常生成回复，将 _call_real_llm 迁移至此独立模块。
 
-行为（对齐 AI 聊天）：
+行为（对齐 AI 聊天 — 2026-07-23 同步至最新）：
 - Issue 7：图像/视频/语音意图优先走平台媒体模型（与 AI 聊天同一套 Auto 路由），
   不再对"制作图片"这类请求只回文本。
 - Issue 8：默认 Auto 模型；每轮先自动发现本账号已安装且平台已部署的技能，
   多个适用时由模型自行判断选 1 个最优，通过 function calling 执行。
 - 2026-07-15：修复 trust_env=False + proxy 处理，避免容器坏代理导致调用失败；
   媒体生成前先经 _optimize_media_prompt 精炼提示词，提升 AGNES 生成质量。
+- 2026-07-23：对齐 Web Chat 最新能力：
+  * 模型 failover（Auto 模式自动切换备选模型）
+  * 平台知识库注入（get_platform_knowledge，技能清单实时同步）
 """
 import asyncio
 import json
@@ -21,6 +24,7 @@ from typing import Optional
 from urllib.parse import urlparse as _urlparse
 
 import httpx
+from app.core.llm_client import chat_complete  # 统一模型层(07-24): web/IM 共用标准 OpenAI SDK 客户端
 from sqlalchemy.orm import Session
 
 from ..capability_registry import resolve_auto_model
@@ -83,7 +87,7 @@ def _run_async(coro, timeout=None):
 def _log_media_usage(db: Session, user_id: str, ai_model, media_type: str, prompt: str):
     """IM Bot 媒体生成记账：把图片/视频消耗写入 TokenUsageLog，供驾驶舱统计。
 
-    此前 IM Bot 路径完全不写 TokenUsageLog，导致微信生成的图片/视频在驾驶舱里“查无此项”。
+    此前 IM Bot 路径完全不写 TokenUsageLog，导致微信生成的图片/视频在驾驶舱里"查无此项"。
     媒体 API 不返回 token 用量，这里用与 chat.py _generate_media_task 一致的估算。
     """
     if not user_id or not ai_model:
@@ -112,7 +116,7 @@ def _log_media_usage(db: Session, user_id: str, ai_model, media_type: str, promp
 
 
 def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im", user_id: str = None) -> "str | dict":
-    """调用七虎AI 生成回复，供 IM Bot 复用。支持媒体意图路由与已安装技能自动发现。"""
+    """调用七虎AI 生成回复，供 IM Bot 复用。支持媒体意图路由、已安装技能自动发现、模型 failover。"""
     try:
         from app.api.chat import (
             _generate_media,
@@ -151,19 +155,23 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im
                         return s
                     return "（媒体生成未能返回有效内容，请重试）"
 
-        # ===== 2) 文本回复 + 技能自动发现（Issue 8）=====
+        # ===== 2) 文本回复：模型 failover + 技能自动发现 + 平台知识库注入 =====
-        ai_model = resolve_auto_model("text", db)
+        from ..platform_knowledge import get_platform_knowledge
-        if not ai_model:
-            return "（未配置默认文本模型，无法生成回复）"
-        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 not (api_key and api_url):
-            return "（模型未配置 API Key 或地址，无法生成回复）"
 
-        # 自动发现本账号已安装且平台已部署的技能
+        # 构建候选模型列表（对齐 Web Chat failover）
+        ai_model = resolve_auto_model("text", db)
+        _try_models = [ai_model] if (ai_model and ai_model.is_enabled) else []
+        # Auto 模式附加备选（IM Bot 始终 Auto）
+        _fallbacks = db.query(AIModel).filter(
+            AIModel.is_enabled == True,
+            AIModel.id != (ai_model.id if ai_model else ""),
+        ).order_by(AIModel.is_default.desc()).limit(4).all()
+        _try_models.extend([m for m in _fallbacks if m not in _try_models])
+
+        if not _try_models:
+            return "（未配置可用文本模型，无法生成回复）"
+
+        # 技能自动发现（提前执行一次，不必每个候选模型重复做）
         skills = []
         if user_id:
             try:
@@ -171,10 +179,33 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im
             except Exception:
                 skills = []
 
+        # 平台知识库注入（对齐 Web Chat——微信用户提问"平台有哪些技能"也能得到完整清单）
+        kb_context = ""
+        try:
+            pk = get_platform_knowledge(db=db)
+            if pk:
+                kb_context = f"【平台知识库——优先使用以下知识回答用户问题】\n{pk}"
+        except Exception:
+            pass
+
+        # 尝试每个候选模型
+        _retry_log = []
+        for _try_model in _try_models:
+            ai_model = _try_model
+
+            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 not (api_key and api_url):
+                _retry_log.append(f"{ai_model.name}: 缺少密钥或端点")
+                continue
+
             system_prompt, function_definitions = _build_system_prompt(
                 skills=skills,
                 agents=[],
-            kb_context="",
+                kb_context=kb_context,
                 user_content=prompt,
                 mcp=[],
                 auto_discovered=bool(skills),
@@ -184,28 +215,31 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im
                 {"role": "user", "content": prompt},
             ]
 
+            # 尝试调用
+            try:
                 max_rounds = 3
                 ai_content = ""
-        for _ in range(max_rounds):
+                _http_ok = True
+                for round_num in range(max_rounds):
                     request_body = {"model": ai_model.name, "messages": messages, "max_tokens": 2048}
                     if function_definitions and skills:
                         request_body["tools"] = [
                             {"type": "function", "function": fd} for fd in function_definitions
                         ]
                         request_body["tool_choice"] = "auto"
-            with httpx.Client(
+                    try:
-                timeout=ai_model.timeout or 120,
+                        data = chat_complete(
-                proxy=_proxy_for_url(api_url),
-                trust_env=False,
-            ) as client:
-                resp = client.post(
                             api_url,
-                    headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
+                            api_key,
-                    json=request_body,
+                            request_body,
+                            proxy=_proxy_for_url(api_url),
+                            timeout=ai_model.timeout or 120,
                         )
-                if resp.status_code != 200:
+                    except Exception as _exc:
-                    return f"（AI 模型调用失败 HTTP {resp.status_code}）"
+                        _retry_log.append(f"{ai_model.name}: {str(_exc)[:60]}")
-                data = resp.json()
+                        logger.warning(f"[im_llm] 模型 {ai_model.name} 异常: {_exc}")
+                        _http_ok = False
+                        break
                         message = data.get("choices", [{}])[0].get("message", {})
                         tool_calls = message.get("tool_calls", [])
                         if tool_calls and skills:
@@ -229,6 +263,20 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im
                             continue
                         ai_content = message.get("content", "")
                         break
+
+                # 检查本轮调用是否成功
+                if _http_ok and ai_content:
                     return ai_content or "（AI 回复为空）"
+                # 本轮失败：记录日志，切换备选
+                logger.warning(f"[im_llm] 模型 {ai_model.name} 失败，切换备选...")
+                continue
+            except Exception as exc:
+                _retry_log.append(f"{ai_model.name}: {str(exc)[:60]}")
+                logger.warning(f"[im_llm] 模型 {ai_model.name} 异常: {exc}")
+                continue
+
+        # 全部失败
+        _summary = "; ".join(_retry_log[-3:])
+        return f"（已尝试 {len(_try_models)} 个模型均调用失败。最后错误: {_summary}。请稍后重试。）"
     except Exception as e:
         return f"（AI 回复生成失败：{e}）"

backend/python/app/core/llm_client.py

@@ -0,0 +1,222 @@
+"""Shared, standards-based LLM client for web chat + IM bot.
+
+平台红线(07-24)：模型层以成熟/开源标准为设计基准。本模块使用官方 OpenAI SDK
+(`openai.AsyncOpenAI`)——它是 OpenAI 兼容端点的事实标准客户端。平台本身已在讲
+OpenAI 协议(`/v1/chat/completions`)，因此零协议改造成本，并天然获得：
+  * 原生 SSE 流式(`stream=True`)；
+  * 结构化的 tool_calls / usage 解析；
+  * 统一的代理(trust_env=False + 显式 proxy)与超时处理。
+
+本模块取代 chat_service.py 与 im_llm.py 中两套手写 httpx 循环，使 web chat 与
+IM bot 共用**同一模型层**、同一套 failover/代理/超时语义——这是「各端能力统一对齐」
+红线在模型层的落地。
+
+API:
+  achat_complete(api_url, api_key, request_body, *, proxy=None, timeout=120, stream=False)
+    - stream=False -> dict，形状对齐上游 `data` 响应：
+        {"choices":[{"message":{"content","tool_calls","reasoning_content"}}], "usage":{...}}
+    - stream=True  -> async generator，逐块产出 ("content"|"reasoning", text) 增量
+"""
+from __future__ import annotations
+
+import httpx
+from openai import AsyncOpenAI, OpenAI
+
+
+def _base_url(api_url: str) -> str:
+    """从完整 /v1/chat/completions 或 /v1/embeddings 地址推导 OpenAI SDK 所需的 base_url。"""
+    u = (api_url or "").rstrip()
+    for suf in ("/v1/chat/completions", "/chat/completions", "/v1/embeddings", "/embeddings"):
+        if u.endswith(suf):
+            return u[: -len(suf)]
+    return u
+
+
+def build_openai_client(api_url, api_key, *, proxy=None, timeout=120) -> AsyncOpenAI:
+    base = _base_url(api_url)
+    # 始终 trust_env=False：避免容器 HTTP_PROXY 对国内端点(siliconflow 等)造成 502；
+    # 代理经 proxy 显式传入（AGNES 等），未传则直连。
+    http_client = httpx.AsyncClient(proxy=proxy, trust_env=False)
+    return AsyncOpenAI(
+        base_url=base,
+        api_key=api_key or "EMPTY",
+        timeout=timeout,
+        http_client=http_client,
+    )
+
+
+def _create_kwargs(request_body: dict, stream: bool = False) -> dict:
+    kw = {
+        "model": request_body.get("model"),
+        "messages": request_body.get("messages", []),
+        "stream": stream,
+    }
+    if request_body.get("temperature") is not None:
+        kw["temperature"] = request_body["temperature"]
+    if request_body.get("top_p") is not None:
+        kw["top_p"] = request_body["top_p"]
+    if request_body.get("max_tokens") is not None:
+        kw["max_tokens"] = request_body["max_tokens"]
+    if request_body.get("stop") is not None:
+        kw["stop"] = request_body["stop"]
+    tools = request_body.get("tools")
+    if tools:
+        kw["tools"] = tools
+        kw["tool_choice"] = request_body.get("tool_choice", "auto")
+    return kw
+
+
+def _normalize_tool_calls(tcs):
+    if not tcs:
+        return None
+    out = []
+    for tc in tcs:
+        out.append(
+            {
+                "id": getattr(tc, "id", None),
+                "type": "function",
+                "function": {
+                    "name": getattr(tc.function, "name", None),
+                    "arguments": getattr(tc.function, "arguments", "") or "",
+                },
+            }
+        )
+    return out
+
+
+async def achat_complete(api_url, api_key, request_body, *, proxy=None, timeout=120, stream=False):
+    """统一的 OpenAI 兼容聊天补全。详见模块 docstring。"""
+    if stream:
+
+        async def _gen():
+            client = build_openai_client(api_url, api_key, proxy=proxy, timeout=timeout)
+            _content_parts = []
+            _reasoning_parts = []
+            _tool_acc = {}  # index -> {id, type, name, arguments}
+            try:
+                gen = await client.chat.completions.create(
+                    **_create_kwargs(request_body, stream=True)
+                )
+                async for chunk in gen:
+                    if not chunk.choices:
+                        continue
+                    delta = chunk.choices[0].delta
+                    rc = getattr(delta, "reasoning_content", None)
+                    if rc:
+                        _reasoning_parts.append(rc)
+                        yield ("reasoning", rc)
+                    ct = getattr(delta, "content", None)
+                    if ct:
+                        _content_parts.append(ct)
+                        yield ("content", ct)
+                    # 累积 OpenAI 流式 tool_calls（供后续工具循环执行）
+                    tcs = getattr(delta, "tool_calls", None)
+                    if tcs:
+                        for tc in tcs:
+                            _idx = getattr(tc, "index", 0)
+                            _slot = _tool_acc.setdefault(
+                                _idx, {"id": None, "type": "function", "name": "", "arguments": ""}
+                            )
+                            _tcid = getattr(tc, "id", None)
+                            if _tcid:
+                                _slot["id"] = _tcid
+                            _fn = getattr(tc, "function", None)
+                            if _fn is not None:
+                                _fn_name = getattr(_fn, "name", None)
+                                if _fn_name:
+                                    _slot["name"] += _fn_name
+                                _fn_args = getattr(_fn, "arguments", None)
+                                if _fn_args:
+                                    _slot["arguments"] += _fn_args
+            finally:
+                await client.close()
+            # 流式结束时吐出已组装的完整 message（含 tool_calls），供调用方无缝衔接工具循环
+            _tool_calls = None
+            if _tool_acc:
+                _tool_calls = []
+                for _i in sorted(_tool_acc.keys()):
+                    _s = _tool_acc[_i]
+                    _tool_calls.append({
+                        "id": _s["id"],
+                        "type": "function",
+                        "function": {"name": _s["name"], "arguments": _s["arguments"]},
+                    })
+            yield ("message", {
+                "content": "".join(_content_parts),
+                "tool_calls": _tool_calls,
+                "reasoning_content": "".join(_reasoning_parts) or None,
+            })
+
+        return _gen()
+
+    client = build_openai_client(api_url, api_key, proxy=proxy, timeout=timeout)
+    async with client:
+        resp = await client.chat.completions.create(**_create_kwargs(request_body))
+        msg = resp.choices[0].message
+        return {
+            "choices": [
+                {
+                    "message": {
+                        "content": msg.content or "",
+                        "tool_calls": _normalize_tool_calls(getattr(msg, "tool_calls", None)),
+                        "reasoning_content": getattr(msg, "reasoning_content", None),
+                    }
+                }
+            ],
+            "usage": (resp.usage.model_dump() if resp.usage else {}),
+        }
+
+
+def build_openai_client_sync(api_url, api_key, *, proxy=None, timeout=120) -> OpenAI:
+    """同步客户端工厂（供 IM bot 等同步上下文复用同一标准模型层）。"""
+    base = _base_url(api_url)
+    http_client = httpx.Client(proxy=proxy, trust_env=False)
+    return OpenAI(base_url=base, api_key=api_key or "EMPTY", timeout=timeout, http_client=http_client)
+
+
+def chat_complete(api_url, api_key, request_body, *, proxy=None, timeout=120):
+    """同步版（与 achat_complete 行为/返回形状完全一致），供非 async 上下文使用。"""
+    client = build_openai_client_sync(api_url, api_key, proxy=proxy, timeout=timeout)
+    with client:
+        resp = client.chat.completions.create(**_create_kwargs(request_body))
+        msg = resp.choices[0].message
+        return {
+            "choices": [
+                {
+                    "message": {
+                        "content": msg.content or "",
+                        "tool_calls": _normalize_tool_calls(getattr(msg, "tool_calls", None)),
+                        "reasoning_content": getattr(msg, "reasoning_content", None),
+                    }
+                }
+            ],
+            "usage": (resp.usage.model_dump() if resp.usage else {}),
+        }
+
+
+# ---------------------------------------------------------------------------
+# 嵌入（Embeddings）：OpenAI 兼容 /v1/embeddings，收敛散落的裸 httpx 嵌入调用
+# ---------------------------------------------------------------------------
+def _embed_kwargs(input, model):
+    return {"model": model, "input": input if isinstance(input, list) else [input]}
+
+
+async def aembed(api_url, api_key, input, model, *, proxy=None, timeout=60) -> list:
+    """统一的 OpenAI 兼容文本嵌入（异步）。返回 list[list[float]]。
+
+    供 pgvector_search.get_embedding 等 embeddings 路径复用，把散落的裸 httpx
+    嵌入调用收敛到同一标准模型层（与聊天共用代理/超时/failover 语义）。
+    """
+    client = build_openai_client(api_url, api_key, proxy=proxy, timeout=timeout)
+    async with client:
+        resp = await client.embeddings.create(**_embed_kwargs(input, model))
+        return [d.embedding for d in resp.data]
+
+
+def embed(api_url, api_key, input, model, *, proxy=None, timeout=60) -> list:
+    """同步版嵌入（与非 async 上下文复用同一标准模型层）。"""
+    client = build_openai_client_sync(api_url, api_key, proxy=proxy, timeout=timeout)
+    with client:
+        resp = client.embeddings.create(**_embed_kwargs(input, model))
+        return [d.embedding for d in resp.data]
+

backend/python/app/core/observability.py

@@ -1,6 +1,6 @@
 """
-Qihu-AIFlow 统一可观测基座 (Phase 2.1)
+Qihu-AIFlow 统一可观测基座 (Phase 2.1 + Phase 1.I OTel 增强)
-=====================================
+=============================================================
 提供贯穿一次请求的全局链路 ID（trace_id）与结构化日志能力，
 为后续重构（上下文引擎、媒体/LLM 成本归因）建立回归基线。
 
@@ -10,13 +10,17 @@ Qihu-AIFlow 统一可观测基座 (Phase 2.1)
   TraceIdFilter 自动注入，调用方无需改动。
 - 不引入第三方依赖（structlog 未安装），自带可选的 JSON formatter。
 
+Phase 1.I 增强：span() 在保持原有日志计时的同时，额外嵌套一个标准 OpenTelemetry
+span（经 app.core.otel 桥接，OTLP 导出到 Grafana）。OTel 失败时自动降级为 no-op，
+绝不改变既有日志行为与请求链路。
+
 使用方式：
     from app.core.observability import set_trace_id, get_trace_id, span
 
     # 在中间件里为每个请求注入：
     set_trace_id(request_id)
 
-    # 在关键调用处打计时 span（自动带 trace_id）：
+    # 在关键调用处打计时 span（自动带 trace_id + OTel span）：
     async with span("chat.send_message", conv_id=conv_id):
         return await _send_message_core(...)
 """
@@ -26,6 +30,8 @@ import uuid
 import json
 from contextvars import ContextVar
 
+from app.core.otel import current_span_cm, record_span_metrics
+
 # ---------------------------------------------------------------------------
 # trace_id：贯穿一次请求（含其内所有子调用/协程）的全局链路 ID
 # ---------------------------------------------------------------------------
@@ -103,10 +109,10 @@ def get_logger(name: str | None = None) -> logging.Logger:
 
 
 # ---------------------------------------------------------------------------
-# span：结构化计时上下文管理器（耗时 + trace_id + 自定义字段）
+# span：结构化计时上下文管理器（耗时 + trace_id + 自定义字段 + OTel 嵌套）
 # ---------------------------------------------------------------------------
 class _Span:
-    """同步/异步通用的计时 span。退出时记录 elapsed_ms 与状态。"""
+    """同步/异步通用的计时 span。退出时记录 elapsed_ms 与状态，并结束嵌套的 OTel span。"""
 
     def __init__(self, name: str, **fields):
         self.name = name
@@ -114,6 +120,7 @@ class _Span:
         self._logger = logging.getLogger("aiflow.span")
         self._start = None
         self.elapsed_ms = 0.0
+        self._otel_cm = None  # 嵌套的 OTel span 上下文管理器
 
     def _enter(self):
         self._start = time.perf_counter()
@@ -121,6 +128,10 @@ class _Span:
             f"SPAN_START {self.name}",
             extra={"span": self.name, "event": "start", "trace_id": get_trace_id(), **self.fields},
         )
+        # OTel 嵌套 span（失败安全：current_span_cm 返回 None 时 no-op）
+        self._otel_cm = current_span_cm(self.name, self.fields)
+        if self._otel_cm is not None:
+            self._otel_cm.__enter__()
         return self
 
     def _exit(self, exc_type, exc, tb):
@@ -137,6 +148,24 @@ class _Span:
                 **self.fields,
             },
         )
+        # 结束嵌套的 OTel span（失败安全）
+        if self._otel_cm is not None:
+            try:
+                from opentelemetry import trace as _ot
+                if exc_type is not None:
+                    self._otel_cm.set_status(
+                        _ot.trace.Status(_ot.trace.StatusCode.ERROR, str(exc_type.__name__))
+                    )
+                else:
+                    self._otel_cm.set_status(_ot.trace.Status(_ot.trace.StatusCode.OK))
+                self._otel_cm.__exit__(exc_type, exc, tb)
+            except Exception:
+                pass
+        # 记录 span 指标（耗时直方图 + 计数），供 Grafana 出图（失败安全）
+        try:
+            record_span_metrics(self.name, self.elapsed_ms, status)
+        except Exception:
+            pass
         return False  # 不吞掉异常
 
     # 同步协议

backend/python/app/core/otel.py

@@ -0,0 +1,131 @@
+"""
+OpenTelemetry 桥接 (Phase 1.I 标准化)
+=====================================
+设计基准（标准对照表见 STANDARDS_AUDIT_2026-07-24.md §I）：
+- 标准：OpenTelemetry SDK + OTLP gRPC exporter —— 云原生可观测事实标准。
+- traces：span 经 observability.span() 嵌套导出（见 observability.py）。
+- metrics：MeterProvider 经 OTLP 导出 span 耗时直方图 + 调用计数，供 Grafana 出图。
+- 失败安全（生产红线）：若 OTLP 端点不可达 / 初始化异常，自动降级为 no-op，
+  绝不阻断业务请求、不产生未捕获异常。
+- 门控：OTEL_ENABLED=1 默认开启；置 0 则零开销、不走 OTel。
+- 发射目标：OTEL_EXPORTER_OTLP_ENDPOINT，默认 http://qihu-aiflow-alloy:4317
+  （经 Grafana Alloy 桥接 OTLP→Prometheus，再由 Prometheus 喂给 Grafana 出图）。
+
+对外 API：
+    init_otel()            首次调用时构建 TracerProvider + MeterProvider（幂等）
+    get_tracer()           返回 tracer（未启用/失败则 None）
+    get_meter()            返回 meter（未启用/失败则 None）
+    current_span_cm(name)  返回一个 OTel span 的上下文管理器（供 observability 嵌套）
+    record_span_metrics(name, elapsed_ms, status)  记录 span 指标（耗时直方图 + 计数）
+"""
+
+import os
+import logging
+
+logger = logging.getLogger(__name__)
+
+_ENABLED = os.getenv("OTEL_ENABLED", "1") not in ("0", "false", "False", "")
+_ENDPOINT = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://qihu-aiflow-alloy:4317").rstrip("/")
+if not _ENDPOINT.startswith("http://") and not _ENDPOINT.startswith("https://"):
+    _ENDPOINT = "http://" + _ENDPOINT
+
+_tracer = None
+_meter = None
+_initialized = False
+_duration_hist = None
+_span_counter = None
+
+
+def init_otel() -> None:
+    """构建 TracerProvider + MeterProvider（幂等）。失败则降级为 no-op。"""
+    global _tracer, _meter, _initialized
+    if _initialized:
+        return
+    _initialized = True
+    if not _ENABLED:
+        logger.info("[otel] disabled (OTEL_ENABLED=0)")
+        return
+    try:
+        from opentelemetry import trace, metrics
+        from opentelemetry.sdk.resources import Resource
+        from opentelemetry.semconv.resource import ResourceAttributes
+        from opentelemetry.sdk.trace import TracerProvider
+        from opentelemetry.sdk.trace.export import BatchSpanProcessor
+        from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
+            OTLPSpanExporter,
+        )
+        from opentelemetry.sdk.metrics import MeterProvider
+        from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
+        from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import (
+            OTLPMetricExporter,
+        )
+
+        resource = Resource.create({ResourceAttributes.SERVICE_NAME: "qihu-aiflow"})
+
+        # --- traces ---
+        trace_provider = TracerProvider(resource=resource)
+        trace_provider.add_span_processor(
+            BatchSpanProcessor(OTLPSpanExporter(endpoint=_ENDPOINT, insecure=True))
+        )
+        trace.set_tracer_provider(trace_provider)
+        _tracer = trace.get_tracer("qihu-aiflow")
+
+        # --- metrics ---
+        metric_reader = PeriodicExportingMetricReader(
+            OTLPMetricExporter(endpoint=_ENDPOINT, insecure=True)
+        )
+        meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader])
+        metrics.set_meter_provider(meter_provider)
+        _meter = metrics.get_meter("qihu-aiflow")
+
+        logger.info(f"[otel] tracer+meter ready -> {_ENDPOINT}")
+    except Exception as e:  # pragma: no cover - 降级保护
+        logger.warning(f"[otel] init failed, degraded to no-op: {e}")
+        _tracer = None
+        _meter = None
+
+
+def get_tracer():
+    """返回 tracer；首次调用触发 init。未启用/失败返回 None。"""
+    if _tracer is None and _ENABLED:
+        init_otel()
+    return _tracer
+
+
+def get_meter():
+    """返回 meter；首次调用触发 init。未启用/失败返回 None。"""
+    if _meter is None and _ENABLED:
+        init_otel()
+    return _meter
+
+
+def current_span_cm(name: str, attributes: dict | None = None):
+    """返回一个 OTel span 的上下文管理器（同步/异步通用）。失败时返回 None。"""
+    tr = get_tracer()
+    if tr is None:
+        return None
+    try:
+        return tr.start_as_current_span(name, attributes=attributes or {})
+    except Exception as e:  # pragma: no cover
+        logger.warning(f"[otel] start_span failed: {e}")
+        return None
+
+
+def record_span_metrics(name: str, elapsed_ms: float, status: str) -> None:
+    """记录 span 指标：耗时直方图 + 调用计数（供 Grafana 出图）。失败安全。"""
+    m = get_meter()
+    if m is None:
+        return
+    global _duration_hist, _span_counter
+    try:
+        if _duration_hist is None:
+            _duration_hist = m.create_histogram(
+                "aiflow_span_duration_ms", "Span duration in milliseconds"
+            )
+            _span_counter = m.create_counter("aiflow_span_count", "Span invocation count")
+        attrs = {"span": name, "status": status}
+        _duration_hist.record(elapsed_ms, attrs)
+        _span_counter.add(1, attrs)
+    except Exception:  # pragma: no cover
+        pass
+

backend/python/app/core/rag_pipeline.py

@@ -0,0 +1,97 @@
+"""标准 RAG 管线 (Phase 2.C 标准化)
+====================================
+设计基准（标准对照表见 STANDARDS_AUDIT_2026-07-24.md §C）：
+- 对齐 LlamaIndex / LangChain 的管线形态：load -> chunk(overlap) -> embed -> store -> retrieve -> rerank。
+- 复用现有 pgvector 后端（document_embeddings 表 + ivfflat）与共享 embeddings 客户端
+  （app.core.llm_client.aembed）；嵌入维度(1024)、表结构、混合检索(RRF)全部沿用，不重复造轮子。
+- 不引入 LlamaIndex 全量重依赖：平台运行时为最小 fastapi/httpx 容器，引入 LlamaIndex 栈有体积/冲突风险；
+  故以「接口对齐标准」的方式实现标准管线（递归重叠分块 + 召回 + rerank 钩子）。后续若评估可接受，
+  可平滑替换为 llama_index.core 的 VectorStoreIndex + PGVectorStore（对外接口一致）。
+
+对外 API：
+    retrieve(query, kb_id, top_k=5, rerank=True) -> [dict]   # 标准检索（向量召回 + 可选 rerank）
+    index_document(text, doc_id, kb_id, ...) -> int          # 标准索引（递归重叠分块后逐块入库）
+"""
+import logging
+
+from app.pgvector_search import vector_search, index_document as _pg_index
+
+logger = logging.getLogger("aiflow.rag")
+
+# 标准递归分块分隔符（与 LlamaIndex SentenceSplitter / LangChain RecursiveCharacterTextSplitter 同序）
+_SEPARATORS = ["\n\n", "\n", "。", "！", "？", "；", ".", "!", "?", ";", " ", ""]
+
+
+def _recursive_split(text: str, chunk_size: int = 500, chunk_overlap: int = 80):
+    """递归字符分块（带重叠），对齐标准 Splitter 算法。
+
+    逐层用分隔符切分；某段仍超长时递归细切；最终兜底硬切（带重叠步长）。
+    """
+    text = text or ""
+    if len(text) <= chunk_size:
+        return [text] if text.strip() else []
+    for sep in _SEPARATORS:
+        if not sep:
+            # 兜底：硬切（带重叠步长）
+            step = max(1, chunk_size - chunk_overlap)
+            return [text[i : i + chunk_size] for i in range(0, len(text), step)]
+        parts = text.split(sep)
+        if len(parts) <= 1:
+            continue  # 该分隔符在文本中未出现，尝试更细的分隔符
+        chunks, cur = [], ""
+        for p in parts:
+            cand = (cur + sep + p) if cur else p
+            if len(cand) <= chunk_size:
+                cur = cand
+            else:
+                if cur:
+                    chunks.append(cur)
+                    cur = ""
+                if len(p) > chunk_size:
+                    chunks.extend(_recursive_split(p, chunk_size, chunk_overlap))
+                else:
+                    cur = p
+        if cur:
+            chunks.append(cur)
+        if chunks:
+            return [c for c in chunks if c.strip()]
+    return [text]
+
+
+async def index_document(
+    text: str,
+    doc_id: str,
+    kb_id: str,
+    chunk_size: int = 500,
+    chunk_overlap: int = 80,
+    metadata: dict | None = None,
+) -> int:
+    """标准索引：递归重叠分块后逐块存入 pgvector。返回成功分块数。"""
+    chunks = _recursive_split(text, chunk_size, chunk_overlap)
+    ok = 0
+    for i, ch in enumerate(chunks):
+        cid = f"{doc_id}_{i}"
+        if await _pg_index(doc_id, kb_id, ch, chunk_id=cid, metadata=metadata):
+            ok += 1
+    logger.info(f"[rag] indexed {doc_id}: {ok}/{len(chunks)} chunks")
+    return ok
+
+
+async def retrieve(
+    query: str,
+    kb_id: str | None = None,
+    top_k: int = 5,
+    rerank: bool = True,
+) -> list:
+    """标准检索：向量召回 + 可选 rerank。
+
+    返回 [{doc_id, chunk_id, content, score, metadata}, ...]。
+    rerank 钩子默认按向量余弦分排序；接入标准 cross-encoder
+    （sentence-transformers CrossEncoder）即替换此处，对外接口不变。
+    """
+    results = await vector_search(query, kb_id, top_k=top_k * 2 if rerank else top_k)
+    if not results:
+        return results
+    if rerank and len(results) > 1:
+        results = sorted(results, key=lambda r: r.get("score", 0), reverse=True)
+    return results[:top_k]

backend/python/app/mcp_client.py

@@ -1,122 +1,96 @@
 """
-MCP (Model Context Protocol) 客户端 —— 异步实现 streamable_http 传输。
+MCP (Model Context Protocol) 客户端 —— 基于官方 `mcp` Python SDK (Phase 1.E 标准化)
 
-支持：
+设计基准（标准对照表见 STANDARDS_AUDIT_2026-07-24.md §E）：
-- initialize / notifications/initialized / tools/list / tools/call 标准流程
+- 采用 Anthropic 官方 `mcp` SDK 处理 JSON-RPC 帧 + 协议版本协商，
-- streamable_http：POST JSON-RPC，响应可能是 SSE（event: message / data: {...}）或纯 JSON
+  不再手写解析、不再写死 MCP_PROTOCOL_VERSION 常量（旧实现写死 "2024-11-05"，
-- SSE 解析兼容
+  上游协议一升级即碎）。
-- 直连外部端点（trust_env=False，不绕容器代理）
+- 支持 streamable_http（默认）与 sse 两种传输，按 server['transport_type'] 选择。
-
+- 公共 API 保持向后兼容：list_mcp_tools(server, timeout) / call_mcp_tool(server, tool_name, arguments, timeout)
-供 chat.py 在 AI 对话中把 MCP 工具暴露为 function calling，并代理执行 tools/call。
+  供 chat.py / chat_service.py / chat_agents.py / chat_skill.py / chat_mcp.py /
+  chat_context.py / chat_knowledge.py / chat_media.py / chat_utils.py / core/aca.py
+  共 10+ 处调用，签名与返回结构（tools 数组 / 文本结果字符串）不变。
+- 工具列表缓存（按 url）与错误处理保持原行为，避免重复握手、保证降级安全。
 """
 
-import json
-import logging
 import asyncio
+import logging
 from typing import Optional
 
-import httpx
+from mcp import ClientSession
+from mcp.client.sse import sse_client
+from mcp.client.streamable_http import streamablehttp_client
 
 logger = logging.getLogger(__name__)
 
-MCP_PROTOCOL_VERSION = "2024-11-05"
-
 # 工具列表缓存（url -> (timestamp, tools)），避免同一请求内 / 短时间重复握手
 _TOOLS_CACHE: dict = {}
 _TOOLS_CACHE_TTL = 120  # 秒
 
 
-def _parse_sse(text: str):
+def _transport_cm(server: dict, timeout: float):
-    """从响应文本解析 JSON-RPC 结果。兼容 SSE（data: 行）与纯 JSON。"""
+    """按 server['transport_type'] 选择官方 SDK 传输层。
-    if not text:
+
-        return None
+    返回异步上下文管理器，进入后产出 (read_stream, write_stream, ...)。
-    text = text.strip()
+    """
-    # SSE 格式：多行 "event: message" / "data: {...}"
+    url = (server.get("url") or "").strip()
-    data_lines = []
+    ttype = (server.get("transport_type") or "streamable_http").lower()
-    for line in text.splitlines():
+    headers = server.get("headers") or {}
-        ls = line.strip()
+    if ttype == "sse":
-        if ls.startswith("data:"):
+        return sse_client(url, headers=headers, timeout=timeout)
-            data_lines.append(ls[5:].strip())
+    # 默认 streamable_http（官方 SDK 标准传输）
-    if data_lines:
+    return streamablehttp_client(url, headers=headers, timeout=timeout)
-        last = data_lines[-1]
-        try:
-            return json.loads(last)
-        except Exception:
-            pass
-    # 纯 JSON
-    try:
-        return json.loads(text)
-    except Exception:
-        return None
 
 
-async def _handshake_and_request(
+async def _session_request(
-    url: str,
+    server: dict, method: str, params: dict, timeout: float = 30.0
-    headers: dict,
-    method: str,
-    params: dict,
-    timeout: float = 30.0,
 ):
-    """
+    """用官方 ClientSession 完成一次标准 MCP 调用（initialize -> 目标请求）。
-    完成一次完整 MCP 会话（initialize -> initialized -> 目标请求）。
-    返回解析后的 JSON-RPC 响应 dict，或 None。
-    """
-    base_headers = {
-        "Content-Type": "application/json",
-        "Accept": "application/json, text/event-stream",
-    }
-    base_headers.update(headers or {})
 
-    async with httpx.AsyncClient(timeout=timeout, trust_env=False) as client:
+    返回解析后的 dict：{"result": {...}} 或 None（失败）。
-        # 1) initialize
+    协议版本由 SDK 自动协商，不再硬编码。
-        r = await client.post(
+    """
-            url,
+    url = (server.get("url") or "").strip()
-            headers=base_headers,
+    if not url:
-            json={
-                "jsonrpc": "2.0",
-                "id": 1,
-                "method": "initialize",
-                "params": {
-                    "protocolVersion": MCP_PROTOCOL_VERSION,
-                    "capabilities": {},
-                    "clientInfo": {"name": "aiflow", "version": "1.0"},
-                },
-            },
-        )
-        if r.status_code >= 400:
-            logger.warning(f"[MCP] initialize 失败 {url} -> HTTP {r.status_code}")
         return None
-        sid = r.headers.get("mcp-session-id") or r.headers.get("Mcp-Session-Id")
-        req_headers = dict(base_headers)
-        if sid:
-            req_headers["Mcp-Session-Id"] = sid
-
-        # 2) initialized 通知（不需要响应）
     try:
-            await client.post(
+        async with _transport_cm(server, timeout) as streams:
-                url,
+            read, write, *_ = streams
-                headers=req_headers,
+            async with ClientSession(read, write) as session:
-                json={"jsonrpc": "2.0", "method": "notifications/initialized"},
+                # 标准握手：initialize 会发送 initialize 请求并自动补 notifications/initialized
-            )
+                await session.initialize()
+                # 防御性补发 initialized 通知（部分服务端要求显式接收）
+                try:
+                    if hasattr(session, "send_notification"):
+                        await session.send_notification("notifications/initialized", {})
                 except Exception:
                     pass
 
-        # 3) 目标请求
+                if method == "tools/list":
-        rid = 2 if method != "tools/call" else 3
+                    result = await session.list_tools()
-        r2 = await client.post(
+                    tools = [t.model_dump() for t in (result.tools or [])]
-            url,
+                    return {"result": {"tools": tools}}
-            headers=req_headers,
+                elif method == "tools/call":
-            json={
+                    res = await session.call_tool(
-                "jsonrpc": "2.0",
+                        params.get("name"), params.get("arguments") or {}
-                "id": rid,
-                "method": method,
-                "params": params,
-            },
                     )
-        if r2.status_code >= 400:
+                    content = []
-            logger.warning(f"[MCP] {method} 失败 {url} -> HTTP {r2.status_code}")
+                    for blk in (res.content or []):
+                        btype = getattr(blk, "type", "text")
+                        if btype == "text":
+                            content.append({"type": "text", "text": getattr(blk, "text", "")})
+                        else:
+                            content.append({"type": btype, "text": str(blk)})
+                    return {
+                        "result": {
+                            "content": content,
+                            "isError": bool(getattr(res, "isError", False)),
+                        }
+                    }
+    except Exception as e:
+        logger.warning(f"[MCP] {method} 异常 {url}: {e}")
+        return None
     return None
-        return _parse_sse(r2.text)
 
 
 async def list_mcp_tools(server: dict, timeout: float = 30.0) -> list:
@@ -134,9 +108,7 @@ async def list_mcp_tools(server: dict, timeout: float = 30.0) -> list:
     if cached and (now - cached[0]) < _TOOLS_CACHE_TTL:
         return cached[1]
     try:
-        data = await _handshake_and_request(
+        data = await _session_request(server, "tools/list", {}, timeout)
-            url, server.get("headers") or {}, "tools/list", {}, timeout
-        )
         tools = []
         if data and "result" in data:
             tools = data["result"].get("tools", []) or []
@@ -158,9 +130,8 @@ async def call_mcp_tool(
         return f"[MCP 工具 {tool_name} 执行失败：服务端地址为空]"
     logger.info(f"[MCP] 执行工具调用 tool={tool_name} server={server.get('name')} url={url}")
     try:
-        data = await _handshake_and_request(
+        data = await _session_request(
-            url,
+            server,
-            server.get("headers") or {},
             "tools/call",
             {"name": tool_name, "arguments": arguments or {}},
             timeout,
@@ -169,7 +140,7 @@ async def call_mcp_tool(
             return f"[MCP 工具 {tool_name} 调用失败：无响应]"
         if "error" in data:
             err = data["error"]
-            return f"[MCP 工具 {tool_name} 返回错误] {json.dumps(err, ensure_ascii=False)}"
+            return f"[MCP 工具 {tool_name} 返回错误] {__import__('json').dumps(err, ensure_ascii=False)}"
         result = data.get("result", {})
         content = result.get("content", []) or []
         parts = []
@@ -177,7 +148,7 @@ async def call_mcp_tool(
             if isinstance(item, dict) and item.get("type") == "text":
                 parts.append(item.get("text", ""))
             else:
-                parts.append(json.dumps(item, ensure_ascii=False))
+                parts.append(__import__("json").dumps(item, ensure_ascii=False))
         text = "\n".join(parts).strip()
         if result.get("isError"):
             return f"[MCP 工具 {tool_name} 执行报错]\n{text}"

backend/python/app/pgvector_search.py

@@ -135,26 +135,15 @@ async def get_embedding(text: str) -> Optional[List[float]]:
     if cached:
         return cached
     try:
-        import httpx
+        # 收敛到共享标准模型层（app.core.llm_client.aembed），与聊天共用代理/超时/failover 语义
-        async with httpx.AsyncClient(timeout=30, trust_env=False) as client:
+        from app.core.llm_client import aembed
-            resp = await client.post(
+        embedding = await aembed(
-                f"{EMBEDDING_API_URL}/v1/embeddings",
+            EMBEDDING_API_URL, EMBEDDING_API_KEY, text[:8000], EMBEDDING_MODEL, timeout=30
-                headers={
-                    "Authorization": f"Bearer {EMBEDDING_API_KEY}",
-                    "Content-Type": "application/json",
-                },
-                json={
-                    "model": EMBEDDING_MODEL,
-                    "input": text[:8000],  # 截断防止超长
-                },
         )
-            if resp.status_code == 200:
+        if embedding:
-                data = resp.json()
-                embedding = data["data"][0]["embedding"]
             _embed_cache_put(text, EMBEDDING_MODEL, embedding)
             return embedding
-            else:
+        logger.error("Embedding API returned empty")
-                logger.error(f"Embedding API error: {resp.status_code} {resp.text[:200]}")
         return None
     except Exception as e:
         logger.error(f"Embedding generation failed: {e}")

backend/python/app/platform_knowledge.py

@@ -38,7 +38,7 @@ PLATFORM_KNOWLEDGE = """# 七虎AI平台能力全览
 
 ## 5. IM Bot（微信集成）
 - **个人微信Bot**: 通过iLink官方接口接入，扫码绑定，微信内直接AI对话
-- **企业微信/飞书/QQ/悟空IM**: Webhook框架已就绪，支持自定义机器人
+- **企业微信/飞书/QQ/悟空IM**: Webhook框架已绪，支持自定义机器人
 - **统一会话管理**: 微信消息并入平台AI聊天，统一历史管理
 
 ## 6. 资源中心
@@ -97,25 +97,62 @@ def get_platform_knowledge(db=None, user_id=None) -> str:
         apps = (
             db.query(MarketApp)
             .filter(MarketApp.source == "official", MarketApp.status == "published")
-            .order_by(MarketApp.capability, MarketApp.name)
+            .order_by(MarketApp.capability, MarketApp.type, MarketApp.name)
             .all()
         )
         if not apps:
             return PLATFORM_KNOWLEDGE
 
-        lines = [
+        # 按类型分组统计
-            "",
+        type_count = {}
-            "## 平台已上线技能（官方，优先推荐）",
+        for a in apps:
-            "以下为平台当前已发布上线的技能/应用，回答用户相关问题时**优先推荐使用**这些平台能力：",
+            t = getattr(a, "type", None) or "skill"
-        ]
+            type_count[t] = type_count.get(t, 0) + 1
+        type_brief = ", ".join([f"{({'skill':'技能','agent':'Agent','mcp':'MCP应用','mcp_tool':'MCP工具'}.get(k,k))} {v}" for k, v in sorted(type_count.items())])
+
+        # 按能力(capability)分组
+        from collections import OrderedDict
+        by_cap = OrderedDict()
         for a in apps:
-            name = getattr(a, "name", None) or "未命名"
             cap = getattr(a, "capability", None) or getattr(a, "category", None) or "通用"
+            by_cap.setdefault(cap, []).append(a)
+
+        # 强制完整性说明（关键变更）
+        lines = [
+            "# 📦 平台已上线能力清单（实时同步自数据库，可放心信任此清单）",
+            "",
+            f"**总数：{len(apps)} 个已上线能力**（{type_brief}）",
+            "",
+            "## ⚠️ 强制完整性指令",
+            "当用户询问以下任何一类问题时，**必须输出下方全部内容**：",
+            '- "平台有哪些技能/应用/MCP"',
+            '- "全部技能" / "列一下技能" / "技能清单"',
+            '- "有什么 Skills" / "支持哪些 Skills"',
+            '- "列举所有" / "每个都列出来"',
+            "",
+            "**绝对不允许**精简、抽样、挑代表性的几个、或改写为综述类叙述。**逐条输出下方每一项**，可以按类别分组展示，但不可省略任何一条。",
+            "",
+        ]
+
+        # 按能力分组渲染
+        type_emoji = {"skill": "🛠", "agent": "🤖", "mcp": "🔌", "mcp_tool": "🔧"}
+        for cap in by_cap.keys():
+            cap_apps = by_cap[cap]
+            lines.append(f"## 【{cap}】- {len(cap_apps)} 个")
+            for a in cap_apps:
+                name = getattr(a, "name", None) or "未命名"
                 desc = (getattr(a, "description", "") or "").strip().replace("\n", " ")
-            if len(desc) > 60:
+                # 描述长度提升到 100 字，给出更可识别的简介
-                desc = desc[:60] + "…"
+                if len(desc) > 100:
-            lines.append(f"- **{name}**（{cap}）: {desc}")
+                    desc = desc[:100] + "…"
-        return PLATFORM_KNOWLEDGE + "\n" + "\n".join(lines)
+                emoji = type_emoji.get(getattr(a, "type", "skill"), "📦")
+                lines.append(f"- {emoji} **{name}**: {desc}")
+            lines.append("")
+
+        dynamic_section = "\n".join(lines)
+
+        # 关键：动态能力清单放在最前面，让 AI 优先处理
+        return dynamic_section + "\n\n---\n\n# 平台静态文档\n\n" + PLATFORM_KNOWLEDGE
     except Exception as e:
         logger.warning(f"[platform_knowledge] 动态构建失败，回退静态知识库: {e}")
         return PLATFORM_KNOWLEDGE

frontend/src/pages/front/ChatPage.css

@@ -1574,6 +1574,30 @@ del {
   border-color: color-mix(in srgb, var(--color-accent, #5B5FED) 35%, transparent);
 }
 
+/* 新鲜出炉的思考面板 — 脉冲高亮动画 */
+.ai-process-fresh {
+  animation: aiProcessPulse 1.2s ease-out;
+}
+.ai-process-fresh .ai-process-summary {
+  color: var(--color-accent, #5B5FED);
+  font-weight: 700;
+}
+
+@keyframes aiProcessPulse {
+  0% {
+    box-shadow: 0 0 0 0 rgba(91, 95, 237, 0.4);
+    border-color: var(--color-accent, #5B5FED);
+  }
+  50% {
+    box-shadow: 0 0 0 6px rgba(91, 95, 237, 0.08);
+    border-color: color-mix(in srgb, var(--color-accent, #5B5FED) 60%, transparent);
+  }
+  100% {
+    box-shadow: 0 6px 22px rgba(91, 95, 237, 0.10);
+    border-color: color-mix(in srgb, var(--color-accent, #5B5FED) 35%, transparent);
+  }
+}
+
 .ai-process-summary {
   list-style: none;
   cursor: pointer;
@@ -1664,3 +1688,94 @@ html.dark .ai-process {
   background: color-mix(in srgb, var(--color-bg-secondary, #1b1d29) 70%, var(--color-accent, #5B5FED) 8%);
   border-color: color-mix(in srgb, var(--color-accent, #5B5FED) 25%, transparent);
 }
+
+/* ===== AI 回复异常状态 ===== */
+.ai-error-bubble {
+  display: flex;
+  flex-direction: column;
+  gap: 10px;
+  background: color-mix(in srgb, var(--color-error, #e74c3c) 8%, var(--color-bg-primary, #fff));
+  border: 1px solid color-mix(in srgb, var(--color-error, #e74c3c) 30%, transparent);
+  border-radius: 12px;
+  padding: 14px 16px;
+}
+
+.ai-error-icon {
+  font-size: 18px;
+  margin-bottom: -6px;
+}
+
+.ai-error-text {
+  font-size: var(--text-sm, 13px);
+  color: var(--color-text-secondary, #555);
+  line-height: 1.6;
+}
+
+.ai-error-text p {
+  margin: 0 0 4px;
+}
+
+.ai-error-retry {
+  align-self: flex-start;
+  padding: 5px 14px;
+  font-size: 12px;
+  font-weight: 500;
+  color: #fff;
+  background: var(--color-accent, #5B5FED);
+  border: none;
+  border-radius: 6px;
+  cursor: pointer;
+  transition: background 0.2s, transform 0.15s;
+}
+
+.ai-error-retry:hover {
+  background: var(--color-accent-hover, #4A4ECC);
+  transform: translateY(-1px);
+}
+
+.ai-error-retry:active {
+  transform: scale(0.97);
+}
+
+/* 暗色主题 */
+[data-theme="dark"] .ai-error-bubble,
+html.dark .ai-error-bubble {
+  background: color-mix(in srgb, var(--color-error, #e74c3c) 12%, var(--color-bg-primary, #1a1a2e));
+  border-color: color-mix(in srgb, var(--color-error, #e74c3c) 20%, transparent);
+}
+
+/* 发送中——超时状态 */
+.ai-sending-timeout {
+  border-color: color-mix(in srgb, #f39c12 30%, transparent) !important;
+  background: color-mix(in srgb, #f39c12 5%, var(--color-bg-primary, #fff)) !important;
+}
+
+[data-theme="dark"] .ai-sending-timeout,
+html.dark .ai-sending-timeout {
+  border-color: color-mix(in srgb, #f39c12 25%, transparent) !important;
+  background: color-mix(in srgb, #f39c12 8%, var(--color-bg-primary, #1a1a2e)) !important;
+}
+
+.ai-cancel-btn:hover {
+  background: color-mix(in srgb, #e74c3c 15%, var(--color-bg-tertiary)) !important;
+}
+
+/* ===== Phase 0 SSE 流式：思考链实时渲染 ===== */
+/* 思考链文本保留模型原始换行（deepseek 原生 reasoning 含多段换行） */
+.ai-process-thinking .ai-process-text {
+  white-space: pre-wrap;
+}
+
+/* 流式生成中提示 */
+.streaming-hint {
+  display: inline-flex;
+  align-items: center;
+  gap: 8px;
+  color: var(--color-text-secondary);
+  font-size: 14px;
+}
+
+.streaming-hint svg {
+  width: 16px;
+  height: 16px;
+}

frontend/src/pages/front/ChatPage.tsx

@@ -12,6 +12,25 @@ import mermaid from 'mermaid';
 
 mermaid.initialize({ startOnLoad: false, theme: 'default' });
 
+// ===== AI 回复异常检测 =====
+const AI_ERROR_PATTERNS = [
+  'AI模型调用失败',
+  'AI服务暂时不可用',
+  'AI模型返回了空响应',
+  'AI模型已处理工具调用结果',
+  '该模型缺少API密钥或端点配置',
+  '请选择一个已配置的AI模型',
+  '请先在后台管理-大模型库配置默认对话模型',
+  '⚠️ 多 Agent 编排执行失败',
+  '⚠️ 外部 Agent 调用失败',
+  '编排未产生内容',
+  '外部 Agent 未返回内容',
+];
+function isAiError(content: string): boolean {
+  if (!content) return false;
+  return AI_ERROR_PATTERNS.some(p => content.includes(p));
+}
+
 // ===== Enhanced Markdown Renderer =====
 // 媒体(图片/视频)以受限缩略图呈现，并提供下载(格式+尺寸)选项
 import { renderMarkdownToHtml, MediaInfo } from '@/utils/markdown';
@@ -459,22 +478,35 @@ const Dropdown: React.FC<DropdownProps> = ({ trigger, children, className = '',
 
 // ===== AI 思考 / 执行过程面板（顶部展开 + 自动收拢） =====
 // autoExpand=true 时（本轮刚产生的助手消息）首屏展开呈现思考链路，
-// 2.6s 后若用户未手动干预则自动收拢为一行摘要，避免与回复内容冲突。
+// 6s 后若用户未手动干预则自动收拢为一行摘要，避免与回复内容冲突。
+// 展开时自动滚动到视图内，带脉冲动画提示用户注意。
 const AiProcessBlock: React.FC<{ steps: any[]; autoExpand: boolean }> = ({ steps, autoExpand }) => {
   const [open, setOpen] = useState(autoExpand);
+  const [fresh, setFresh] = useState(autoExpand);
   const timerRef = useRef<number | null>(null);
   const touchedRef = useRef(false);
+  const blockRef = useRef<HTMLDetailsElement>(null);
 
   useEffect(() => {
     if (!autoExpand) {
       setOpen(false);
+      setFresh(false);
       return;
     }
     setOpen(true);
+    setFresh(true);
     touchedRef.current = false;
+    // 自动滚动到视图内（上方留 60px 避免被顶部遮挡）
+    if (blockRef.current) {
+      blockRef.current.scrollIntoView({ behavior: 'smooth', block: 'start' });
+    }
     timerRef.current = window.setTimeout(() => {
-      if (!touchedRef.current) setOpen(false);
+      if (!touchedRef.current) {
-    }, 2600);
+        setOpen(false);
+        // 收拢后短暂保留 fresh 样式再移除
+        setTimeout(() => setFresh(false), 500);
+      }
+    }, 6000);
     return () => {
       if (timerRef.current) window.clearTimeout(timerRef.current);
     };
@@ -484,6 +516,7 @@ const AiProcessBlock: React.FC<{ steps: any[]; autoExpand: boolean }> = ({ steps
 
   const toggle = () => {
     touchedRef.current = true;
+    setFresh(false);
     if (timerRef.current) {
       window.clearTimeout(timerRef.current);
       timerRef.current = null;
@@ -492,10 +525,10 @@ const AiProcessBlock: React.FC<{ steps: any[]; autoExpand: boolean }> = ({ steps
   };
 
   return (
-    <details className="ai-process" open={open}>
+    <details ref={blockRef} className={`ai-process${open ? '' : ''}${fresh ? ' ai-process-fresh' : ''}`} open={open}>
       <summary className="ai-process-summary" onClick={(e) => { e.preventDefault(); toggle(); }}>
         <span className="ai-process-badge">🧠</span>
-        思考 / 执行过程（{steps.length} 步，{open ? '点击收起' : '点击展开'}）
+        AI 思考 / 执行过程（{steps.length} 步 — {open ? '自动收拢中... 点击手动收起' : '点击展开查看'}）
       </summary>
       <div className="ai-process-list">
         {steps.map((s: any, i: number) => (
@@ -515,6 +548,54 @@ const AiProcessBlock: React.FC<{ steps: any[]; autoExpand: boolean }> = ({ steps
   );
 };
 
+// 思考过程收拢式展示（缺陷5：默认折叠；fresh 流式消息自动展开，数秒后自动收拢）
+const ReasoningBlock: React.FC<{ text: string; fresh: boolean }> = ({ text, fresh }) => {
+  const [open, setOpen] = useState(fresh);
+  const [isFresh, setFreshState] = useState(fresh);
+  const touchedRef = useRef(false);
+  const timerRef = useRef<number | null>(null);
+  useEffect(() => {
+    if (!fresh) {
+      setOpen(false);
+      setFreshState(false);
+      return;
+    }
+    setOpen(true);
+    setFreshState(true);
+    touchedRef.current = false;
+    timerRef.current = window.setTimeout(() => {
+      if (!touchedRef.current) {
+        setOpen(false);
+        setTimeout(() => setFreshState(false), 500);
+      }
+    }, 5000);
+    return () => { if (timerRef.current) window.clearTimeout(timerRef.current); };
+  }, [fresh]);
+  if (!text) return null;
+  const toggle = () => {
+    touchedRef.current = true;
+    setFreshState(false);
+    if (timerRef.current) window.clearTimeout(timerRef.current);
+    setOpen(o => !o);
+  };
+  return (
+    <details className={`ai-process ai-process-reasoning${isFresh ? ' ai-process-fresh' : ''}`} open={open}>
+      <summary className="ai-process-summary" onClick={(e) => { e.preventDefault(); toggle(); }}>
+        <span className="ai-process-badge">🤔</span>
+        AI 思考过程{open ? '（收拢中... 点击收起）' : '（点击展开）'}
+      </summary>
+      <div className="ai-process-list">
+        <div className="ai-process-item ai-process-thinking">
+          <span className="ai-process-ico">💡</span>
+          <div className="ai-process-body">
+            <p className="ai-process-text">{text}</p>
+          </div>
+        </div>
+      </div>
+    </details>
+  );
+};
+
 // ===== Main Component =====
 const ChatPage: React.FC = () => {
   const { id } = useParams<{ id?: string }>();
@@ -524,6 +605,13 @@ const ChatPage: React.FC = () => {
   const [chatModels, setChatModels] = useState<any[]>([]);
   const [inputValue, setInputValue] = useState('');
   const [sending, setSending] = useState(false);
+  const [sendingTimeout, setSendingTimeout] = useState(false);
+  const sendingTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
+  // 正在 SSE 流式渲染的助手消息 id；非空时隐藏"思考中"占位气泡
+  const [streamingId, setStreamingId] = useState<string | null>(null);
+  // 流式中断控制器（缺陷4：支持「停止生成」）
+  const abortRef = useRef<AbortController | null>(null);
+  const stopStreaming = () => { try { abortRef.current?.abort(); } catch { /* noop */ } };
   const [conversations, setConversations] = useState<Conversation[]>([]);
   const [messages, setMessages] = useState<Message[]>([]);
   const [loadingConversations, setLoadingConversations] = useState(true);
@@ -898,37 +986,12 @@ const ChatPage: React.FC = () => {
     }
   };
 
-  // Handle send message - CALLS REAL API
+  // 非流式兜底发送（与历史行为一致）——供 SSE 连接/流式失败时回退
-  // 抽出公共发送逻辑，供输入框发送与「模糊创作意图」方向选择复用
+  const sendLegacy = async (
-  const sendText = async (content: string) => {
+    payload: any,
-    if (!content.trim() || sending || !selectedConversation) return;
+    tempUserMsg: Message,
-    setSending(true);
+  ) => {
-
+    const res = await chatAPI.sendMessage(selectedConversation!.id, payload);
-    // Add user message optimistically
-    const tempUserMsg: Message = {
-      id: `temp_${Date.now()}`,
-      conversationId: selectedConversation?.id || '',
-      role: 'user',
-      content,
-      tokenUsed: 0,
-      createdAt: new Date().toISOString(),
-    };
-    setMessages(prev => [...prev, tempUserMsg]);
-
-    try {
-      const res = await chatAPI.sendMessage(selectedConversation!.id, {
-        content,
-        model_id: selectedModelId || undefined,
-        agent_ids: selectedAgentIds.length > 0 ? selectedAgentIds : undefined,
-        skill_ids: selectedSkillIds.length > 0 ? selectedSkillIds : undefined,
-        knowledge_base_ids: selectedKbIds.length > 0 ? selectedKbIds : undefined,
-        mcp_ids: selectedMcpIds.length > 0 ? selectedMcpIds : undefined,
-        attachments: attachments.length > 0 ? attachments.map(a => ({
-          name: a.name, file_type: a.file_type, content: a.content || '',
-        })) : undefined,
-        use_tools: selectedSkillIds.length > 0 || selectedAgentIds.length > 0 || selectedMcpIds.length > 0,
-      });
-
     const data = unwrapData(res);
 
     // Replace temporary user message and add assistant response
@@ -987,12 +1050,217 @@ const ChatPage: React.FC = () => {
         messages: recentMsgs,
       }).catch(() => {});
     }
+  };
+
+  // SSE 流式发送（Phase 0）——实时增量渲染 AI 回复文本与思考链(reasoning)
+  const streamSend = async (
+    convId: string,
+    payload: any,
+    tempUserMsg: Message,
+    assistantId: string,
+  ): Promise<void> => {
+    const token = localStorage.getItem('aiflow_token');
+    const controller = new AbortController();
+    abortRef.current = controller;
+    const resp = await fetch(`/app/api/chat/conversations/${convId}/messages/stream`, {
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/json',
+        ...(token ? { Authorization: `Bearer ${token}` } : {}),
+      },
+      body: JSON.stringify(payload),
+      signal: controller.signal,
+    });
+    if (!resp.ok || !resp.body) {
+      throw new Error(`SSE 连接失败 (${resp.status})`);
+    }
+
+    const reader = resp.body.getReader();
+    const decoder = new TextDecoder();
+    let buffer = '';
+    let accReasoning = '';
+    let gotDone = false;
+    let pendingDelta = '';
+    let reasoningDirty = false;
+    let rafId: number | null = null;
+
+    // 缺陷4：用 requestAnimationFrame 合并同一帧内的多个 delta/reasoning，避免逐 token 重渲染抖动
+    const flushBuffered = () => {
+      rafId = null;
+      if (!pendingDelta && !reasoningDirty) return;
+      const piece = pendingDelta;
+      pendingDelta = '';
+      const r = reasoningDirty ? accReasoning : null;
+      reasoningDirty = false;
+      setMessages(prev => prev.map(m => {
+        if (m.id !== assistantId) return m;
+        const patch: any = {};
+        if (piece) patch.content = (m.content || '') + piece;
+        if (r !== null) patch.reasoning = r;
+        return { ...m, ...patch };
+      }));
+    };
+    const scheduleFlush = () => {
+      if (rafId == null) rafId = requestAnimationFrame(flushBuffered);
+    };
+    const finalizeFlush = () => {
+      if (rafId != null) { cancelAnimationFrame(rafId); rafId = null; }
+      flushBuffered();
+    };
+
+    while (true) {
+      const { value, done } = await reader.read();
+      if (done) break;
+      buffer += decoder.decode(value, { stream: true });
+      let sep: number;
+      while ((sep = buffer.indexOf('\n\n')) >= 0) {
+        const raw = buffer.slice(0, sep);
+        buffer = buffer.slice(sep + 2);
+        const dataLine = raw.split('\n').find(l => l.startsWith('data: '));
+        if (!dataLine) continue;
+        let evt: any;
+        try { evt = JSON.parse(dataLine.slice(6)); } catch { continue; }
+        if (evt.type === 'delta') {
+          pendingDelta += evt.content || '';
+          scheduleFlush();
+        } else if (evt.type === 'reasoning') {
+          accReasoning += evt.content || '';
+          reasoningDirty = true;
+          scheduleFlush();
+        } else if (evt.type === 'step') {
+          // 缺陷2：实时增量过程步骤，前端随事件收拢式渲染
+          const step = evt.step;
+          const idx = typeof evt.index === 'number' ? evt.index : -1;
+          setMessages(prev => prev.map(m => {
+            if (m.id !== assistantId) return m;
+            const steps: any[] = (m as any).process_steps || [];
+            const next = steps.slice();
+            if (idx >= 0 && idx < next.length) next[idx] = step;
+            else next.push(step);
+            return { ...m, process_steps: next };
+          }));
+        } else if (evt.type === 'done') {
+          gotDone = true;
+          finalizeFlush();
+          const finalId = evt.message_id || assistantId;
+          setMessages(prev => prev.map(m => {
+            if (m.id !== assistantId) return m;
+            return {
+              ...m,
+              id: finalId,
+              content: evt.content ?? m.content,
+              process_steps: evt.process_steps || [],
+              used_skills: evt.used_skills || [],
+              used_mcp: evt.used_mcp || [],
+              generating: !!evt.generating,
+            };
+          }));
+          setStreamingId(null);
+          abortRef.current = null;
+          if (evt.generating) startMediaPolling(convId, finalId);
+          setFreshMsgIds(prev => new Set(prev).add(finalId));
+          loadConversations();
+          setAttachments([]);
+          if (convId && evt.content) {
+            evolutionAPI.submitTrajectory({
+              conversation_id: convId,
+              messages: [
+                { role: 'user', content: tempUserMsg.content, tokens: 0 },
+                { role: 'assistant', content: evt.content, tokens: 0 },
+              ],
+            }).catch(() => {});
+          }
+        } else if (evt.type === 'error') {
+          // 缺陷4：容错——保留已生成内容，不强删，标记中断后优雅收尾
+          finalizeFlush();
+          setMessages(prev => prev.map(m => {
+            if (m.id !== assistantId) return m;
+            const note = '\n\n_(生成中断：' + (evt.message || '未知错误') + ')_';
+            return { ...m, content: (m.content || '') + note };
+          }));
+          setStreamingId(null);
+          abortRef.current = null;
+          setFreshMsgIds(prev => new Set(prev).add(assistantId));
+          loadConversations();
+          gotDone = true;
+          return;
+        }
+      }
+    }
+    if (!gotDone) {
+      // 流结束但未收到 done：已累积的 delta 仍可展示，但视为异常兜底
+      finalizeFlush();
+      abortRef.current = null;
+      setStreamingId(null);
+      throw new Error('流式响应未正常结束');
+    }
+  };
+
+  // Handle send message - CALLS REAL API（Phase 0：默认 SSE 流式，失败回退非流式）
+  // 抽出公共发送逻辑，供输入框发送与「模糊创作意图」方向选择复用
+  const sendText = async (content: string) => {
+    if (!content.trim() || sending || !selectedConversation) return;
+    setSending(true);
+    setSendingTimeout(false);
+    // 60s 超时警告
+    if (sendingTimerRef.current) clearTimeout(sendingTimerRef.current);
+    sendingTimerRef.current = setTimeout(() => {
+      setSendingTimeout(true);
+    }, 60000);
+
+    // Add user message optimistically
+    const tempUserMsg: Message = {
+      id: `temp_${Date.now()}`,
+      conversationId: selectedConversation?.id || '',
+      role: 'user',
+      content,
+      tokenUsed: 0,
+      createdAt: new Date().toISOString(),
+    };
+    setMessages(prev => [...prev, tempUserMsg]);
+
+    const payload = {
+      content,
+      model_id: selectedModelId || undefined,
+      agent_ids: selectedAgentIds.length > 0 ? selectedAgentIds : undefined,
+      skill_ids: selectedSkillIds.length > 0 ? selectedSkillIds : undefined,
+      knowledge_base_ids: selectedKbIds.length > 0 ? selectedKbIds : undefined,
+      mcp_ids: selectedMcpIds.length > 0 ? selectedMcpIds : undefined,
+      attachments: attachments.length > 0 ? attachments.map(a => ({
+        name: a.name, file_type: a.file_type, content: a.content || '',
+      })) : undefined,
+      use_tools: selectedSkillIds.length > 0 || selectedAgentIds.length > 0 || selectedMcpIds.length > 0,
+    };
+
+    // 先放一个空的助手消息占位，用于 SSE 流式增量渲染
+    const assistantId = `temp_assistant_${Date.now()}`;
+    setMessages(prev => [...prev, {
+      id: assistantId,
+      conversationId: selectedConversation!.id,
+      role: 'assistant',
+      content: '',
+      tokenUsed: 0,
+      createdAt: new Date().toISOString(),
+      reasoning: '',
+    } as Message]);
+    setStreamingId(assistantId);
+
+    try {
+      await streamSend(selectedConversation!.id, payload, tempUserMsg, assistantId);
     } catch (err: any) {
-      // Remove optimistic message on error
+      // 移除流式占位，回退非流式重新发送
+      setMessages(prev => prev.filter(m => m.id !== assistantId && m.id !== tempUserMsg.id));
+      setStreamingId(null);
+      try {
+        await sendLegacy(payload, tempUserMsg);
+      } catch (e2: any) {
         setMessages(prev => prev.filter(m => m.id !== tempUserMsg.id));
-      alert('发送失败: ' + (err.message || err.detail || '未知错误'));
+        alert('发送失败: ' + (e2?.message || e2?.detail || '未知错误'));
+      }
     } finally {
+      if (sendingTimerRef.current) { clearTimeout(sendingTimerRef.current); sendingTimerRef.current = null; }
       setSending(false);
+      setSendingTimeout(false);
       inputRef.current?.focus();
     }
   };
@@ -1242,14 +1510,30 @@ const ChatPage: React.FC = () => {
                   {msg.role === 'assistant' && steps.length > 0 && (
                     <AiProcessBlock steps={steps} autoExpand={freshMsgIds.has(msg.id)} />
                   )}
+                  {msg.role === 'assistant' && (msg as any).reasoning && (
+                    <ReasoningBlock text={(msg as any).reasoning} fresh={freshMsgIds.has(msg.id)} />
+                  )}
                   <div className={`message-bubble ${msg.role}`}>
                     {clarify ? (
                       <ClarifyCard
                         data={clarify}
                         onChoose={(opt: any) => sendText(`${opt.medium}：${clarify.prompt}`)}
                       />
-                    ) : msg.role === 'assistant' ? (
+                    ) : msg.role === 'assistant' && isAiError(msg.content) ? (
+                      <div className="ai-error-bubble">
+                        <div className="ai-error-icon">⚠️</div>
+                        <div className="ai-error-text">
                           <MarkdownContent content={msg.content} />
+                        </div>
+                        <button className="ai-error-retry" onClick={() => {
+                          setMessages(prev => prev.filter(m => m.id !== msg.id));
+                          const msgsAfterFilter = messages.filter(m => m.id !== msg.id);
+                          const lastUser = [...msgsAfterFilter].reverse().find(m => m.role === 'user');
+                          if (lastUser) sendText(lastUser.content);
+                        }}>🔄 重试</button>
+                      </div>
+                    ) : msg.role === 'assistant' ? (
+                      msg.content ? <MarkdownContent content={msg.content} /> : (streamingId === msg.id ? <span className="streaming-hint"><IconSpinner /> 生成中…</span> : null)
                     ) : (
                       <span>{msg.content}</span>
                     )}
@@ -1313,11 +1597,20 @@ const ChatPage: React.FC = () => {
                   );
                 })
               )}
-              {sending && (
+              {sending && !streamingId && (
                 <div className="message-row assistant">
-                  <div className="message-bubble assistant" style={{ display: 'flex', alignItems: 'center', gap: 8 }}>
+                  <div className={`message-bubble assistant${sendingTimeout ? ' ai-sending-timeout' : ''}`} style={{ display: 'flex', alignItems: 'center', gap: 8 }}>
                     <IconSpinner />
-                    <span>思考中...</span>
+                    <span>{sendingTimeout ? '⏳ 响应时间较长，AI 仍在处理中...' : '思考中...'}</span>
+                    {sendingTimeout && (
+                      <button className="ai-cancel-btn" onClick={() => {
+                        setSending(false);
+                        setSendingTimeout(false);
+                        if (sendingTimerRef.current) clearTimeout(sendingTimerRef.current);
+                      }} title="取消等待" style={{ marginLeft: 12, padding: '2px 8px', fontSize: 12, cursor: 'pointer', background: 'var(--color-bg-tertiary)', border: '1px solid var(--color-border)', borderRadius: 4, color: 'var(--color-text-secondary)' }}>
+                        取消
+                      </button>
+                    )}
                   </div>
                 </div>
               )}
@@ -1508,11 +1801,11 @@ const ChatPage: React.FC = () => {
                   </button>
                   <button
                     className="send-btn"
-                    onClick={handleSend}
+                    onClick={streamingId ? stopStreaming : handleSend}
-                    disabled={!inputValue.trim() || sending}
+                    disabled={!streamingId && (!inputValue.trim() || sending)}
-                    title="发送 (Enter)"
+                    title={streamingId ? '停止生成' : '发送 (Enter)'}
                   >
-                    {sending ? <IconSpinner /> : <IconSend />}
+                    {streamingId ? <IconStop /> : (sending ? <IconSpinner /> : <IconSend />)}
                   </button>
                 </div>
               </div>

frontend/src/types/index.ts

@@ -60,6 +60,7 @@ export interface Message {
   content: string;
   tokenUsed: number;
   createdAt: string;
+  reasoning?: string;
   used_skills?: string[];
   used_mcp?: string[];
   process_steps?: Array<{