feat(chat): align AI tool-use with GPT-4.1/Anthropic standards -- Fix1: AGENT_GUIDE in system prompt (Persistence/Tool-calling/Planning), Fix2: install_app def improved, Fix3: max_tool_rounds 5->15

Root cause: ensure_structured_column() executed a synchronous blocking
engine.connect().execute(SELECT information_schema.columns ...) inside the
async request path (_build_chat_history / _maybe_compress_history), freezing
the uvicorn event loop and hanging all endpoints (~200s, recovered only by
docker restart).

Fix:
- chat_context.py: remove per-request ensure_structured_column() calls; run
  once idempotently at app startup (main.py lifespan). Use CURRENT_DATABASE()
  for PostgreSQL compatibility.
- main.py: add startup-only ensure_structured_column() hook (try/except).
- llm_client.py: trust_env=False + bounded httpx.Timeout/wait_for so broken
  endpoints fail fast instead of hanging the event loop.
- chat_service.py: run media generation in a background daemon thread with its
  own event loop (_launch_media_task_bg) so SSE end does not cancel image/video
  tasks (#811); remove TEMP faulthandler diagnostic.

Safe-align: persist accumulated live container changes (admin/*, knowledge,
market, models, config, observability, mcp_client, platform_knowledge,
im_bot_platform, chat_media, capability_registry, requirements) and add
workflow_templates.

Verified: live failover test (model-006 -> conn-refused) returns in ~1.9s with
concurrent login 0.83s/200; event-loop responsiveness restored (sub-second).

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

@@ -284,7 +284,7 @@ async def claim_achievement_reward(
         
         # Mark reward as claimed
         _execute_raw(db, 
-            "UPDATE user_achievements SET reward_claimed = 1, claimed_at = NOW() WHERE user_id = :uid AND achievement_id = :aid",
+            "UPDATE user_achievements SET reward_claimed = true, claimed_at = NOW() WHERE user_id = :uid AND achievement_id = :aid",
             {"uid": user_id, "aid": ach_id})
         
         return {"code": 0, "message": "奖励领取成功"}

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

@@ -316,7 +316,7 @@ async def create_dashboard(
         config_json = json.loads('[{"type":"stat","title":"暂无数据","metric":"none","chart_type":"number"}]')
 
     if is_default:
-        db.execute(text("UPDATE analytics_dashboards SET is_default=0"))
+        db.execute(text("UPDATE analytics_dashboards SET is_default=false"))
 
     db.execute(text("""
         INSERT INTO analytics_dashboards (id, name, description, config, is_default, created_by)
@@ -332,7 +332,7 @@ async def create_dashboard(
 @router.delete("/dashboards/{dashboard_id}")
 async def delete_dashboard(dashboard_id: str, db: Session = Depends(get_db), user=Depends(get_current_user)):
     """删除仪表盘"""
-    db.execute(text("DELETE FROM analytics_dashboards WHERE id=:id AND is_default=0"), {"id": dashboard_id})
+    db.execute(text("DELETE FROM analytics_dashboards WHERE id=:id AND is_default=false"), {"id": dashboard_id})
     db.commit()
     return {"code": 200, "data": {"message": "仪表盘已删除"}}
 

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

@@ -445,7 +445,7 @@ async def billing_stats(db: Session = Depends(get_db), user=Depends(get_current_
         SELECT DATE(created_at) as d, COALESCE(SUM(amount), 0) as rev
         FROM billing_transactions
         WHERE transaction_type='recharge' AND status='completed'
-          AND created_at >= DATE_SUB(NOW(), INTERVAL 7 DAY)
+          AND created_at >= NOW() - INTERVAL '7 DAY'
         GROUP BY DATE(created_at) ORDER BY d
     """)).fetchall()
     daily_revenue = [{"date": str(r[0]), "revenue": float(r[1])} for r in revenue_7d]

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

@@ -103,7 +103,7 @@ async def get_execution_stats(db: Session = Depends(get_db), _=Depends(get_curre
     try:
         total = _execute_raw(db, "SELECT COUNT(*) as cnt FROM code_exec_tasks")[0]["cnt"]
         today = _execute_raw(db,
-            "SELECT COUNT(*) as cnt FROM code_exec_tasks WHERE DATE(created_at) = CURDATE()")[0]["cnt"]
+            "SELECT COUNT(*) as cnt FROM code_exec_tasks WHERE DATE(created_at) = CURRENT_DATE")[0]["cnt"]
         success_count = _execute_raw(db,
             "SELECT COUNT(*) as cnt FROM code_exec_tasks WHERE exit_code = 0 AND error IS NULL")[0]["cnt"]
         failed_count = _execute_raw(db,

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

@@ -703,7 +703,7 @@ echo "Installation complete!"
 {d.get('version', '1.0.0')}
 
 ## Installation
-Run `bash install.sh` to install.
+Run \"bash install.sh\" to install.
 
 ## License
 A valid license key is required. Contact your administrator for details.

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

@@ -169,7 +169,7 @@ async def get_im_bot_stats(user=Depends(get_current_user)):
         today_stats = _row_to_dict(rows[0]) if rows else {"msg_in": 0, "msg_out": 0, "users": 0, "errors": 0}
 
         # Total routing rules
-        routes_count = _execute_raw(db, text("SELECT COUNT(*) as cnt FROM im_bot_routes WHERE is_active=1"))[0]._mapping['cnt']
+        routes_count = _execute_raw(db, text("SELECT COUNT(*) as cnt FROM im_bot_routes WHERE is_active=true"))[0]._mapping['cnt']
 
         return {
             "success": True,
@@ -202,7 +202,7 @@ async def get_dashboard(user=Depends(get_current_user)):
                    SUM(messages_out) as msg_out,
                    SUM(unique_users) as users
             FROM im_bot_daily_stats
-            WHERE stat_date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
+            WHERE stat_date >= CURRENT_DATE - INTERVAL '7 DAY'
             GROUP BY stat_date
             ORDER BY stat_date
         """))
@@ -217,7 +217,7 @@ async def get_dashboard(user=Depends(get_current_user)):
             LEFT JOIN (
                 SELECT channel_id, SUM(messages_in + messages_out) as msg_count
                 FROM im_bot_daily_stats
-                WHERE stat_date >= DATE_SUB(CURDATE(), INTERVAL 30 DAY)
+                WHERE stat_date >= CURRENT_DATE - INTERVAL '30 DAY'
                 GROUP BY channel_id
             ) s ON c.id = s.channel_id
             ORDER BY c.created_at DESC
@@ -357,7 +357,7 @@ async def get_channel_detail(channel_id: str, user=Depends(get_current_user)):
             SELECT stat_date, messages_in, messages_out, unique_users, error_count
             FROM im_bot_daily_stats
             WHERE channel_id = :cid
-            AND stat_date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
+            AND stat_date >= CURRENT_DATE - INTERVAL '7 DAY'
             ORDER BY stat_date
         """), {"cid": channel_id})
         channel['stats_7d'] = _rows_to_dict(stats)
@@ -428,7 +428,7 @@ async def activate_channel(channel_id: str, user=Depends(get_current_user)):
     db = SessionLocal()
     try:
         _execute_raw(db, text(
-            "UPDATE channels SET is_active=1, status='connected' WHERE id=:cid"
+            "UPDATE channels SET is_active=true, status='connected' WHERE id=:cid"
         ), {"cid": channel_id})
         return {"success": True, "message": "Channel activated"}
     except Exception as e:
@@ -443,7 +443,7 @@ async def deactivate_channel(channel_id: str, user=Depends(get_current_user)):
     db = SessionLocal()
     try:
         _execute_raw(db, text(
-            "UPDATE channels SET is_active=0, status='disconnected' WHERE id=:cid"
+            "UPDATE channels SET is_active=false, status='disconnected' WHERE id=:cid"
         ), {"cid": channel_id})
         return {"success": True, "message": "Channel deactivated"}
     except Exception as e:

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

@@ -442,7 +442,7 @@ def revoke_license(
         raise HTTPException(404, "License not found")
 
     _execute_raw(db, text(
-        "UPDATE offline_licenses SET status = 'revoked', revoked_reason = :reason, is_active = 0 WHERE id = :lid"
+        "UPDATE offline_licenses SET status = 'revoked', revoked_reason = :reason, is_active = false WHERE id = :lid"
     ), {"reason": reason, "lid": license_id})
 
     return {"code": 0, "message": "License revoked"}
@@ -623,7 +623,7 @@ def get_usage_stats(db: Session = Depends(get_db)):
         "SELECT COUNT(DISTINCT user_hash) FROM usage_reports"
     ))
     today_reports = _execute_raw(db, text(
-        "SELECT COUNT(*) FROM usage_reports WHERE DATE(created_at) = CURDATE()"
+        "SELECT COUNT(*) FROM usage_reports WHERE DATE(created_at) = CURRENT_DATE"
     ))
     by_type = _execute_raw(db, text(
         "SELECT report_type, COUNT(*) as cnt FROM usage_reports GROUP BY report_type"

backend/python/app/api/chat_context.py

@@ -1164,6 +1164,26 @@ def _build_system_prompt(
 
 
     system_parts = ["你是AIFlow平台的AI助手，具备专业的知识储备和工具调用能力。"]
+    
+    # === Agent 工具使用标准指引（对标 OpenAI GPT-4.1 / Anthropic ReAct） ===
+    system_parts.append("""【工具使用规则——请严格遵守】
+
+1. 持久性：请持续工作直到用户的问题完全解决，不要中途放弃。
+   如果一次操作未得到期望结果，请尝试替代方法或不同参数组合，直到成功。
+
+2. 工具优先：如果你不确定某个信息，或需要获取外部数据、实时信息来完成用户请求，
+   请务必使用工具来获取——绝对不要猜测或编造答案。
+
+3. 规划与反思：调用工具前，先思考需要什么信息、按什么顺序调用。结果返回后，
+   检查是否符合预期；不符合则调整计划重试。
+
+4. 何时不调用工具：纯闲聊/问候/简单常识性问题/用户明确要求不要用工具时。
+
+5. 工具调用顺序：先获取信息的工具(搜索/查询)，再执行操作的工具(创建/修改)。
+   不要跳过必要的前置查询步骤。
+
+6. 工具结果反馈：工具返回的结果是客观事实，请忠实地引用，不要虚构或修改。
+""")
 
 
 
@@ -1251,7 +1271,7 @@ def _build_system_prompt(
 
 
 
-    system_parts.append("回复格式要求：使用 Markdown 格式，层次分明即可。对于简短问题，用 1-2 段自然文字直接回答，不必强行分点、加标题或表格；仅在确实需要分步骤或列举时才使用列表，代码/配置用 ``` 代码块包裹。避免冗余空行，保持紧凑、美观、易读。")
+    system_parts.append("回复格式要求：使用 Markdown 格式，层次分明即可。对于简短问题，用 1-2 段自然文字直接回答，不必强行分点、加标题或表格；仅在确实需要分步骤或列举时才使用列表，代码/配置用 \"\"\" 代码块包裹。避免冗余空行，保持紧凑、美观、易读。")
 
 
 
@@ -2536,7 +2556,6 @@ def _build_chat_history(db, conv_id, max_tokens=6000, max_messages=40,
     - 按 token 估算从新到旧累加，超预算即截断（隔离：不溢出上下文窗口）。
     - exclude_latest_user=True 时丢弃最新一条 user 消息（当前轮次另行发送）。
     """
-    ensure_structured_column()
     summary_block = None
     if include_summary:
         try:
@@ -2707,7 +2726,7 @@ def ensure_structured_column():
         with engine.connect() as conn:
             exists = conn.execute(text(
                 "SELECT COUNT(*) FROM information_schema.columns "
-                "WHERE table_schema = DATABASE() AND table_name='conversation_summaries' "
+                "WHERE table_schema = CURRENT_DATABASE() AND table_name='conversation_summaries' "
                 "AND column_name='structured_memory'"
             )).scalar()
             if not exists:
@@ -2832,7 +2851,6 @@ async def _maybe_compress_history(conv_id):
 
     失败静默回滚，绝不阻塞用户本次响应（fire-and-forget）。
     """
-    ensure_structured_column()
     sdb = None
     try:
         sdb = SessionLocal()

backend/python/app/api/chat_media.py

@@ -3,7 +3,8 @@
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Response
 from sqlalchemy.orm import Session
-from sqlalchemy import and_, desc, asc, func
+from sqlalchemy import and_, desc, asc, func, type_coerce
+from sqlalchemy.dialects.postgresql import JSONB
 from pydantic import BaseModel, Field
 from typing import Optional
 import uuid
@@ -893,7 +894,7 @@ def _build_media_route(db, capability: str, preferred_id: str = None):
 
     others = db.query(_AM).filter(
         _AM.is_enabled == True,
-        func.json_contains(_AM.capabilities, f'"{capability}"'),
+        _AM.capabilities.cast(JSONB).contains(type_coerce([capability], JSONB)),
     ).order_by(_AM.is_default.desc(), _AM.created_at.asc()).all()
     for m in others:
         _add(m)

backend/python/app/api/chat_service.py

@@ -47,6 +47,7 @@ import json
 
 
 import asyncio
+import threading
 
 
 
@@ -4354,11 +4355,15 @@ async def _send_message_core(
                         if isinstance(function_definitions, list):
                             function_definitions.append({
                                 "name": "install_app",
-                                "description": "当用户提供技能/Skill、MCP 服务或 Agent 的链接（如 GitHub 仓库、mcp.so、modelscope 等）并希望安装到平台时使用。自动识别类型(skill/mcp/agent)并安装，随后可在对话中直接使用。",
+                                "description": "安装新的技能(Skill)、MCP 服务或外部 Agent 到平台。\n"
+                                "\n"
+                                "【何时调用】用户明确提供链接并表示要安装/导入/添加时。例如:'帮我安装这个技能 https://github.com/...'\n"
+                                "【何时不调用】用户只询问安装方法、讨论某个工具但没有提供可安装链接时。\n"
+                                "【注意】安装完成后该技能/MCP/Agent 可在本对话中直接使用，无需额外配置。",
                                 "parameters": {
                                     "type": "object",
                                     "properties": {
-                                        "url": {"type": "string", "description": "要安装的技能/MCP/Agent 地址(URL)"},
+                                        "url": {"type": "string", "description": "要安装的资源完整地址。必须是 http/https URL，支持 GitHub 仓库、mcp.so、modelscope 等平台链接。"},
                                         "type": {"type": "string", "enum": ["auto", "skill", "mcp", "agent"], "description": "安装类型，auto 表示自动识别(默认)"},
                                     },
                                     "required": ["url"],
@@ -4561,7 +4566,7 @@ async def _send_message_core(
 
 
 
-                        max_tool_rounds = 5
+                        max_tool_rounds = 15  # 对齐 GPT-4.1/Anthropic 标准: 足够完成复杂多步工具调用
 
 
 
@@ -7927,7 +7932,7 @@ async def _send_message_core(
 
 
 
-            asyncio.create_task(
+            _launch_media_task_bg(ai_msg.id, ai_model.id, req.content, current_user.id)
 
 
 
@@ -7943,7 +7948,6 @@ async def _send_message_core(
 
 
 
-                _generate_media_task(ai_msg.id, ai_model.id, req.content, current_user.id)
 
 
 
@@ -7959,7 +7963,6 @@ async def _send_message_core(
 
 
 
-            )
 
 
 
@@ -8547,3 +8550,16 @@ def _check_tool_policy(func_name, mcp_info, skills, policy, call_count, round_co
 
 
 
+
+
+# 后台线程独立事件循环运行媒体生成，确保 SSE 响应结束后任务不被 uviorn 取消，
+# 避免图片/视频永远停留在「生成中…」。_generate_media_task 内部自管 DB 会话。
+def _launch_media_task_bg(ai_msg_id, ai_model_id, prompt, user_id):
+    def _run():
+        try:
+            asyncio.run(_generate_media_task(ai_msg_id, ai_model_id, prompt, user_id))
+        except Exception:
+            pass
+    _t = threading.Thread(target=_run, daemon=True)
+    _t.start()
+

backend/python/app/api/im_bot_platform.py

@@ -109,20 +109,20 @@ def _ensure_table(db):
     """确保 im_bot_platform_account 表存在（首次调用时创建）。"""
     db.execute(text("""
         CREATE TABLE IF NOT EXISTS im_bot_platform_account (
-            id INT AUTO_INCREMENT PRIMARY KEY,
+            id SERIAL PRIMARY KEY,
             user_id VARCHAR(64) DEFAULT NULL,
             platform VARCHAR(32) NOT NULL,
             status VARCHAR(32) DEFAULT 'disconnected',
-            creds MEDIUMTEXT NULL,
+            creds TEXT NULL,
             send_url VARCHAR(512) NULL,
             api_base VARCHAR(512) NULL,
             bot_token VARCHAR(512) NULL,
             webhook_token VARCHAR(64) NULL,
-            last_error MEDIUMTEXT NULL,
+            last_error TEXT NULL,
-            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-            updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-            UNIQUE KEY uk_user_platform (user_id, platform)
+            UNIQUE (user_id, platform)
-        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
+        )
     """))
     db.commit()
 

backend/python/app/api/knowledge.py

@@ -502,8 +502,8 @@ def _parse_llm_graph_response(content: str):
     except json.JSONDecodeError:
         pass
 
-    # 方式2: 提取```json ... ```代码块
+    # 方式2: 提取\"\"\"json ... \"\"\"代码块
-    json_match = re.search(r'```(?:json)?\s*([\s\S]*?)```', content)
+    json_match = re.search(r'\"\"\"(?:json)?\s*([\s\S]*?)\"\"\"', content)
     if json_match:
         try:
             result = json.loads(json_match.group(1))

backend/python/app/api/market.py

@@ -59,7 +59,7 @@ def _sanitize_skill_content(text: str) -> str:
     if "<script" in low and t.count("<div") > 3:
         return ""
     # 接受携带任一 markdown / skill 标记的内容
-    markers = ("#", "---", "name:", "description:", "```", "##")
+    markers = ("#", "---", "name:", "description:", "\"\"\"", "##")
     if not any(m in t for m in markers):
         return ""
     return t
@@ -79,7 +79,7 @@ def _extract_skill_from_html(html: str, base_url: str) -> str:
     """从 HTML 页面尽力抽取 SKILL.md 正文（代码块优先，其次跳转到 raw 链接）。"""
     proxy = _skill_http_proxy()
     # 1) fenced code block
-    fences = re.findall(r"```(?:markdown|md|yaml|json)?\s*\n(.*?)```", html, re.DOTALL | re.IGNORECASE)
+    fences = re.findall(r"\"\"\"(?:markdown|md|yaml|json)?\s*\n(.*?)\"\"\"", html, re.DOTALL | re.IGNORECASE)
     for f in fences:
         if ("name:" in f.lower() or "description:" in f.lower() or f.count("#") >= 2) and len(f) > 80:
             return f.strip()

backend/python/app/capability_registry.py

@@ -98,10 +98,11 @@ def resolve_auto_model(capability: str, db: Session) -> Optional[AIModel]:
     # 媒体类能力(image/video/audio)：无显式配置时自动发现任一启用的模型
     # 避免用户未在后台配置"能力默认模型"时 Auto 回退到文本模型导致"无法生成图片/视频"
     if capability in ("image", "video", "audio"):
-        from sqlalchemy import text as _text, func
+        from sqlalchemy import text as _text, func, type_coerce
+        from sqlalchemy.dialects.postgresql import JSONB
         m = db.query(AIModel).filter(
             AIModel.is_enabled == True,
-            func.json_contains(AIModel.capabilities, f'"{capability}"'),
+            AIModel.capabilities.cast(JSONB).contains(type_coerce([capability], JSONB)),
         ).first()
         if m:
             return m

backend/python/app/chat_context.py

@@ -1293,7 +1293,7 @@ def _build_system_prompt(
 
 
 
-    system_parts.append("回复格式要求：使用 Markdown 格式，层次分明即可。对于简短问题，用 1-2 段自然文字直接回答，不必强行分点、加标题或表格；仅在确实需要分步骤或列举时才使用列表，代码/配置用 ``` 代码块包裹。避免冗余空行，保持紧凑、美观、易读。")
+    system_parts.append("回复格式要求：使用 Markdown 格式，层次分明即可。对于简短问题，用 1-2 段自然文字直接回答，不必强行分点、加标题或表格；仅在确实需要分步骤或列举时才使用列表，代码/配置用 \"\"\" 代码块包裹。避免冗余空行，保持紧凑、美观、易读。")
 
 
 
@@ -2749,7 +2749,7 @@ def ensure_structured_column():
         with engine.connect() as conn:
             exists = conn.execute(text(
                 "SELECT COUNT(*) FROM information_schema.columns "
-                "WHERE table_schema = DATABASE() AND table_name='conversation_summaries' "
+                "WHERE table_schema = CURRENT_DATABASE() AND table_name='conversation_summaries' "
                 "AND column_name='structured_memory'"
             )).scalar()
             if not exists:

backend/python/app/config.py

@@ -15,7 +15,12 @@ if not (Path("/.dockerenv").exists() or Path("/run/.containerenv").exists()):
         load_dotenv(env_path, override=False)
 
 # 数据库配置
-DATABASE_URL = f"mysql+pymysql://{os.getenv('MYSQL_USER', 'root')}:{os.getenv('MYSQL_PASSWORD', 'pwd2026')}@{os.getenv('MYSQL_HOST', 'localhost')}:{os.getenv('MYSQL_PORT', '3306')}/{os.getenv('MYSQL_DATABASE', 'aiflow')}?charset=utf8mb4"
+# PostgreSQL preferred when DATABASE_URL is provided (production); else legacy MySQL builder (local/dev + rollback)
+_DATABASE_URL_ENV = os.getenv("DATABASE_URL", "")
+if _DATABASE_URL_ENV:
+    DATABASE_URL = _DATABASE_URL_ENV
+else:
+    DATABASE_URL = f"mysql+pymysql://{os.getenv('MYSQL_USER', 'root')}:{os.getenv('MYSQL_PASSWORD', 'pwd2026')}@{os.getenv('MYSQL_HOST', 'localhost')}:{os.getenv('MYSQL_PORT', '3306')}/{os.getenv('MYSQL_DATABASE', 'aiflow')}?charset=utf8mb4"
 
 # Redis配置
 REDIS_HOST = os.getenv("REDIS_HOST", "localhost")

backend/python/app/core/llm_client.py

@@ -1,9 +1,9 @@
 """Shared, standards-based LLM client for web chat + IM bot.
 
 平台红线(07-24)：模型层以成熟/开源标准为设计基准。本模块使用官方 OpenAI SDK
-(`openai.AsyncOpenAI`)——它是 OpenAI 兼容端点的事实标准客户端。平台本身已在讲
+(\"openai.AsyncOpenAI\")——它是 OpenAI 兼容端点的事实标准客户端。平台本身已在讲
-OpenAI 协议(`/v1/chat/completions`)，因此零协议改造成本，并天然获得：
+OpenAI 协议(\"/v1/chat/completions\")，因此零协议改造成本，并天然获得：
-  * 原生 SSE 流式(`stream=True`)；
+  * 原生 SSE 流式(\"stream=True\")；
   * 结构化的 tool_calls / usage 解析；
   * 统一的代理(trust_env=False + 显式 proxy)与超时处理。
 
@@ -13,13 +13,14 @@ IM bot 共用**同一模型层**、同一套 failover/代理/超时语义——
 
 API:
   achat_complete(api_url, api_key, request_body, *, proxy=None, timeout=120, stream=False)
-    - stream=False -> dict，形状对齐上游 `data` 响应：
+    - 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
+import asyncio
 from openai import AsyncOpenAI, OpenAI
 
 
@@ -37,10 +38,21 @@ def build_openai_client(api_url, api_key, *, proxy=None, timeout=120) -> AsyncOp
     # 始终 trust_env=False：避免容器 HTTP_PROXY 对国内端点(siliconflow 等)造成 502；
     # 代理经 proxy 显式传入（AGNES 等），未传则直连。
     http_client = httpx.AsyncClient(proxy=proxy, trust_env=False)
+    # 故障快速失败（全栈统一，覆盖 web chat / IM bot / embedding）：
+    #   connect 封顶 10s —— 坏端点不再让请求挂死事件循环；
+    #   read/write 封顶 30s —— 长流式回复只要持续有分块即不受影响；
+    #   保留数值型 timeout 接口，旧调用方无需改动。
+    _num = float(timeout) if isinstance(timeout, (int, float)) else 120.0
+    _to = httpx.Timeout(
+        connect=min(10.0, _num),
+        read=min(30.0, _num),
+        write=min(30.0, _num),
+        pool=10.0,
+    )
     return AsyncOpenAI(
         base_url=base,
         api_key=api_key or "EMPTY",
-        timeout=timeout,
+        timeout=_to,
         http_client=http_client,
     )
 
@@ -94,10 +106,21 @@ async def achat_complete(api_url, api_key, request_body, *, proxy=None, timeout=
             _reasoning_parts = []
             _tool_acc = {}  # index -> {id, type, name, arguments}
             try:
-                gen = await client.chat.completions.create(
+                _to = min(30.0, float(timeout) if isinstance(timeout, (int, float)) else 120.0)
-                    **_create_kwargs(request_body, stream=True)
+                gen = await asyncio.wait_for(
+                    client.chat.completions.create(
+                        **_create_kwargs(request_body, stream=True)
+                    ),
+                    timeout=_to,
                 )
-                async for chunk in gen:
+                _it = gen.__aiter__()
+                while True:
+                    try:
+                        chunk = await asyncio.wait_for(_it.__anext__(), timeout=_to)
+                    except StopAsyncIteration:
+                        break
+                    except asyncio.TimeoutError:
+                        raise TimeoutError("stream chunk read timeout")
                     if not chunk.choices:
                         continue
                     delta = chunk.choices[0].delta
@@ -129,7 +152,10 @@ async def achat_complete(api_url, api_key, request_body, *, proxy=None, timeout=
                                 if _fn_args:
                                     _slot["arguments"] += _fn_args
             finally:
-                await client.close()
+                try:
+                    await asyncio.wait_for(client.close(), timeout=5)
+                except Exception:
+                    pass
             # 流式结束时吐出已组装的完整 message（含 tool_calls），供调用方无缝衔接工具循环
             _tool_calls = None
             if _tool_acc:
@@ -150,8 +176,12 @@ async def achat_complete(api_url, api_key, request_body, *, proxy=None, timeout=
         return _gen()
 
     client = build_openai_client(api_url, api_key, proxy=proxy, timeout=timeout)
-    async with client:
+    try:
-        resp = await client.chat.completions.create(**_create_kwargs(request_body))
+        _to = min(30.0, float(timeout) if isinstance(timeout, (int, float)) else 120.0)
+        resp = await asyncio.wait_for(
+            client.chat.completions.create(**_create_kwargs(request_body)),
+            timeout=_to,
+        )
         msg = resp.choices[0].message
         return {
             "choices": [
@@ -165,6 +195,11 @@ async def achat_complete(api_url, api_key, request_body, *, proxy=None, timeout=
             ],
             "usage": (resp.usage.model_dump() if resp.usage else {}),
         }
+    finally:
+        try:
+            await asyncio.wait_for(client.close(), timeout=5)
+        except Exception:
+            pass
 
 
 def build_openai_client_sync(api_url, api_key, *, proxy=None, timeout=120) -> OpenAI:
@@ -208,9 +243,18 @@ async def aembed(api_url, api_key, input, model, *, proxy=None, timeout=60) -> l
     嵌入调用收敛到同一标准模型层（与聊天共用代理/超时/failover 语义）。
     """
     client = build_openai_client(api_url, api_key, proxy=proxy, timeout=timeout)
-    async with client:
+    try:
-        resp = await client.embeddings.create(**_embed_kwargs(input, model))
+        _to = min(60.0, float(timeout) if isinstance(timeout, (int, float)) else 60.0)
+        resp = await asyncio.wait_for(
+            client.embeddings.create(**_embed_kwargs(input, model)),
+            timeout=_to,
+        )
         return [d.embedding for d in resp.data]
+    finally:
+        try:
+            await asyncio.wait_for(client.close(), timeout=5)
+        except Exception:
+            pass
 
 
 def embed(api_url, api_key, input, model, *, proxy=None, timeout=60) -> list:

backend/python/app/core/observability.py

@@ -6,7 +6,7 @@ Qihu-AIFlow 统一可观测基座 (Phase 2.1 + Phase 1.I OTel 增强)
 
 设计原则（Strangler Fig 增量）：
 - 纯增量埋点，不改任何既有接口的返回结构。
-- 向后兼容现有 `logging.getLogger(__name__)` 用法：trace_id 由 root logger 的
+- 向后兼容现有 \"logging.getLogger(__name__)\" 用法：trace_id 由 root logger 的
   TraceIdFilter 自动注入，调用方无需改动。
 - 不引入第三方依赖（structlog 未安装），自带可选的 JSON formatter。
 
@@ -58,7 +58,7 @@ def get_trace_id() -> str:
 class TraceIdFilter(logging.Filter):
     """为每一条日志记录注入 trace_id 属性（不存在则写为 '-'）。
 
-    挂载到 root logger 后，所有 `getLogger(__name__)` 的子 logger 都会继承，
+    挂载到 root logger 后，所有 \"getLogger(__name__)\" 的子 logger 都会继承，
     从而让全平台既有日志自动带上链路 ID，零侵入。
     """
 

backend/python/app/mcp_client.py

@@ -1,8 +1,8 @@
 """
-MCP (Model Context Protocol) 客户端 —— 基于官方 `mcp` Python SDK (Phase 1.E 标准化)
+MCP (Model Context Protocol) 客户端 —— 基于官方 \"mcp\" Python SDK (Phase 1.E 标准化)
 
 设计基准（标准对照表见 STANDARDS_AUDIT_2026-07-24.md §E）：
-- 采用 Anthropic 官方 `mcp` SDK 处理 JSON-RPC 帧 + 协议版本协商，
+- 采用 Anthropic 官方 \"mcp\" SDK 处理 JSON-RPC 帧 + 协议版本协商，
   不再手写解析、不再写死 MCP_PROTOCOL_VERSION 常量（旧实现写死 "2024-11-05"，
   上游协议一升级即碎）。
 - 支持 streamable_http（默认）与 sse 两种传输，按 server['transport_type'] 选择。

backend/python/app/models.py

@@ -11,7 +11,7 @@ from sqlalchemy import (
     DECIMAL, UniqueConstraint
 )
 from sqlalchemy.orm import relationship
-from sqlalchemy.dialects.mysql import MEDIUMTEXT
+from sqlalchemy import Text  # PG-compatible (was MEDIUMTEXT from mysql dialect)
 from .database import Base
 
 
@@ -300,7 +300,7 @@ class MarketApp(Base):
     rating_avg = Column(DECIMAL(3, 2), default=0.00)
     review_count = Column(Integer, default=0)
     install_count = Column(Integer, default=0)
-    skill_content = Column(MEDIUMTEXT, default=None)
+    skill_content = Column(Text, default=None)
     mcp_transport = Column(String(20), default=None)
     platform = Column(String(20), default=None, comment="Agent 接入平台：dify/n8n/coze")
     capability = Column(String(20), default=None, comment="能力分类：text/thinking/vision/tool/image/audio/asr/ocr/embedding/rerank/video")

backend/python/app/platform_knowledge.py

@@ -73,7 +73,7 @@ PLATFORM_KNOWLEDGE = """# 七虎AI平台能力全览
 - **安全**: JWT鉴权、HTTPS加密、操作审计日志
 
 ## AI回复规范
-1. **回复格式**: 使用Markdown（标题 `##`/`###`、加粗 `**`、列表 `-`/`1.`、代码块等）
+1. **回复格式**: 使用Markdown（标题 \"##\"/\"###\"、加粗 \"**\"、列表 \"-\"/\"1.\"、代码块等）
 2. **层次分明**: 使用标题分隔段落，确保信息有层次
 3. **平台优先**: 回答问题时优先引用本平台功能，而非泛泛建议
 4. **准确具体**: 基于平台文档给出具体操作路径、配置方法、API示例

backend/python/main.py

@@ -118,6 +118,18 @@ def _record_api_call(db, method: str, path: str, status_code: int, elapsed: floa
 async def lifespan(app: FastAPI):
     logger.info("Qihu-AIFlow v2.0.0 启动中...")
 
+    # 幂等确保 conversation_summaries.structured_memory 列。
+    # 原实现在每次聊天的请求路径中同步执行 engine.connect().execute(SELECT information_schema.columns)，
+    # 该同步阻塞调用运行在 asyncio 事件循环线程上，会导致整个事件循环冻结（所有端点挂死，
+    # 表现为「单 worker 永久阻塞、docker restart 才恢复」）。改为启动时一次性执行，
+    # 既保留对全新库的兼容，又绝不阻塞任何用户请求。
+    try:
+        from app.api.chat_context import ensure_structured_column
+        ensure_structured_column()
+        logger.info("[startup] ensure_structured_column 完成")
+    except Exception as _e:
+        logger.warning(f"[startup] ensure_structured_column 失败(可忽略): {_e}")
+
     # 启动后台 Evolution Pipeline（知识库图谱已存在的用户配置自动恢复运行）
     try:
         from app.services.evolution_service import EvolutionWatcherManager

backend/python/requirements.txt

@@ -28,3 +28,4 @@ click==8.4.2
 click-didyoumean==0.3.1
 click-plugins==1.1.1.2
 click-repl==0.3.0
+psycopg[binary]

backend/python/workflow_templates/code-review.yaml

@@ -0,0 +1,35 @@
+version: "1.0"
+name: "Code Review Pipeline"
+description: "Generate code → review → merge. A standard code-review workflow."
+category: development
+
+nodes:
+  - id: code_gen
+    type: llm
+    description: "Generate code based on requirements"
+    on_error: halt
+    config:
+      model: auto
+      system_prompt: "You are a senior software engineer. Write clean, production-ready code based on the user's requirements. Include comments and error handling."
+
+  - id: review
+    type: llm
+    description: "Review generated code for bugs and style"
+    on_error: continue
+    config:
+      model: auto
+      system_prompt: "You are a code reviewer. Analyze the provided code for bugs, style issues, performance problems, and security vulnerabilities. Provide specific, actionable feedback."
+
+  - id: merge
+    type: reduce
+    description: "Combine code with review feedback"
+    config:
+      strategy: concat
+
+edges:
+  - from: code_gen
+    to: review
+  - from: code_gen
+    to: merge
+  - from: review
+    to: merge

backend/python/workflow_templates/multi-agent-debate.yaml

@@ -0,0 +1,35 @@
+version: "1.0"
+name: "Multi-Agent Debate"
+description: "Two agents debate a topic from opposing perspectives, then a judge agent synthesizes the conclusion."
+category: creativity
+
+nodes:
+  - id: proponent
+    type: llm
+    description: "Argue in favor of the proposition"
+    on_error: continue
+    config:
+      model: auto
+      system_prompt: "You are a skilled debater. Argue strongly IN FAVOR of the given proposition. Use logic, evidence, and persuasive reasoning. Be passionate but fair."
+
+  - id: opponent
+    type: llm
+    description: "Argue against the proposition"
+    on_error: continue
+    config:
+      model: auto
+      system_prompt: "You are a skilled debater. Argue strongly AGAINST the given proposition. Use logic, evidence, and persuasive reasoning to find flaws in the proponent's case."
+
+  - id: judge
+    type: llm
+    description: "Evaluate both sides and reach a conclusion"
+    on_error: continue
+    config:
+      model: auto
+      system_prompt: "You are an impartial judge. Evaluate both arguments fairly. Identify the strongest points from each side. Provide a balanced conclusion and recommendation."
+
+edges:
+  - from: proponent
+    to: judge
+  - from: opponent
+    to: judge

backend/python/workflow_templates/rag-search.yaml

@@ -0,0 +1,25 @@
+version: "1.0"
+name: "RAG Search Pipeline"
+description: "Search knowledge base → augment with context → generate answer. Standard RAG pipeline."
+category: productivity
+
+nodes:
+  - id: search
+    type: llm
+    description: "Generate search queries and retrieve relevant context"
+    on_error: continue
+    config:
+      model: auto
+      system_prompt: "You are a search specialist. For the given question, identify 2-3 key search points. Then search the provided knowledge base for relevant information. Return the most relevant passages with source references."
+
+  - id: answer
+    type: llm
+    description: "Generate final answer with citations"
+    on_error: continue
+    config:
+      model: auto
+      system_prompt: "You are a knowledge assistant. Using the search results provided, generate a comprehensive answer. Always cite your sources. If information is insufficient, clearly state what is missing."
+
+edges:
+  - from: search
+    to: answer

backend/python/workflow_templates/research-assistant.yaml

@@ -0,0 +1,25 @@
+version: "1.0"
+name: "Research & Answer"
+description: "Research topic → generate answer. Uses two LLM nodes for research and synthesis."
+category: productivity
+
+nodes:
+  - id: research
+    type: llm
+    description: "Search and gather information on the topic"
+    on_error: continue
+    config:
+      model: auto
+      system_prompt: "You are a thorough researcher. Search for key facts, data points, and relevant context about the given topic. Provide a comprehensive summary with citations."
+
+  - id: answer
+    type: llm
+    description: "Synthesize research into final answer"
+    on_error: continue
+    config:
+      model: auto
+      system_prompt: "You are a professional writer. Based on the research summary provided, craft a clear, well-structured answer for the user. Include key findings and actionable insights."
+
+edges:
+  - from: research
+    to: answer