feat(chat): 外部Agent(Dify/n8n/Coze)运行期真实远程调用

- _resolve_market_agents 补齐 platform/agent_endpoint/agent_api_key 字段
- 新增 _call_external_agent 及 dify/n8n/coze 分支; Dify 走SSE流式,兼容
  chat/agent(message)与chatflow/workflow(text_chunk/workflow_finished)事件,
  失败时surface真实错误; n8n走webhook POST; coze走v3 chat+轮询
- send_message 选中带endpoint的外部Agent时直接远程调用,跳过本地LLM扮演
- Conversation新增ext_agent_conversation_id维持Dify/Coze多轮(迁移016)
- 出网统一trust_env=False+_proxy_for_url直连,规避坏代理502

backend/python/app/api/chat.py

@@ -282,6 +282,260 @@ async def _generate_media_task(ai_msg_id: str, ai_model_id: str, prompt: str, us
             pass
 
 
+# ============================================================
+# 外部 Agent 远程调用 (Dify / n8n / Coze)
+# ============================================================
+# 当前台对话关联了一个带有效 agent_endpoint 的外部 Agent 时，运行期直接远程调用
+# 该 Agent 平台，而不是用本地 LLM「扮演」。所有出网请求必须 trust_env=False +
+# _proxy_for_url() 直连，避免容器 env 里的坏代理(172.18.0.1:7890 返回 502)。
+
+
+def _normalize_dify_chat_url(endpoint: str) -> str:
+    """把用户填写的 Dify 地址归一化为 /v1/chat-messages 端点。与 market.py 保持一致。"""
+    ep = (endpoint or "").rstrip("/")
+    if ep.endswith("/chat-messages"):
+        return ep
+    if ep.endswith("/v1"):
+        return ep + "/chat-messages"
+    return ep + "/v1/chat-messages"
+
+
+def _extract_text_from_json(obj) -> str:
+    """从 n8n / 通用 webhook 的任意 JSON 响应中尽力抽取文本回答。"""
+    if obj is None:
+        return ""
+    if isinstance(obj, str):
+        return obj
+    if isinstance(obj, (int, float, bool)):
+        return str(obj)
+    if isinstance(obj, list):
+        # 取第一个可抽取到文本的元素
+        for it in obj:
+            t = _extract_text_from_json(it)
+            if t:
+                return t
+        return ""
+    if isinstance(obj, dict):
+        # 常见回答字段优先级
+        for k in ("output", "text", "answer", "response", "reply",
+                  "message", "content", "result", "data"):
+            if k in obj:
+                t = _extract_text_from_json(obj[k])
+                if t:
+                    return t
+        # 兜底：整体序列化
+        try:
+            return json.dumps(obj, ensure_ascii=False)
+        except Exception:
+            return str(obj)
+    return str(obj)
+
+
+async def _call_dify_agent(endpoint: str, api_key: str, query: str,
+                           user_id: str, ext_conv_id: str | None) -> tuple:
+    """调用 Dify(新版 Agent) /v1/chat-messages，流式(SSE)解析最终回答。
+
+    新版 Agent API 仅支持 response_mode=streaming(blocking 会 400)。事件流中
+    message / agent_message 事件携带增量 answer，累加即为最终回答；每个事件都带
+    conversation_id，用于维持多轮上下文。返回 (answer, new_conversation_id)。
+    """
+    url = _normalize_dify_chat_url(endpoint)
+    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+    payload = {
+        "inputs": {},
+        "query": query,
+        "user": (user_id or "web-user")[:64],
+        "response_mode": "streaming",
+        "conversation_id": ext_conv_id or "",
+    }
+    answer_parts = []
+    new_conv_id = ext_conv_id
+    err_msg = None
+    wf_outputs = None
+    async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False) as client:
+        async with client.stream("POST", url, headers=headers, json=payload) as resp:
+            if resp.status_code != 200:
+                body = (await resp.aread()).decode("utf-8", "replace")
+                raise RuntimeError(f"Dify HTTP {resp.status_code}: {body[:300]}")
+            async for line in resp.aiter_lines():
+                if not line:
+                    continue
+                line = line.strip()
+                if not line.startswith("data:"):
+                    continue
+                data_str = line[5:].strip()
+                if not data_str or data_str == "[DONE]":
+                    continue
+                try:
+                    evt = json.loads(data_str)
+                except Exception:
+                    continue
+                ev = evt.get("event")
+                if evt.get("conversation_id"):
+                    new_conv_id = evt["conversation_id"]
+                data = evt.get("data") or {}
+                # 兼容三类 Dify 应用：
+                #  1) Chat/Agent 应用: message / agent_message 事件带增量 answer
+                #  2) Chatflow/Workflow 应用: text_chunk 事件带 data.text；
+                #     结束时 workflow_finished 带 status / outputs / error
+                if ev in ("message", "agent_message"):
+                    answer_parts.append(evt.get("answer", "") or "")
+                elif ev == "message_replace":
+                    # 内容审查替换：以替换后的完整 answer 为准
+                    answer_parts = [evt.get("answer", "") or ""]
+                elif ev == "text_chunk":
+                    answer_parts.append((data.get("text") or ""))
+                elif ev == "node_finished":
+                    # answer / end 节点的 outputs 可能含最终文本
+                    if data.get("node_type") in ("answer", "end") and data.get("outputs"):
+                        wf_outputs = data.get("outputs")
+                    if data.get("status") == "failed" and data.get("error"):
+                        err_msg = str(data.get("error"))
+                elif ev == "workflow_finished":
+                    if data.get("status") == "failed":
+                        err_msg = str(data.get("error") or "Dify 工作流执行失败")
+                    elif data.get("outputs"):
+                        wf_outputs = data.get("outputs")
+                elif ev == "error":
+                    err_msg = evt.get("message") or evt.get("code") or "Dify 流式返回错误"
+    answer = "".join(answer_parts).strip()
+    # 流式无增量文本时，尝试从 workflow 结束 outputs 兜底提取
+    if not answer and wf_outputs:
+        answer = _extract_text_from_json(wf_outputs).strip()
+    if not answer and err_msg:
+        raise RuntimeError(err_msg)
+    return answer, new_conv_id
+
+
+async def _call_n8n_agent(endpoint: str, api_key: str, query: str,
+                          user_id: str, history: list) -> tuple:
+    """调用 n8n 生产 Webhook (POST)。webhook 用 Respond to Webhook 节点返回 JSON，
+    尽力从中抽取文本回答。n8n webhook 无平台级会话态，多轮靠我们下发 history。
+    返回 (answer, None)。"""
+    url = (endpoint or "").strip()
+    headers = {"Content-Type": "application/json"}
+    if api_key:
+        # n8n Header Auth 自定义头名不固定，同时下发常见几种以提高兼容性
+        headers["Authorization"] = f"Bearer {api_key}"
+        headers["X-N8N-API-KEY"] = api_key
+    # 携带多种字段别名，兼容 n8n AI Agent(chatInput) 与自定义工作流
+    payload = {
+        "query": query,
+        "chatInput": query,
+        "message": query,
+        "input": query,
+        "user": (user_id or "web-user")[:64],
+        "history": history or [],
+    }
+    async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False) as client:
+        resp = await client.post(url, headers=headers, json=payload)
+    if resp.status_code not in (200, 201):
+        raise RuntimeError(f"n8n Webhook HTTP {resp.status_code}: {resp.text[:300]}")
+    ctype = resp.headers.get("content-type", "")
+    if "application/json" in ctype:
+        try:
+            answer = _extract_text_from_json(resp.json())
+        except Exception:
+            answer = resp.text
+    else:
+        answer = resp.text
+    return (answer or "").strip(), None
+
+
+async def _call_coze_agent(endpoint: str, api_key: str, query: str,
+                           user_id: str, ext_conv_id: str | None) -> tuple:
+    """调用扣子(Coze) v3 对话 API：create → 轮询 retrieve → 拉取 message list。
+
+    约定 agent_endpoint 存放 bot_id(纯 id)，或形如
+    https://api.coze.cn/v3/chat?bot_id=xxx 的完整地址。默认基址 api.coze.cn。
+    返回 (answer, conversation_id)。
+    """
+    ep = (endpoint or "").strip()
+    from urllib.parse import urlparse, parse_qs
+    bot_id = None
+    base = "https://api.coze.cn"
+    if ep.startswith("http"):
+        pr = urlparse(ep)
+        base = f"{pr.scheme}://{pr.netloc}"
+        qs = parse_qs(pr.query)
+        bot_id = (qs.get("bot_id") or [None])[0]
+        if not bot_id and "/bot/" in pr.path:
+            bot_id = pr.path.rstrip("/").split("/bot/")[-1]
+    else:
+        bot_id = ep  # 纯 bot_id
+    if not bot_id:
+        raise RuntimeError("Coze 接入缺少 bot_id(请在 Endpoint 填入 bot_id 或带 ?bot_id= 的地址)")
+    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+    create_url = f"{base}/v3/chat"
+    if ext_conv_id:
+        create_url += f"?conversation_id={ext_conv_id}"
+    payload = {
+        "bot_id": bot_id,
+        "user_id": (user_id or "web-user")[:64],
+        "stream": False,
+        "auto_save_history": True,
+        "additional_messages": [
+            {"role": "user", "content": query, "content_type": "text"}
+        ],
+    }
+    async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(create_url), trust_env=False) as client:
+        r = await client.post(create_url, headers=headers, json=payload)
+        if r.status_code != 200:
+            raise RuntimeError(f"Coze create HTTP {r.status_code}: {r.text[:300]}")
+        cd = r.json()
+        if cd.get("code") not in (0, None):
+            raise RuntimeError(f"Coze 错误: {cd.get('msg') or cd}")
+        chat = cd.get("data") or {}
+        chat_id = chat.get("id")
+        conv_id2 = chat.get("conversation_id") or ext_conv_id
+        if not chat_id or not conv_id2:
+            raise RuntimeError(f"Coze 未返回 chat_id/conversation_id: {str(cd)[:200]}")
+        # 轮询直到完成(最多约 3 分钟)
+        status = chat.get("status")
+        for _ in range(60):
+            if status == "completed":
+                break
+            if status in ("failed", "requires_action", "canceled"):
+                raise RuntimeError(f"Coze 对话状态 {status}")
+            await asyncio.sleep(3)
+            rr = await client.get(
+                f"{base}/v3/chat/retrieve?chat_id={chat_id}&conversation_id={conv_id2}",
+                headers=headers,
+            )
+            if rr.status_code == 200:
+                status = ((rr.json() or {}).get("data") or {}).get("status")
+        # 拉取消息列表，取 type=answer 的内容
+        rm = await client.get(
+            f"{base}/v3/chat/message/list?chat_id={chat_id}&conversation_id={conv_id2}",
+            headers=headers,
+        )
+        answer = ""
+        if rm.status_code == 200:
+            for m in ((rm.json() or {}).get("data") or []):
+                if m.get("type") == "answer" and m.get("role") == "assistant":
+                    answer = m.get("content", "") or answer
+    return (answer or "").strip(), conv_id2
+
+
+async def _call_external_agent(db, agent: dict, query: str, user_id: str,
+                               ext_conversation_id: str | None,
+                               history: list) -> tuple:
+    """按平台分发到对应外部 Agent 远程调用。返回 (answer, new_ext_conversation_id)。"""
+    platform = (agent.get("platform") or "dify").lower()
+    endpoint = agent.get("agent_endpoint")
+    api_key = agent.get("agent_api_key") or ""
+    if not endpoint:
+        raise RuntimeError("该 Agent 未配置远程调用地址(agent_endpoint)")
+    if platform == "dify":
+        return await _call_dify_agent(endpoint, api_key, query, user_id, ext_conversation_id)
+    if platform == "n8n":
+        return await _call_n8n_agent(endpoint, api_key, query, user_id, history)
+    if platform == "coze":
+        return await _call_coze_agent(endpoint, api_key, query, user_id, ext_conversation_id)
+    # 未知平台：默认按 Dify 协议尝试
+    return await _call_dify_agent(endpoint, api_key, query, user_id, ext_conversation_id)
+
+
 # ============================================================
 # Request Models
 # ============================================================
@@ -434,6 +688,11 @@ def _resolve_market_agents(db: Session, agent_ids: list, user_id: str) -> list:
             "model_id": agent_model_id,
             "tools": tools,
             "description": ag.description or "",
+            # 外部 Agent 接入(Dify/n8n/Coze)运行期远程调用所需字段：
+            # 这三列此前被丢弃，导致选中外部 Agent 时仍用本地 LLM「扮演」而非真远程调用。
+            "platform": (ag.platform or "").lower() if ag.platform else None,
+            "agent_endpoint": (ag.agent_endpoint or "").strip() if ag.agent_endpoint else None,
+            "agent_api_key": ag.agent_api_key or None,
         })
 
     return results
@@ -1329,7 +1588,43 @@ async def send_message(
     ai_model = None
     skills = []  # 初始化避免UnboundLocalError
     agents = []  # 初始化避免UnboundLocalError
+    media_generating = False  # 提前初始化：外部Agent分支与末尾均会引用
+    external_handled = False   # 是否已由外部Agent(Dify/n8n/Coze)远程处理
+
+    # ===== 外部 Agent 远程调用分支 (Dify / n8n / Coze) =====
+    # 若本轮对话关联了带有效 agent_endpoint 的外部 Agent，则直接远程调用其平台，
+    # 跳过本地 LLM「扮演」逻辑。effective_agent_ids 已在前面从请求/上下文继承。
+    external_agents = _resolve_market_agents(db, effective_agent_ids, user_id)
+    external_agent = next((a for a in external_agents if a.get("agent_endpoint")), None)
+    if external_agent:
+        agents = external_agents
+        # 组装最近历史(供 n8n 等无平台会话态的服务维持多轮上下文)
+        _hist_rows = db.query(Message).filter(
+            Message.conversation_id == conv_id
+        ).order_by(asc(Message.created_at)).limit(10).all()
+        _history = [{"role": h.role, "content": h.content}
+                    for h in _hist_rows if h.role in ("user", "assistant")]
+        try:
+            answer, new_ext_id = await _call_external_agent(
+                db, external_agent,
+                query=req.content + attachment_block,
+                user_id=user_id,
+                ext_conversation_id=conv.ext_agent_conversation_id,
+                history=_history,
+            )
+            ai_content = answer or "外部 Agent 未返回内容。"
+            if new_ext_id:
+                conv.ext_agent_conversation_id = new_ext_id
+        except Exception as e:
+            logger.warning(
+                f"[external_agent] 远程调用失败 platform={external_agent.get('platform')} "
+                f"endpoint={external_agent.get('agent_endpoint')}: {e}"
+            )
+            ai_content = f"⚠️ 外部 Agent 调用失败：{str(e)[:200]}"
+        external_handled = True
+
     # 解析模型：支持 Auto 模式（model_id='auto' 或未指定时，按能力自动匹配平台默认模型）
+    if not external_handled:
         if model_id and model_id != "auto":
             ai_model = db.query(AIModel).filter(AIModel.id == model_id).first()
         if not ai_model:
@@ -1514,7 +1809,7 @@ async def send_message(
                 ai_content = "该模型缺少API密钥或端点配置, 请在后台管理-大模型库中配置API Key和Endpoint。"
         else:
             ai_content = "请选择一个已配置的AI模型进行对话。"
-    else:
+    elif not external_handled:
         ai_content = "请先在后台管理-大模型库配置默认对话模型（或聊天中选择模型），Auto 模式下将自动使用平台默认模型。"
 
     ai_token_count = len(ai_content) // 2

backend/python/app/models.py

@@ -86,6 +86,7 @@ class Conversation(Base):
     knowledge_graph_count = Column(Integer, default=0)
     knowledge_graph_base_id = Column(String(64), default=None, comment="已生成的知识图谱知识库ID")
     source = Column(String(32), default=None, comment="对话来源，如 wechat_bot")
+    ext_agent_conversation_id = Column(String(128), default=None, comment="外部Agent(Dify/Coze)会话ID，用于维持多轮上下文")
     created_at = Column(DateTime, default=datetime.now)
     updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
 

migrations/016_add_conversation_ext_agent.sql

@@ -0,0 +1,14 @@
+-- 016: 为 conversations 增加 ext_agent_conversation_id 列
+-- 用途：外部 Agent(Dify/Coze) 远程调用时保存其平台侧 conversation_id，
+--       以维持多轮上下文。n8n(无平台会话态) 不使用该列。
+-- 幂等：列已存在则跳过。
+SET @col = (SELECT COUNT(*) FROM information_schema.columns
+            WHERE table_schema = DATABASE()
+              AND table_name = 'conversations'
+              AND column_name = 'ext_agent_conversation_id');
+SET @sql = IF(@col = 0,
+    'ALTER TABLE conversations ADD COLUMN ext_agent_conversation_id VARCHAR(128) NULL COMMENT \'外部Agent(Dify/Coze)会话ID，用于维持多轮上下文\'',
+    'SELECT 1');
+PREPARE stmt FROM @sql;
+EXECUTE stmt;
+DEALLOCATE PREPARE stmt;