diff --git a/backend/python/app/api/im_bot_user.py b/backend/python/app/api/im_bot_user.py index 43f7c45..9b18389 100644 --- a/backend/python/app/api/im_bot_user.py +++ b/backend/python/app/api/im_bot_user.py @@ -143,12 +143,33 @@ def _qihua_reply(text: str, user_id: str = None) -> str: f"{hist}\n\n" f"用户当前问题：{text}" ) + # 磁盘调试日志（不依赖 uvicorn logger） + try: + with open("/tmp/im_llm_debug.log", "a", encoding="utf-8") as _f: + _f.write(_beijing_now().isoformat() + f" _qihua_reply user={user_id} prompt_len={len(prompt)} prompt={repr(prompt)[:500]}\n") + except Exception: + pass db = None try: from app.api.im_llm import _call_real_llm db = SessionLocal() - return _call_real_llm(db, None, prompt, "个人微信", user_id=user_id) + result = _call_real_llm(db, None, prompt, "个人微信", user_id=user_id) + try: + with open("/tmp/im_llm_debug.log", "a", encoding="utf-8") as _f: + _f.write(_beijing_now().isoformat() + f" _qihua_reply RESULT len={len(result) if result else 0} preview={repr(result)[:200]}\n") + except Exception: + pass + return result except Exception as e: + import traceback as _tb2 + import logging + _log = logging.getLogger('im_bot_user') + _log.warning(f'[_qihua_reply] 异常: {e}\n{_tb2.format_exc()[:1000]}') + try: + with open("/tmp/im_llm_debug.log", "a", encoding="utf-8") as _f: + _f.write(_beijing_now().isoformat() + f" _qihua_reply EXCEPTION: {e}\n{_tb2.format_exc()[:800]}\n") + except Exception: + pass return f"（AI 回复生成失败：{e}）" finally: if db: @@ -164,7 +185,7 @@ def _wechat_get_updates(baseurl: str, bot_token: str, cursor: str): resp = requests.post( baseurl.rstrip("/") + "/ilink/bot/getupdates", headers=_wechat_headers(bot_token), - json={"get_updates_buf": cursor, "base_info": {"channel_version": "1.0.2"}}, + json={"get_updates_buf": cursor}, timeout=40, ) if not resp.ok: @@ -178,20 +199,26 @@ def _wechat_get_updates(baseurl: str, bot_token: str, cursor: str): def _wechat_post_msg(baseurl: str, bot_token: str, to_user_id: str, context_token: str, item_list: list): """向 iLink 发送一条消息（item_list 可含文本/图片/视频等）。返回 (ok, ret, body)。 ret=-14=会话过期需重扫；ret=-2=token 失效。""" - import uuid as _uuid payload = { "msg": { - "from_user_id": "", "to_user_id": to_user_id, - "client_id": "py-" + str(_uuid.uuid4()), "message_type": 2, "message_state": 2, "context_token": context_token, "item_list": item_list, - }, - "base_info": {"channel_version": "1.0.2"}, + } } last_body = {} + # 07-30 diagnostic + import json as _json3 + try: + with open("/tmp/wechat_send_debug.log", "a", encoding="utf-8") as _f: + _f.write(_beijing_now().isoformat() + " SEND_DEBUG token=" + repr(bot_token[:20] if bot_token else "NULL") + + " to=" + repr(to_user_id[:30]) + " ctx=" + repr(context_token[:30]) + "\n") + _f.write(" payload_keys=" + str(list(payload.get("msg", {}).keys())) + "\n") + except Exception: + pass + for _ in range(2): try: resp = requests.post( @@ -239,21 +266,21 @@ def _platform_relay_url(raw_url: str) -> str: def _wechat_send_media(baseurl: str, bot_token: str, to_user_id: str, context_token: str, media_type: str, url: str): - """发送媒体（图片/视频）并在微信内联渲染。 + """发送媒体（图片/视频）。 - 问题：AGNES 媒体 URL 位于境外，iLink(中国大陆服务器)直连常被墙/超时， - 微信里只能看到"链接"而无法内联播放。 - 解决：改写为平台中转地址(经香港代理可达 AGNES)，微信即可内联显示图片/视频。 + 问题：AGNES 媒体 URL 位于境外，iLink(中国大陆服务器)直连常被墙/超时。 + 解决：改写为平台中转地址(经香港代理可达 AGNES)。 + + 2026-07-31 兼容修复：iLink 对 type=2/5 的媒体 item 返回 ret=-2 invalid arguments， + 且旧代码把 ret=-2 误判为 token 失效，导致连接假性断开。先以纯文本链接形式发送， + 确保用户可点击访问且不会破坏会话；待拿到 iLink 媒体协议文档后再恢复内联 item。 """ relay_url = _platform_relay_url(url) if media_type == "image": - media_item = {"type": 2, "image_item": {"url": relay_url}} - caption = "🖼️ 图片已生成（若未直接显示，请点此查看）：" + relay_url + caption = "🖼️ 图片已生成，点击查看：" + relay_url else: - media_item = {"type": 5, "video_item": {"url": relay_url}} - caption = "🎬 视频已生成（若未直接播放，请点此查看）：" + relay_url - return _wechat_post_msg(baseurl, bot_token, to_user_id, context_token, - [media_item, {"type": 1, "text_item": {"text": caption}}]) + caption = "🎬 视频已生成，点击查看：" + relay_url + return _wechat_send_message(baseurl, bot_token, to_user_id, context_token, caption) def _wechat_extract_text(msg: dict) -> str: @@ -326,7 +353,6 @@ def _get_or_create_wechat_conv(db: Session, user_id: str) -> str: ), {"uid": user_id}) if rows: return _row_to_dict(rows[0])["id"] - import uuid as _uuid cid = str(_uuid.uuid4()) now = _beijing_now() _execute_raw(db, text( @@ -338,7 +364,6 @@ def _get_or_create_wechat_conv(db: Session, user_id: str) -> str: def _append_ai_message(db: Session, conv_id: str, role: str, content: str, created=None): - import uuid as _uuid _execute_raw(db, text( "INSERT INTO messages (id, conversation_id, role, content, token_used, created_at) " "VALUES (:id, :cid, :role, :content, 0, :c)" @@ -425,7 +450,14 @@ def _wechat_receiver_loop(account_id: int): except Exception: pass # 先生成回复（此时历史尚未包含本条，避免重复） + import logging as _lrx, inspect as _insp + _lrx.warning(f'[receiver] calling _qihua_reply({repr(content)}, {repr(owner_uid)})') + # 验证使用的 _call_real_llm 版本 + from app.api.im_llm import _call_real_llm as _crl + _src = _insp.getsource(_crl) + _lrx.warning(f'[receiver] _call_real_llm source len={len(_src)}, has _try_models={"_try_models" in _src}') reply = _qihua_reply(content, owner_uid) + _lrx.warning(f'[receiver] reply ({type(reply).__name__}): {repr(reply)[:200]}') # 媒体结构化返回：_call_real_llm 生成图片/视频时返回 # {"__media__": True, "media_type": "image"/"video", "url": "..."} if isinstance(reply, dict) and reply.get("__media__"): @@ -457,11 +489,23 @@ def _wechat_receiver_loop(account_id: int): % (ret, repr(sender))) except Exception: pass - if ret in (-2, -14): - # 会话/鉴权失效：标记需重扫但继续循环，让下一轮 poll 自然检测过期 + if ret == -14: + # -14=会话过期，需重扫 _mark_wechat_expired(d["id"]) + elif ret == -2: + # -2=参数错误（如媒体消息格式不被 iLink 接受），并非 token 失效。 + # 标记 expired 会导致连接假性断开，后续消息全部丢失，故仅记录。 + pass time.sleep(1) except Exception: + # 2026-07-31 修复：此前裸 except 静默吞掉所有异常，导致接收器出错时毫无日志、 + # 完全无法排查。改为记录异常堆栈到磁盘，便于定位。 + import traceback as _tb + try: + with open("/tmp/wechat_receiver_err.log", "a", encoding="utf-8") as _f: + _f.write(_beijing_now().isoformat() + " RECEIVER EXCEPTION:\n" + _tb.format_exc() + "\n") + except Exception: + pass time.sleep(5) finally: db.close() @@ -555,6 +599,13 @@ async def wechat_qrcode_status( data = j.get("data", j) if isinstance(j, dict) else j status = data.get("status") if status == "confirmed": + # 07-30 diagnostic: log full iLink confirm response + import json as _json + try: + with open("/tmp/wechat_confirm.log", "a", encoding="utf-8") as _f: + _f.write(_beijing_now().isoformat() + " CONFIRM_RAW=" + _json.dumps(j, ensure_ascii=False)[:2000] + "\n") + except Exception: + pass creds = data.get("credentials", {}) or {} bot_token = data.get("bot_token") or creds.get("bot_token") baseurl = data.get("baseurl") or creds.get("baseurl") or ILINK_BASE @@ -590,7 +641,15 @@ async def wechat_qrcode_status( "updated_at": _beijing_now(), "id": _row_to_dict(row)["id"], }) row = _wechat_row_for_user(db, user.id) - aid = _row_to_dict(row)["id"] if row else 1 + saved = _row_to_dict(row) if row else {} + import json as _json2 + try: + with open("/tmp/wechat_confirm.log", "a", encoding="utf-8") as _f: + _f.write(_beijing_now().isoformat() + " SAVED bot_token=" + repr(saved.get("bot_token", "NULL")[:20]) + + " ilink_bot_id=" + repr(saved.get("ilink_bot_id")) + "\n") + except Exception: + pass + aid = saved.get("id", 1) _start_wechat_receiver(aid) finally: db.close() diff --git a/backend/python/app/api/im_llm.py b/backend/python/app/api/im_llm.py index b819890..4027133 100644 --- a/backend/python/app/api/im_llm.py +++ b/backend/python/app/api/im_llm.py @@ -59,6 +59,17 @@ def _proxy_for_url(url: str) -> str | None: return os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY") or "http://172.18.0.1:7890" +_DEBUG_LOG_PATH = "/tmp/im_llm_debug.log" + +def _debug_log(msg: str): + """写磁盘调试日志（不依赖 uvicorn logging 配置）。""" + try: + with open(_DEBUG_LOG_PATH, "a", encoding="utf-8") as _f: + _f.write(datetime.now(timezone(timedelta(hours=8))).isoformat() + " " + msg + "\n") + except Exception: + pass + + def _run_async(coro, timeout=None): """在同步上下文中运行协程，兼容「无事件循环」与「已有事件循环」两种场景。 @@ -126,7 +137,15 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im _discover_installed_skills, _build_system_prompt, _execute_skill_tool, + _list_enabled_mcp_servers, ) + # 2026-08-04 双 Claw 对齐：能力发现收敛为唯一实现，Web / IM 共用 + from app.core.capability_discovery import ( + build_mcp_tool_defs, + discover_mcp_servers, + discover_skills, + ) + from app.mcp_client import call_mcp_tool # ===== 1) 媒体意图路由（Issue 7）===== # 2026-07-30 fix: 只对当前用户消息做媒体意图检测，避免历史中的图像请求污染 @@ -169,22 +188,58 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im ai_model = resolve_auto_model("text", db) _try_models = [ai_model] if (ai_model and ai_model.is_enabled) else [] # Auto 模式附加备选（IM Bot 始终 Auto） + # 2026-07-31 根因修复：仅选用「纯文本」模型作为 failover 备选。 + # 此前按 is_enabled 顺序取前 4 个，会混入图像/视频/语音(ASR)模型 + # （这些模型 capability 含 "text" 但本质是媒体模型，收到文本 prompt 返回空内容、 + # 不抛异常 → 被误判为「模型失败」，最终导致「已尝试N个模型均调用失败」）。 + _MEDIA_CAPS = {"image", "video", "audio", "asr", "embedding", "rerank", "ocr"} + def _is_text_only(m): + _caps = m.capabilities or [] + return "text" in _caps and not any(c in _MEDIA_CAPS for c in _caps) _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]) + ).order_by(AIModel.is_default.desc()).all() + _fallbacks = [m for m in _fallbacks if _is_text_only(m)] + _try_models.extend([m for m in _fallbacks[:4] if m not in _try_models]) + _debug_log(f"TRY_MODELS user={user_id} count={len(_try_models)} names={[m.name for m in _try_models]}") if not _try_models: return "（未配置可用文本模型，无法生成回复）" - # 技能自动发现（提前执行一次，不必每个候选模型重复做） + # 能力自动发现（提前执行一次，不必每个候选模型重复做） + # 2026-08-04 根因修复： + # 旧实现 `_discover_installed_skills(db, user_id)[:15]` 强依赖 + # app_installations 表，线上该表 0 行 → 97 个 published skill + # 在 Web / IM 双端同时发现不到；`[:15]` 截断的是一个空列表。 + # 现改走统一发现层（安装优先 → published 降级 → 相关性裁剪）。 + _query_hint = _cur_msg or prompt skills = [] - if user_id: - try: - skills = _discover_installed_skills(db, user_id)[:15] - except Exception: - skills = [] + try: + skills = discover_skills(db, user_id, query=_query_hint, claw="im") + except Exception as _e: + logger.warning(f"[im_llm] 技能发现失败(降级为空): {_e}") + skills = [] + + # MCP 自动挂载：IM Bot Claw 没有勾选 UI，自动发现即其等价通道。 + # （Web Claw 靠用户在界面勾选 → effective_mcp_ids） + mcp_servers = [] + mcp_function_definitions = [] + mcp_tool_map = {} + try: + mcp_servers = discover_mcp_servers(db, None, claw="im") + if mcp_servers: + mcp_function_definitions, mcp_tool_map = _run_async( + build_mcp_tool_defs(mcp_servers), timeout=45 + ) + except Exception as _e: + logger.warning(f"[im_llm] MCP 发现失败(降级为空): {_e}") + mcp_servers, mcp_function_definitions, mcp_tool_map = [], [], {} + + _debug_log( + f"CAP skills={len(skills)} mcp_servers={len(mcp_servers)} " + f"mcp_tools={len(mcp_function_definitions)}" + ) # 平台知识库注入（对齐 Web Chat——微信用户提问"平台有哪些技能"也能得到完整清单） kb_context = "" @@ -213,13 +268,18 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im _retry_log.append(f"{ai_model.name}: 缺少密钥或端点") continue + # agents 暂留空：线上 market_apps 中 type='agent' 记录为 0， + # 且 IM 侧尚无 agent 执行分支；待薄壳化接入 EngineCore 后自动获得。 system_prompt, function_definitions = _build_system_prompt( skills=skills, agents=[], kb_context=kb_context, user_content=prompt, - mcp=[], + mcp=mcp_servers, + mcp_functions=mcp_function_definitions, auto_discovered=bool(skills), + all_mcp=_list_enabled_mcp_servers(db), + db=db, ) messages = [ {"role": "system", "content": system_prompt}, @@ -233,7 +293,7 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im _http_ok = True for round_num in range(max_rounds): request_body = {"model": ai_model.name, "messages": messages, "max_tokens": 2048, "enable_thinking": False, "temperature": 0.7} - if function_definitions and skills: + if function_definitions: request_body["tools"] = [ {"type": "function", "function": fd} for fd in function_definitions ] @@ -241,6 +301,7 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im try: import logging as _l5 _l5.warning(f'[im_llm REQ] {ai_model.name}: url={api_url[:60]}, proxy={_proxy_for_url(api_url)}, api_key={api_key[:20]}..., messages_count={len(request_body.get("messages", []))}, tools_count={len(request_body.get("tools", []))}') + _debug_log(f"REQ model={ai_model.name} url={api_url} proxy={_proxy_for_url(api_url)} body={json.dumps(request_body, ensure_ascii=False)[:600]}") data = chat_complete( api_url, api_key, @@ -249,11 +310,13 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im timeout=ai_model.timeout or 120, ) _l5.warning(f'[im_llm RES] {ai_model.name}: data_keys={list(data.keys()) if data else None}, resp_type={type(data).__name__}') + _debug_log(f"RES model={ai_model.name} data={json.dumps(data, ensure_ascii=False)[:600]}") except Exception as _exc: _retry_log.append(f"{ai_model.name}: {str(_exc)[:60]}") import traceback as _tb logger.warning(f"[im_llm] 模型 {ai_model.name} 异常: {_exc}") logger.warning(f"[im_llm] {ai_model.name} 异常详情: {_tb.format_exc()[:2000]}") + _debug_log(f"EXC model={ai_model.name} err={str(_exc)[:200]} trace={_tb.format_exc()[:500]}") _http_ok = False break message = data.get("choices", [{}])[0].get("message", {}) @@ -265,10 +328,11 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im _msg_keys = list(message.keys()) if message else [] _tool_calls = tool_calls _lcr.warning(f'[CRITICAL] MODEL={ai_model.name} msg_keys={_msg_keys} content_len={len(_raw_content or "")} reasoning_len={len(_raw_reasoning or "")} has_tool_calls={bool(_tool_calls)}') + _debug_log(f"PARSE model={ai_model.name} content_len={len(_raw_content or '')} reasoning_len={len(_raw_reasoning or '')} content_preview={repr(_raw_content)[:120]}") ai_content = message.get("content", "") or message.get("reasoning_content") or "" if not ai_content and (message.get("reasoning_content") or ""): ai_content = f"（AI 思考：{message.get('reasoning_content')[:500]}...）" - if tool_calls and skills: + if tool_calls and (skills or mcp_tool_map): messages.append({ "role": "assistant", "content": message.get("content") or message.get("reasoning_content") or "", @@ -280,7 +344,40 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im func_args = json.loads(tc.get("function", {}).get("arguments", "{}")) except Exception: func_args = {"input": tc.get("function", {}).get("arguments", "")} - tool_result = _execute_skill_tool(db, func_name, func_args, skills, prompt) + if func_name.startswith("mcp__"): + # MCP 工具：命名规则与 Web Claw 完全一致 + # mcp__{server}__{tool}__{idx} + _info = mcp_tool_map.get(func_name) + if not _info: + tool_result = ( + "[MCP 工具未找到] 可用工具: " + + str(list(mcp_tool_map.keys())[:5]) + ) + else: + _srv = next( + (s for s in mcp_servers + if s.get("id") == _info.get("server_id")), + None, + ) + if not _srv: + tool_result = f"[MCP 服务未找到: {_info.get('server_name')}]" + else: + try: + tool_result = _run_async( + call_mcp_tool(_srv, _info["name"], func_args), + timeout=90, + ) + except Exception as _mexc: + tool_result = ( + f"[MCP 工具 {_info.get('name')} 执行失败：" + f"{str(_mexc)[:160]}]" + ) + _debug_log( + f"MCP_CALL name={func_name} " + f"result_len={len(tool_result or '')}" + ) + else: + tool_result = _execute_skill_tool(db, func_name, func_args, skills, prompt) messages.append({ "role": "tool", "tool_call_id": tc.get("id", "call_1"), @@ -294,20 +391,25 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im # 检查本轮调用是否成功 import logging as _log2 _log2.warning(f'[im_llm trace] {ai_model.name}: http_ok={_http_ok}, ai_content_len={len(ai_content) if ai_content else 0}, ai_content_preview={repr(ai_content)[:200]}') + _debug_log(f"CHECK model={ai_model.name} http_ok={_http_ok} ai_content_len={len(ai_content) if ai_content else 0}") if _http_ok and ai_content: + _debug_log(f"SUCCESS model={ai_model.name} returning len={len(ai_content)}") return ai_content or "（AI 回复为空）" # 本轮失败：记录日志，切换备选 logger.warning('[im_llm] 模型 %s 失败 (http_ok=%s, content=%s), 切换备选...' % ( ai_model.name, _http_ok, repr(ai_content)[:100] )) + _debug_log(f"FAIL model={ai_model.name} http_ok={_http_ok} content_preview={repr(ai_content)[:100]}") continue except Exception as exc: _retry_log.append(f"{ai_model.name}: {str(exc)[:60]}") logger.warning(f"[im_llm] 模型 {ai_model.name} 异常: {exc}") + _debug_log(f"OUTER_EXC model={ai_model.name} err={str(exc)[:120]}") continue # 全部失败 _summary = "; ".join(_retry_log[-3:]) + _debug_log(f"ALL_FAILED try_models={len(_try_models)} retry_log={_retry_log}") return f"（已尝试 {len(_try_models)} 个模型均调用失败。最后错误: {_summary}。请稍后重试。）" except Exception as e: return f"（AI 回复生成失败：{e}）" diff --git a/backend/python/app/core/capability_discovery.py b/backend/python/app/core/capability_discovery.py new file mode 100644 index 0000000..2de3939 --- /dev/null +++ b/backend/python/app/core/capability_discovery.py @@ -0,0 +1,277 @@ +"""统一能力发现层（双 Claw 共用 · 单一实现）。 + +背景 +---- +七虎平台对外有两个 Claw 端口： + * Web Claw —— SSE 流式，用户可在 UI 勾选 skills / mcp / agents + * IM Bot Claw —— 微信等 IM，server push 整条回复，**没有任何勾选 UI** + +历史问题（2026-08-04 线上实证）： + 1. `im_llm.py` 写死 `agents=[]` / `mcp=[]`，IM 端永远拿不到 MCP 工具； + 2. `_discover_installed_skills()` 强依赖 `app_installations` 表，而线上该表 + **0 行** → 97 个 published skill 在 Web / IM **双端同时发现不到**； + 3. IM 端 skills 还被 `[:15]` 截断（截断一个空列表，属噪声）。 + +设计原则 +-------- +能力发现只允许有 **一份实现**（本模块）。任何 Claw 想要"技能/工具"， +都必须调用这里的函数；禁止在端侧另写一套发现逻辑——历史上 #866 部署 +就是因为存在第二份实现而把 #764/#765 的注入改动整体覆盖回退。 + +发现策略（三级降级，双端一致） +-------------------------- + 1) 用户显式安装（app_installations）—— 最高优先，尊重用户意愿 + 2) 无安装记录 → 降级为平台已发布（status='published'）全集 + 3) 池子超过预算 → 按当前问题做轻量相关性排序，取 top-K 控制 token + +运维旋钮（环境变量，改完重启生效） +-------------------------------- + CAP_SKILL_LIMIT 单轮注入技能数上限，默认 12 + CAP_MCP_LIMIT 单轮注入 MCP 服务数上限，默认 8 + CAP_SKILL_FALLBACK_PUBLISHED 安装表为空时是否降级为 published，默认 开 + CAP_MCP_AUTO 未显式指定时是否自动挂载已启用 MCP，默认 开 +""" + +from __future__ import annotations + +import logging +import os +import re +from typing import Any, Iterable + +logger = logging.getLogger(__name__) + +__all__ = [ + "discover_skills", + "discover_mcp_servers", + "build_mcp_tool_defs", + "rank_by_query", +] + + +# -------------------------------------------------------------------------- +# 环境旋钮 +# -------------------------------------------------------------------------- +def _env_int(name: str, default: int) -> int: + try: + raw = (os.environ.get(name) or "").strip() + return int(raw) if raw else default + except Exception: + return default + + +def _env_flag(name: str, default: bool = True) -> bool: + raw = (os.environ.get(name) or "").strip().lower() + if not raw: + return default + return raw in ("1", "true", "yes", "on") + + +# -------------------------------------------------------------------------- +# 轻量相关性排序（中文友好：整词 + 2-gram） +# -------------------------------------------------------------------------- +def _tokens(text: str) -> set: + text = re.sub(r"[^\w\u4e00-\u9fff]+", " ", (text or "").lower()) + out: set = set() + for w in text.split(): + if len(w) >= 2: + out.add(w) + for i in range(len(w) - 1): + out.add(w[i : i + 2]) + return out + + +def rank_by_query(items: list, query: str, top_k: int) -> list: + """按与 query 的词元重叠度排序取前 top_k。 + + 池子本来就不超预算时原样返回，避免无谓抖动（同一问题两次调用结果稳定）。 + """ + if not items: + return [] + if top_k <= 0 or len(items) <= top_k: + return items[:top_k] if top_k > 0 else [] + q = _tokens(query) + if not q: + return items[:top_k] + + scored = [] + for idx, it in enumerate(items): + blob = " ".join( + str(it.get(k) or "") + for k in ("name", "description", "category", "func_name") + ) + t = _tokens(blob) + if t: + # 交集数按文档长度做平方根归一，抑制"长描述天然占优" + score = len(q & t) / (len(t) ** 0.5) + else: + score = 0.0 + # idx 作为稳定次序的 tie-breaker，保证同分时结果可复现 + scored.append((-score, idx, it)) + scored.sort(key=lambda x: (x[0], x[1])) + return [it for _, _, it in scored[:top_k]] + + +# -------------------------------------------------------------------------- +# 技能发现 +# -------------------------------------------------------------------------- +def discover_skills( + db, + user_id: str | None, + query: str = "", + limit: int | None = None, + claw: str = "web", +) -> list: + """发现当前请求可用的技能列表（双 Claw 共用）。 + + 返回 `_resolve_market_skills` 的结构： + [{id, name, description, skill_content, category, func_name, parameters}] + """ + limit = limit if limit is not None else _env_int("CAP_SKILL_LIMIT", 12) + if limit <= 0: + return [] + + # 延迟导入：避免 app.core ←→ app.api 循环依赖 + from app.api.chat_skill import _resolve_market_skills + from app.models import AppInstallation, MarketApp + + pool: list = [] + source = "empty" + + # 1) 用户显式安装 + if user_id: + try: + app_ids = [ + i.app_id + for i in db.query(AppInstallation) + .filter(AppInstallation.user_id == user_id) + .all() + ] + if app_ids: + ids = [ + a.id + for a in db.query(MarketApp) + .filter( + MarketApp.id.in_(app_ids), + MarketApp.type == "skill", + MarketApp.status == "published", + ) + .all() + ] + if ids: + pool = _resolve_market_skills(db, ids, user_id) + source = "installed" + except Exception as e: + logger.warning("[cap] 安装技能发现失败(降级继续): %s", e) + + # 2) 降级：平台已发布全集 + if not pool and _env_flag("CAP_SKILL_FALLBACK_PUBLISHED", True): + try: + ids = [ + a.id + for a in db.query(MarketApp) + .filter(MarketApp.type == "skill", MarketApp.status == "published") + .all() + ] + if ids: + pool = _resolve_market_skills(db, ids, user_id or "") + source = "published" + except Exception as e: + logger.warning("[cap] published 技能发现失败: %s", e) + + # 3) 预算裁剪 + picked = rank_by_query(pool, query, limit) + logger.info( + "[cap] skills claw=%s source=%s pool=%d picked=%d", + claw, + source, + len(pool), + len(picked), + ) + return picked + + +# -------------------------------------------------------------------------- +# MCP 发现 +# -------------------------------------------------------------------------- +def discover_mcp_servers( + db, + explicit_ids: Iterable[str] | None = None, + limit: int | None = None, + claw: str = "web", +) -> list: + """发现当前请求可用的 MCP 服务连接（双 Claw 共用）。 + + Web Claw 传 explicit_ids（用户在 UI 勾选的）； + IM Bot Claw 无勾选通道 → 传 None，自动挂载平台已启用的 MCP。 + 返回 [{id, name, description, transport, endpoint, url, headers}] + """ + limit = limit if limit is not None else _env_int("CAP_MCP_LIMIT", 8) + if limit <= 0: + return [] + + from app.api.chat_mcp import _list_enabled_mcp_servers, _resolve_active_mcp_servers + + ids = [i for i in (explicit_ids or []) if i] + if ids: + out = _resolve_active_mcp_servers(db, ids)[:limit] + logger.info("[cap] mcp claw=%s source=explicit picked=%d", claw, len(out)) + return out + + if not _env_flag("CAP_MCP_AUTO", True): + return [] + + out = _list_enabled_mcp_servers(db)[:limit] + logger.info("[cap] mcp claw=%s source=auto picked=%d", claw, len(out)) + return out + + +# -------------------------------------------------------------------------- +# MCP → function definitions +# -------------------------------------------------------------------------- +def _safe_name(raw: str, maxlen: int) -> str: + return re.sub(r"[^A-Za-z0-9_-]", "_", (raw or "")[:maxlen]) + + +async def build_mcp_tool_defs(servers: list, idx_start: int = 0) -> tuple: + """把 MCP 服务列表展开为 (function_definitions, tool_map)。 + + 命名规则与 Web Claw 现有实现严格一致：`mcp__{server}__{tool}__{idx}`， + 保证两端产生的工具名可互相对照、可被同一套审批/统计逻辑识别。 + + 返回: + function_definitions: [{name, description, parameters}] + tool_map: {func_name: {server_id, name, server_name}} + """ + from app.mcp_client import list_mcp_tools + + defs: list = [] + tool_map: dict = {} + + for offset, srv in enumerate(servers or []): + idx = idx_start + offset + try: + tools = await list_mcp_tools(srv) + except Exception as e: + logger.warning("[cap] MCP 拉取工具失败 [%s]: %s", srv.get("name"), e) + continue + for t in tools or []: + sname = _safe_name(srv.get("name") or "unknown", 30) + tname = _safe_name(t.get("name") or "tool", 50) + fname = f"mcp__{sname}__{tname}__{idx}" + tool_map[fname] = { + "server_id": srv.get("id"), + "name": t.get("name"), + "server_name": srv.get("name"), + } + defs.append( + { + "name": fname, + "description": ( + f"[MCP:{srv.get('name')}] {t.get('description') or t.get('name') or ''}" + )[:900], + "parameters": t.get("inputSchema") + or {"type": "object", "properties": {}}, + } + ) + return defs, tool_map diff --git a/backend/python/app/core/llm_client.py b/backend/python/app/core/llm_client.py index a06c6d6..0b8ff7e 100644 --- a/backend/python/app/core/llm_client.py +++ b/backend/python/app/core/llm_client.py @@ -83,6 +83,11 @@ def _create_kwargs(request_body: dict, stream: bool = False) -> dict: if tools: kw["tools"] = tools kw["tool_choice"] = request_body.get("tool_choice", "auto") + # 2026-07-30: 将未在白名单的额外字段（如 enable_thinking）作为 extra_body 传递 + _whitelist = {"model", "messages", "stream", "temperature", "top_p", "max_tokens", "stop", "tools", "tool_choice"} + _extras = {k: v for k, v in request_body.items() if k not in _whitelist} + if _extras: + kw["extra_body"] = _extras return kw @@ -223,29 +228,85 @@ def build_openai_client_sync(api_url, api_key, *, proxy=None, timeout=120, auth_ return OpenAI(base_url=base, api_key=api_key or "EMPTY", timeout=timeout, http_client=http_client, default_headers=_hdrs if _hdrs else None) -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, auth_header=auth_header) - 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), - } +def chat_complete(api_url, api_key, request_body, *, proxy=None, timeout=120, auth_header=None): + """同步版（与 achat_complete 行为/返回形状完全一致），供非 async 上下文使用。 + + 2026-07-30 重写：直接使用 requests 替代 OpenAI SDK/httpx。 + OpenAI SDK 的 httpx 连接池在慢长轮询（如 iLink QR 状态 12-19s）后会出现 + Connection error，已实测复现。requests 库没有这个问题。 + """ + import requests + import base64 + import random + body = dict(request_body) + body["stream"] = False + # enable_thinking 等额外字段已在 request_body 中 + + # Headers: Bearer token + X-WECHAT-UIN (兼容 iLink) + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {api_key}" if api_key else "Bearer empty", + } + if auth_header and ":" in auth_header: + _k, _v = auth_header.split(":", 1) + headers[_k.strip()] = _v.strip() + else: + # X-WECHAT-UIN 防重放 + headers["X-WECHAT-UIN"] = base64.b64encode(str(random.randint(0, 2**32-1)).encode()).decode() + + # 2026-07-31 修复根因: proxy 为 None 时传 proxies={} 而非 None。 + # requests 在 proxies=None 时会继续读取环境 HTTP_PROXY(trust_env 默认 True), + # 而 uvicorn 进程(run.sh)带有 HTTP_PROXY=172.18.0.1:7890, 该代理无法解析 + # Docker 内部域名(如 qihu-aiflow-litellm), 导致所有内部模型调用静默失败、返回空。 + # 显式 proxies={} + trust_env=False 可完全绕过环境代理, 使 _proxy_for_url 的 + # 决策(返回 None=不代理)真正生效。 + _proxies = {"http": proxy, "https": proxy} if proxy else {} + _session = requests.Session() + _session.trust_env = False + last_body = {} + for attempt in range(2): + try: + resp = _session.post( + api_url.rstrip("/"), + headers=headers, + json=body, + proxies=_proxies, + timeout=timeout, + ) + try: + last_body = resp.json() if resp.content else {} + except Exception: + last_body = {} + if resp.ok: + import logging as _l6 + _l6.warning(f'[chat_complete] body keys={list(last_body.keys()) if isinstance(last_body, dict) else type(last_body)}, content_len={len(resp.content)}, content_preview={str(resp.content)[:200]}') + return _shape_response(last_body) + except Exception: + pass + return {"choices": [{"message": {"content": "", "tool_calls": None, "reasoning_content": None}}]} + + +def _shape_response(body: dict) -> dict: + """Normalize OpenAI-compatible response to standard shape.""" + choices = body.get("choices", []) + import logging as _l7 + if choices: + msg = choices[0].get("message", {}) + result = { + "choices": [{ + "message": { + "content": msg.get("content") or "", + "tool_calls": _normalize_tool_calls(msg.get("tool_calls")), + "reasoning_content": msg.get("reasoning_content") or "", } - ], - "usage": (resp.usage.model_dump() if resp.usage else {}), + }] } + _l7.warning(f'[_shape] content_len={len(result["choices"][0]["message"]["content"])}, keys={list(result["choices"][0]["message"].keys())}') + return result + _l7.warning(f'[_shape] NO choices in body, keys={list(body.keys())[:5]}') + return body -# --------------------------------------------------------------------------- -# 嵌入（Embeddings）：OpenAI 兼容 /v1/embeddings，收敛散落的裸 httpx 嵌入调用 -# --------------------------------------------------------------------------- def _embed_kwargs(input, model): return {"model": model, "input": input if isinstance(input, list) else [input]} diff --git a/backend/python/app/mcp_client.py b/backend/python/app/mcp_client.py index 574b038..44c5fe5 100644 --- a/backend/python/app/mcp_client.py +++ b/backend/python/app/mcp_client.py @@ -20,6 +20,7 @@ from typing import Optional from mcp import ClientSession from mcp.client.sse import sse_client from mcp.client.streamable_http import streamablehttp_client +import httpx logger = logging.getLogger(__name__) @@ -28,18 +29,38 @@ _TOOLS_CACHE: dict = {} _TOOLS_CACHE_TTL = 120 # 秒 +def _no_proxy_httpx_client(headers=None, timeout=None, auth=None): + """MCP 传输用 httpx 客户端工厂：强制直连、不走环境代理。 + + 容器 run.sh 注入了 HTTP_PROXY=http://172.18.0.1:7890，该代理无法解析 + 外部 MCP 域名（xiaobenyang.com / anysearch.com 等），且默认 trust_env + 会读取 HTTP_PROXY 导致 tools/list 全部静默失败。这里 trust_env=False + 直连，与 app/core/llm_client.py:chat_complete 修复同源。 + """ + kwargs: dict = {"follow_redirects": True, "trust_env": False} + kwargs["timeout"] = timeout if timeout is not None else httpx.Timeout(30.0, read=300.0) + if headers is not None: + kwargs["headers"] = headers + if auth is not None: + kwargs["auth"] = auth + return httpx.AsyncClient(**kwargs) + + def _transport_cm(server: dict, timeout: float): """按 server['transport_type'] 选择官方 SDK 传输层。 返回异步上下文管理器，进入后产出 (read_stream, write_stream, ...)。 + 关键：注入 _no_proxy_httpx_client，使 MCP 流量直连（见其 docstring）。 """ url = (server.get("url") or "").strip() ttype = (server.get("transport_type") or "streamable_http").lower() headers = server.get("headers") or {} if ttype == "sse": - return sse_client(url, headers=headers, timeout=timeout) + return sse_client(url, headers=headers, timeout=timeout, + httpx_client_factory=_no_proxy_httpx_client) # 默认 streamable_http（官方 SDK 标准传输） - return streamablehttp_client(url, headers=headers, timeout=timeout) + return streamablehttp_client(url, headers=headers, timeout=timeout, + httpx_client_factory=_no_proxy_httpx_client) async def _session_request(