From d274733fb39370cdd3edc8725f3c86db44359147 Mon Sep 17 00:00:00 2001 From: qihu-bot Date: Wed, 15 Jul 2026 13:55:04 +0800 Subject: [PATCH] =?UTF-8?q?feat(chat):=20=E5=A4=96=E9=83=A8Agent(Dify/n8n/?= =?UTF-8?q?Coze)=E8=BF=90=E8=A1=8C=E6=9C=9F=E7=9C=9F=E5=AE=9E=E8=BF=9C?= =?UTF-8?q?=E7=A8=8B=E8=B0=83=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _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 | 311 +++++++++++++++++- backend/python/app/models.py | 1 + migrations/016_add_conversation_ext_agent.sql | 14 + 3 files changed, 318 insertions(+), 8 deletions(-) create mode 100644 migrations/016_add_conversation_ext_agent.sql diff --git a/backend/python/app/api/chat.py b/backend/python/app/api/chat.py index a962cb8..40dd798 100644 --- a/backend/python/app/api/chat.py +++ b/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,14 +1588,50 @@ 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 model_id and model_id != "auto": - ai_model = db.query(AIModel).filter(AIModel.id == model_id).first() - if not ai_model: - # Auto / 未指定 → 按用户输入意图匹配对应能力的默认模型，否则回退文本对话 - intent = detect_media_intent(req.content) - resolved = resolve_auto_model(intent, db) if intent else None - ai_model = resolved or resolve_auto_model("text", db) + if not external_handled: + if model_id and model_id != "auto": + ai_model = db.query(AIModel).filter(AIModel.id == model_id).first() + if not ai_model: + # Auto / 未指定 → 按用户输入意图匹配对应能力的默认模型，否则回退文本对话 + intent = detect_media_intent(req.content) + resolved = resolve_auto_model(intent, db) if intent else None + ai_model = resolved or resolve_auto_model("text", db) if ai_model: # ====== 图像/视频生成模型分支 ====== @@ -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 diff --git a/backend/python/app/models.py b/backend/python/app/models.py index b0414d6..8fa005e 100644 --- a/backend/python/app/models.py +++ b/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) diff --git a/migrations/016_add_conversation_ext_agent.sql b/migrations/016_add_conversation_ext_agent.sql new file mode 100644 index 0000000..75d4f38 --- /dev/null +++ b/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;