From 9d67b0544c391a13270c0360de50df5ca6d8b5d2 Mon Sep 17 00:00:00 2001 From: QihuAI Design Date: Thu, 23 Jul 2026 19:43:56 +0800 Subject: [PATCH] chore(align): sync Git with running container as authoritative source One-time alignment to close drift between the running container (qihu-aiflow / image qihuaiflow:with-deps) and this Git repository. Container hotfixes previously persisted only via `docker commit` are now captured in Git so Git becomes the single source of truth. Backend: - chat_service.py: accumulated fixes incl. EigenFlux README guidance - chat_context.py / chat_media.py: URL/web-repo context injection - api/url_reader.py: NEW SSRF-safe web/README reader (Issue #725) - skills_admin.py & Skills.tsx: 14-scene capability unification - pgvector_search.py, orchestrator.py, platform_knowledge.py, market.py, admin/models.py: drift sync Frontend: - ChatPage.tsx: collapsible AI think/act block (Issue #724) - MarketPage / EvolutionDashboard / TokenDashboard styling sync Migrations: - 018_normalize_marketapp_scene.sql: fixed capability allowlist (prevents capability wipe on restart) Alignment performed 2026-07-23 19:43:56. No deletions; secrets/venv/node_modules/ caches were not touched (tracked-source-only copy). --- backend/python/app/api/chat_service.py | 1369 ++++++++++++------------ 1 file changed, 714 insertions(+), 655 deletions(-) diff --git a/backend/python/app/api/chat_service.py b/backend/python/app/api/chat_service.py index de3ed79..619e1cf 100644 --- a/backend/python/app/api/chat_service.py +++ b/backend/python/app/api/chat_service.py @@ -1366,132 +1366,145 @@ async def _send_message_core( - if ai_model: + # ===== Model Failover: Auto 模式下自动切换备选模型 ===== + _try_models = [ai_model] if (ai_model and ai_model.is_enabled) else [] + if not model_id or model_id == "auto": + # Auto 模式：附加其他已启用的文本模型作为备选 + _fallbacks = db.query(AIModel).filter( + AIModel.is_enabled == True, + AIModel.id != (ai_model.id if ai_model else "") + ).order_by(AIModel.is_default.desc()).limit(5).all() + _try_models.extend([m for m in _fallbacks if m not in _try_models]) + _retry_log = [] + _all_failed = False + for _try_model in _try_models: + ai_model = _try_model + if ai_model: - # ====== 图像/视频生成模型分支 ====== + # ====== 图像/视频生成模型分支 ====== - caps = [c.lower() for c in (ai_model.capabilities or [])] if ai_model else [] + caps = [c.lower() for c in (ai_model.capabilities or [])] if ai_model else [] - media_generating = False + media_generating = False - if ai_model and ("image" in caps or "video" in caps): + if ai_model and ("image" in caps or "video" in caps): - # 异步生成：立即返回占位内容，后台任务轮询 AGNES 并回写， + # 异步生成：立即返回占位内容，后台任务轮询 AGNES 并回写， - # 避免最长 5 分钟的视频轮询阻塞 HTTP 请求导致客户端超时断开。 + # 避免最长 5 分钟的视频轮询阻塞 HTTP 请求导致客户端超时断开。 - media_generating = True + media_generating = True - if "video" in caps: + if "video" in caps: - ai_content = "⏳ 视频生成中…（通常 1~3 分钟，完成后自动刷新）" + ai_content = "⏳ 视频生成中…（通常 1~3 分钟，完成后自动刷新）" - else: + else: - ai_content = "⏳ 图像生成中…（完成后自动刷新）" + ai_content = "⏳ 图像生成中…（完成后自动刷新）" - elif ai_model and ai_model.endpoint: + elif ai_model and ai_model.endpoint: - provider = db.query(ModelProvider).filter(ModelProvider.id == ai_model.provider_id).first() + provider = db.query(ModelProvider).filter(ModelProvider.id == ai_model.provider_id).first() - api_key = ai_model.api_key or (provider.api_key if provider else None) + api_key = ai_model.api_key or (provider.api_key if provider else None) - api_url = ai_model.endpoint + api_url = ai_model.endpoint @@ -1502,20 +1515,20 @@ async def _send_message_core( - if not api_url and provider and provider.api_base_url: + if not api_url and provider and provider.api_base_url: - api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions" + api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions" @@ -1526,101 +1539,80 @@ async def _send_message_core( - if api_key and api_url: + if api_key and api_url: - try: + try: - # Stage A+B: 解析Skills和Agents + # Stage A+B: 解析Skills和Agents - skills = _resolve_market_skills(db, effective_skill_ids, user_id) + skills = _resolve_market_skills(db, effective_skill_ids, user_id) - agents = _resolve_market_agents(db, effective_agent_ids, user_id) + agents = _resolve_market_agents(db, effective_agent_ids, user_id) - mcps = _resolve_active_mcp_servers(db, effective_mcp_ids) + mcps = _resolve_active_mcp_servers(db, effective_mcp_ids) - # MCP 工具调用链路：拉取各 MCP 工具并转为 function definitions - mcp_servers_conn = mcps + # MCP 工具调用链路：拉取各 MCP 工具并转为 function definitions - mcp_function_definitions = [] + mcp_servers_conn = mcps - mcp_tool_map = {} + mcp_function_definitions = [] - mcp_idx = 0 + mcp_tool_map = {} - for _srv in mcp_servers_conn: + mcp_idx = 0 - try: - - async with span("mcp.list_tools", server=_srv.get("name") if isinstance(_srv, dict) else None): - - _tools = await list_mcp_tools(_srv) - - - - except Exception as _e: - - - - logger.warning(f"MCP 拉取工具失败 [{_srv.get('name')}]: {_e}") - - - - _tools = [] - - - - if _tools: + for _srv in mcp_servers_conn: @@ -1628,123 +1620,128 @@ async def _send_message_core( - _upd = db.query(MCPServer).filter(MCPServer.id == _srv["id"]).first() + async with span("mcp.list_tools", server=_srv.get("name") if isinstance(_srv, dict) else None): + _tools = await list_mcp_tools(_srv) - if _upd: + except Exception as _e: - _upd.tools_count = len(_tools) + logger.warning(f"MCP 拉取工具失败 [{_srv.get('name')}]: {_e}") - db.commit() + _tools = [] - except Exception: + if _tools: - pass + try: - for _t in _tools: + _upd = db.query(MCPServer).filter(MCPServer.id == _srv["id"]).first() - _fname = f"mcp_tool_{mcp_idx}" + if _upd: - mcp_tool_map[_fname] = { + _upd.tools_count = len(_tools) - "server_id": _srv["id"], + db.commit() - "name": _t.get("name"), + except Exception: - "server_name": _srv["name"], + pass - } + for _t in _tools: - _schema = _t.get("inputSchema") or {"type": "object", "properties": {}} + _fname = f"mcp_tool_{mcp_idx}" - if not isinstance(_schema, dict) or _schema.get("type") != "object": + mcp_tool_map[_fname] = { - _schema = {"type": "object", "properties": {}} + "server_id": _srv["id"], - mcp_function_definitions.append({ + "name": _t.get("name"), - "name": _fname, + "server_name": _srv["name"], - "description": f"[MCP·{_srv['name']}] {_t.get('description', '')}", + } - "parameters": _schema, + _schema = _t.get("inputSchema") or {"type": "object", "properties": {}} - }) + if not isinstance(_schema, dict) or _schema.get("type") != "object": - mcp_idx += 1 + _schema = {"type": "object", "properties": {}} - if mcp_function_definitions: + mcp_function_definitions.append({ - effective_use_tools = True + "name": _fname, + "description": f"[MCP·{_srv['name']}] {_t.get('description', '')}", + "parameters": _schema, + }) + mcp_idx += 1 - # Issue 8: 默认技能自动发现（仅当用户未显式选择技能时启用） + if mcp_function_definitions: + effective_use_tools = True - # 每轮先寻找本账号已安装且平台已部署的技能，多个适用时由模型自行判断选 1 个最优。 @@ -1752,79 +1749,268 @@ async def _send_message_core( - used_auto_skills = False + # Issue 8: 默认技能自动发现（仅当用户未显式选择技能时启用） - if not effective_skill_ids: + # 每轮先寻找本账号已安装且平台已部署的技能，多个适用时由模型自行判断选 1 个最优。 - try: + used_auto_skills = False - discovered = _discover_installed_skills(db, user_id) + if not effective_skill_ids: - except Exception as e: + try: - logger.warning(f"自动发现已安装技能失败: {e}") + discovered = _discover_installed_skills(db, user_id) - discovered = [] + except Exception as e: - if discovered: + logger.warning(f"自动发现已安装技能失败: {e}") - skills = discovered + discovered = [] - used_auto_skills = True + + + + + + if discovered: + + + + + + + + skills = discovered + + + + + + + + used_auto_skills = True + + + + + + + + effective_use_tools = True + + + + + + + + + + + + + + + + # Stage B.1: Agent可能自带skill_ids和kb_ids + + + + + + + + agent_extra_skill_ids = [] + + + + + + + + agent_extra_kb_ids = [] + + + + + + + + agent_model_override = None + + + + + + + + for agent in agents: + + + + + + + + if agent.get("skill_ids"): + + + + + + + + agent_extra_skill_ids.extend(agent["skill_ids"]) + + + + + + + + if agent.get("knowledge_base_ids"): + + + + + + + + agent_extra_kb_ids.extend(agent["knowledge_base_ids"]) + + + + + + + + if agent.get("model_id"): + + + + + + + + agent_model_override = agent["model_id"] + + + + + + + + + + + + + + + + # 合并Agent自带的skills和KBs + + + + + + + + all_skill_ids = list(set(effective_skill_ids + agent_extra_skill_ids)) + + + + + + + + all_kb_ids = list(set(effective_kb_ids + agent_extra_kb_ids)) + + + + + + + + + + + + + + + + if agent_extra_skill_ids or agent_extra_kb_ids: + + + + + + + + # 重新解析合并后的skills + + + + + + + + skills = _resolve_market_skills(db, all_skill_ids, user_id) @@ -1848,7 +2034,7 @@ async def _send_message_core( - # Stage B.1: Agent可能自带skill_ids和kb_ids + # Stage C: 增强版KB检索 @@ -1856,7 +2042,7 @@ async def _send_message_core( - agent_extra_skill_ids = [] + kb_context = await _retrieve_kb_context(db, all_kb_ids, req.content) @@ -1864,7 +2050,6 @@ async def _send_message_core( - agent_extra_kb_ids = [] @@ -1872,66 +2057,70 @@ async def _send_message_core( - agent_model_override = None + # Stage B: 动态组装system prompt + function definitions - for agent in agents: + system_prompt, function_definitions = _build_system_prompt( - if agent.get("skill_ids"): + skills=skills, - agent_extra_skill_ids.extend(agent["skill_ids"]) + agents=agents, - if agent.get("knowledge_base_ids"): + kb_context=kb_context, - agent_extra_kb_ids.extend(agent["knowledge_base_ids"]) + user_content=req.content, - if agent.get("model_id"): + mcp=mcps, + mcp_functions=mcp_function_definitions, - agent_model_override = agent["model_id"] + auto_discovered=used_auto_skills, + all_mcp=_list_enabled_mcp_servers(db), + db=db, + ) @@ -1944,24 +2133,26 @@ async def _send_message_core( - # 合并Agent自带的skills和KBs + # 构建消息列表 - all_skill_ids = list(set(effective_skill_ids + agent_extra_skill_ids)) + # 实际调用追踪（Issue：反馈真实使用的技能/MCP，而非注入的全部） + called_skill_names = [] + called_mcp_names = [] - all_kb_ids = list(set(effective_kb_ids + agent_extra_kb_ids)) + messages = [{"role": "system", "content": system_prompt}] @@ -1976,15 +2167,15 @@ async def _send_message_core( - if agent_extra_skill_ids or agent_extra_kb_ids: + # 分层上下文引擎：预算感知历史 + 压缩摘要(若有) + messages.extend(_build_chat_history(db, conv_id, max_tokens=6000, exclude_latest_user=True)) - # 重新解析合并后的skills @@ -1992,18 +2183,18 @@ async def _send_message_core( - skills = _resolve_market_skills(db, all_skill_ids, user_id) + # Issue 1: 必须把附件内容(attachment_block)一并发给模型， - effective_use_tools = True + # 否则用户在对话窗上传文件后模型“看不到”文件内容。 @@ -2011,12 +2202,36 @@ async def _send_message_core( + # Issue 2 修复：用户输入含 URL 且要求读取/加入/安装时，服务端抓取真实页面内容注入上下文 + try: + from app.api.url_reader import fetch_url_context + _url_ctx = await fetch_url_context(req.content) + except Exception as _e: + import logging as _logging + _logging.getLogger(__name__).warning("URL 上下文抓取失败(已跳过): %s", _e) + _url_ctx = None + if _url_ctx and _url_ctx.get("text"): + messages.append({ + "role": "system", + "content": ( + "【用户随消息提供的参考链接的真实内容】\n" + f"链接地址: {_url_ctx.get('url')}\n" + "下面是该链接页面的真实内容，请严格基于它来回答用户的需求" + "（例如按其『Quick Start / 加入步骤』逐步引导用户完成安装或加入），" + "不要凭空编造步骤：\n\n" + _url_ctx["text"] + ), + }) + process_steps.append({ + "type": "skill", + "name": "网页/仓库读取", + "result": f"已读取 {_url_ctx.get('url')}（{len(_url_ctx['text'])} 字符）", + }) + messages.append({"role": "user", "content": req.content + attachment_block}) - # Stage C: 增强版KB检索 @@ -2024,10 +2239,10 @@ async def _send_message_core( - kb_context = await _retrieve_kb_context(db, all_kb_ids, req.content) + # Stage B.3: Function Calling循环(最多5轮) @@ -2035,78 +2250,79 @@ async def _send_message_core( + max_tool_rounds = 5 - # Stage B: 动态组装system prompt + function definitions + # ===== 3.1 tool_policy 沙箱：聚合策略 + 初始化预算计数器 ===== + _agent_tool_policy = _resolve_agent_tool_policy(agents) + _tool_call_count = 0 + for round_num in range(max_tool_rounds): + _round_tool_count = 0 - system_prompt, function_definitions = _build_system_prompt( + request_body = { - skills=skills, + "model": ai_model.name, - agents=agents, + "messages": messages, - kb_context=kb_context, + "max_tokens": 2048, - user_content=req.content, + } - mcp=mcps, - mcp_functions=mcp_function_definitions, - auto_discovered=used_auto_skills, + if function_definitions and effective_use_tools: - all_mcp=_list_enabled_mcp_servers(db), - db=db, - ) + request_body["tools"] = [ @@ -2114,32 +2330,31 @@ async def _send_message_core( + {"type": "function", "function": fd} - # 构建消息列表 + for fd in function_definitions - # 实际调用追踪（Issue：反馈真实使用的技能/MCP，而非注入的全部） - called_skill_names = [] - called_mcp_names = [] + ] - messages = [{"role": "system", "content": system_prompt}] + request_body["tool_choice"] = "auto" @@ -2150,13 +2365,12 @@ async def _send_message_core( - # 分层上下文引擎：预算感知历史 + 压缩摘要(若有) - messages.extend(_build_chat_history(db, conv_id, max_tokens=6000, exclude_latest_user=True)) + async with httpx.AsyncClient( @@ -2164,54 +2378,31 @@ async def _send_message_core( + timeout=ai_model.timeout or 120, - # Issue 1: 必须把附件内容(attachment_block)一并发给模型， + proxy=_proxy_for_url(api_url), - # 否则用户在对话窗上传文件后模型“看不到”文件内容。 + trust_env=False, - # Issue 2 修复：用户输入含 URL 且要求读取/加入/安装时，服务端抓取真实页面内容注入上下文 - try: - from app.api.url_reader import fetch_url_context - _url_ctx = await fetch_url_context(req.content) - except Exception as _e: - import logging as _logging - _logging.getLogger(__name__).warning("URL 上下文抓取失败(已跳过): %s", _e) - _url_ctx = None - if _url_ctx and _url_ctx.get("text"): - messages.append({ - "role": "system", - "content": ( - "【用户随消息提供的参考链接的真实内容】\n" - f"链接地址: {_url_ctx.get('url')}\n" - "下面是该链接页面的真实内容，请严格基于它来回答用户的需求" - "（例如按其『Quick Start / 加入步骤』逐步引导用户完成安装或加入），" - "不要凭空编造步骤：\n\n" + _url_ctx["text"] - ), - }) - process_steps.append({ - "type": "skill", - "name": "网页/仓库读取", - "result": f"已读取 {_url_ctx.get('url')}（{len(_url_ctx['text'])} 字符）", - }) - messages.append({"role": "user", "content": req.content + attachment_block}) + ) as client: @@ -2219,100 +2410,98 @@ async def _send_message_core( + async with span("llm.completion", model=ai_model.id if ai_model else None): + resp = await client.post( - # Stage B.3: Function Calling循环(最多5轮) + api_url, - max_tool_rounds = 5 + headers={ - # ===== 3.1 tool_policy 沙箱：聚合策略 + 初始化预算计数器 ===== - _agent_tool_policy = _resolve_agent_tool_policy(agents) - _tool_call_count = 0 - for round_num in range(max_tool_rounds): + "Authorization": f"Bearer {api_key}", - _round_tool_count = 0 + "Content-Type": "application/json" - request_body = { + }, - "model": ai_model.name, + json=request_body - "messages": messages, + ) - "max_tokens": 2048, + if resp.status_code != 200: - } + if round_num == 0: - if function_definitions and effective_use_tools: + ai_content = f"AI模型调用失败 (HTTP {resp.status_code}), 请检查模型配置。" - request_body["tools"] = [ + break - {"type": "function", "function": fd} @@ -2320,7 +2509,6 @@ async def _send_message_core( - for fd in function_definitions @@ -2328,7 +2516,7 @@ async def _send_message_core( - ] + data = resp.json() @@ -2336,7 +2524,7 @@ async def _send_message_core( - request_body["tool_choice"] = "auto" + choice = data.get("choices", [{}])[0] @@ -2344,6 +2532,7 @@ async def _send_message_core( + message = choice.get("message", {}) @@ -2352,7 +2541,6 @@ async def _send_message_core( - async with httpx.AsyncClient( @@ -2360,7 +2548,7 @@ async def _send_message_core( - timeout=ai_model.timeout or 120, + # 记录token使用量 @@ -2368,7 +2556,7 @@ async def _send_message_core( - proxy=_proxy_for_url(api_url), + usage = data.get("usage", {}) @@ -2376,7 +2564,7 @@ async def _send_message_core( - trust_env=False, + prompt_tokens += usage.get("prompt_tokens", 0) @@ -2384,7 +2572,7 @@ async def _send_message_core( - ) as client: + completion_tokens += usage.get("completion_tokens", 0) @@ -2392,9 +2580,8 @@ async def _send_message_core( - async with span("llm.completion", model=ai_model.id if ai_model else None): + total_tokens += usage.get("total_tokens", 0) - resp = await client.post( @@ -2402,7 +2589,6 @@ async def _send_message_core( - api_url, @@ -2410,7 +2596,7 @@ async def _send_message_core( - headers={ + # 检查是否有tool_calls @@ -2418,7 +2604,14 @@ async def _send_message_core( - "Authorization": f"Bearer {api_key}", + tool_calls = message.get("tool_calls", []) + # 过程步骤：捕获模型思考/推理(产生工具调用前的推理) + _think = (message.get("content") or "").strip() + if _think and tool_calls: + process_steps.append({ + "type": "thinking", + "text": _think, + }) @@ -2426,458 +2619,16 @@ async def _send_message_core( - "Content-Type": "application/json" + if tool_calls and effective_use_tools: - }, - - - - - - - - json=request_body - - - - - - - - ) - - - - - - - - if resp.status_code != 200: - - - - - - - - if round_num == 0: - - - - - - - - ai_content = f"AI模型调用失败 (HTTP {resp.status_code}), 请检查模型配置。" - - - - - - - - break - - - - - - - - - - - - - - - - data = resp.json() - - - - - - - - choice = data.get("choices", [{}])[0] - - - - - - - - message = choice.get("message", {}) - - - - - - - - - - - - - - - - # 记录token使用量 - - - - - - - - usage = data.get("usage", {}) - - - - - - - - prompt_tokens += usage.get("prompt_tokens", 0) - - - - - - - - completion_tokens += usage.get("completion_tokens", 0) - - - - - - - - total_tokens += usage.get("total_tokens", 0) - - - - - - - - - - - - - - - - # 检查是否有tool_calls - - - - - - - - tool_calls = message.get("tool_calls", []) - # 过程步骤：捕获模型思考/推理(产生工具调用前的推理) - _think = (message.get("content") or "").strip() - if _think and tool_calls: - process_steps.append({ - "type": "thinking", - "text": _think, - }) - - - - - - - - - if tool_calls and effective_use_tools: - - - - - - - - # 将assistant消息(含tool_calls)加入上下文 - - - - - - - - messages.append({ - - - - - - - - "role": "assistant", - - - - - - - - "content": message.get("content") or "", - - - - - - - - "tool_calls": tool_calls, - - - - - - - - }) - - - - - - - - - - - - - - - - # Stage A.2: 真实Skill执行 - - - - - - - - for tc in tool_calls: - - - - - - - - func_name = tc.get("function", {}).get("name", "") - - # ===== 3.1 tool_policy 沙箱：逐工具校验（白名单 + 预算 + 每轮上限） ===== - - _mcp_info = mcp_tool_map.get(func_name) if func_name.startswith("mcp_tool_") else None - - _policy_check = _check_tool_policy(func_name, _mcp_info, skills, _agent_tool_policy, _tool_call_count, _round_tool_count) - - if not _policy_check["allowed"]: - - _reason = _policy_check["reason"] - - logger.warning( - - f"[tool_policy] blocked tool call: agent={_agent_tool_policy.get('agent_id') if _agent_tool_policy else None} " - - f"tool={func_name} reason={_reason} trace={get_trace_id()}" - - ) - - tool_result = f"[工具调用已被策略拦截] {_policy_check.get('message', _reason)}" - - messages.append({ - - "role": "tool", - - "tool_call_id": tc.get("id", f"call_{round_num}"), - - "content": tool_result, - - }) - - _tool_call_count += 1 - - _round_tool_count += 1 - - continue - - _tool_call_count += 1 - - _round_tool_count += 1 - - - - - - - - func_args_str = tc.get("function", {}).get("arguments", "{}") - - - - - - - - try: - - - - - - - - func_args = json.loads(func_args_str) - - - - - - - - except Exception: - - - - - - - - func_args = {"input": func_args_str} - - - - - - - - - - - - - - - - if func_name.startswith("mcp_tool_"): - - - - # MCP 工具调用：代理到对应 MCP 服务执行 tools/call - - - - _info = mcp_tool_map.get(func_name) - - - - tool_result = "[MCP 工具未找到或映射缺失]" - - - - if _info: - - - - _srv = next( - - - - (s for s in mcp_servers_conn if s["id"] == _info["server_id"]), - - - - None, - - - - ) - - - - if _srv: - - - - async with span(f"mcp.call:{_info.get('name') if isinstance(_info, dict) else None}"): - - tool_result = await call_mcp_tool( - - - - _srv, _info["name"], func_args - - - - ) - - - - called_mcp_names.append(_srv.get("name", "")) - process_steps.append({ - "type": "mcp", - "name": _info.get("name") if isinstance(_info, dict) else func_name, - "server": _srv.get("name", ""), - "args": func_args, - "result": (tool_result or "")[:800], - }) - - - - - else: - - - - async with span(f"skill.execute:{func_name}"): - - tool_result = _execute_skill_tool( - - - - db, func_name, func_args, skills, req.content - - - - ) - - - - _sk = next((s for s in skills if s.get("func_name") == func_name), None) - - - - if _sk: - - - - called_skill_names.append(_sk.get("name", "")) - process_steps.append({ - "type": "skill", - "name": (_sk.get("name") if _sk else func_name), - "args": func_args, - "result": (tool_result or "")[:800], - }) + # 将assistant消息(含tool_calls)加入上下文 @@ -2893,7 +2644,7 @@ async def _send_message_core( - "role": "tool", + "role": "assistant", @@ -2901,7 +2652,7 @@ async def _send_message_core( - "tool_call_id": tc.get("id", f"call_{round_num}"), + "content": message.get("content") or "", @@ -2909,7 +2660,7 @@ async def _send_message_core( - "content": tool_result, + "tool_calls": tool_calls, @@ -2933,7 +2684,7 @@ async def _send_message_core( - # 继续下一轮让模型处理tool结果 + # Stage A.2: 真实Skill执行 @@ -2941,7 +2692,7 @@ async def _send_message_core( - continue + for tc in tool_calls: @@ -2949,53 +2700,71 @@ async def _send_message_core( - else: + func_name = tc.get("function", {}).get("name", "") + # ===== 3.1 tool_policy 沙箱：逐工具校验（白名单 + 预算 + 每轮上限） ===== + _mcp_info = mcp_tool_map.get(func_name) if func_name.startswith("mcp_tool_") else None + _policy_check = _check_tool_policy(func_name, _mcp_info, skills, _agent_tool_policy, _tool_call_count, _round_tool_count) + if not _policy_check["allowed"]: + _reason = _policy_check["reason"] + logger.warning( - # 普通回复 + f"[tool_policy] blocked tool call: agent={_agent_tool_policy.get('agent_id') if _agent_tool_policy else None} " + f"tool={func_name} reason={_reason} trace={get_trace_id()}" + ) + tool_result = f"[工具调用已被策略拦截] {_policy_check.get('message', _reason)}" + messages.append({ + "role": "tool", + "tool_call_id": tc.get("id", f"call_{round_num}"), - ai_content = message.get("content", "") + "content": tool_result, + }) + _tool_call_count += 1 + _round_tool_count += 1 + continue + _tool_call_count += 1 + _round_tool_count += 1 - if not ai_content and round_num > 0: + func_args_str = tc.get("function", {}).get("arguments", "{}") - ai_content = "AI模型已处理工具调用结果, 但返回了空响应。" + try: - break + func_args = json.loads(func_args_str) @@ -3003,17 +2772,17 @@ async def _send_message_core( + except Exception: - if not ai_content: + func_args = {"input": func_args_str} - ai_content = "AI模型返回了空响应, 请稍后重试。" @@ -3027,25 +2796,290 @@ async def _send_message_core( + if func_name.startswith("mcp_tool_"): - except Exception as e: + # MCP 工具调用：代理到对应 MCP 服务执行 tools/call + _info = mcp_tool_map.get(func_name) - logger.warning(f"AI API调用异常: {e}") + tool_result = "[MCP 工具未找到或映射缺失]" + if _info: + _srv = next( - ai_content = "AI服务暂时不可用, 请稍后重试或检查模型API配置。" + + + (s for s in mcp_servers_conn if s["id"] == _info["server_id"]), + + + + None, + + + + ) + + + + if _srv: + + + + async with span(f"mcp.call:{_info.get('name') if isinstance(_info, dict) else None}"): + + tool_result = await call_mcp_tool( + + + + _srv, _info["name"], func_args + + + + ) + + + + called_mcp_names.append(_srv.get("name", "")) + process_steps.append({ + "type": "mcp", + "name": _info.get("name") if isinstance(_info, dict) else func_name, + "server": _srv.get("name", ""), + "args": func_args, + "result": (tool_result or "")[:800], + }) + + + + + else: + + + + async with span(f"skill.execute:{func_name}"): + + tool_result = _execute_skill_tool( + + + + db, func_name, func_args, skills, req.content + + + + ) + + + + _sk = next((s for s in skills if s.get("func_name") == func_name), None) + + + + if _sk: + + + + called_skill_names.append(_sk.get("name", "")) + process_steps.append({ + "type": "skill", + "name": (_sk.get("name") if _sk else func_name), + "args": func_args, + "result": (tool_result or "")[:800], + }) + + + + + + + + + messages.append({ + + + + + + + + "role": "tool", + + + + + + + + "tool_call_id": tc.get("id", f"call_{round_num}"), + + + + + + + + "content": tool_result, + + + + + + + + }) + + + + + + + + + + + + + + + + # 继续下一轮让模型处理tool结果 + + + + + + + + continue + + + + + + + + else: + + + + + + + + # 普通回复 + + + + + + + + ai_content = message.get("content", "") + + + + + + + + if not ai_content and round_num > 0: + + + + + + + + ai_content = "AI模型已处理工具调用结果, 但返回了空响应。" + + + + + + + + break + + + + + + + + + + + + + + + + if not ai_content: + + + + + + + + ai_content = "AI模型返回了空响应, 请稍后重试。" + + + + + + + + + + + + + + + + except Exception as e: + + + + + + + + logger.warning(f"AI API调用异常: {e}") + + + + + + + + ai_content = "AI服务暂时不可用, 请稍后重试或检查模型API配置。" + + + + + + + + else: + + + + + + + + ai_content = "该模型缺少API密钥或端点配置, 请在后台管理-大模型库中配置API Key和Endpoint。" @@ -3061,7 +3095,7 @@ async def _send_message_core( - ai_content = "该模型缺少API密钥或端点配置, 请在后台管理-大模型库中配置API Key和Endpoint。" + ai_content = "请选择一个已配置的AI模型进行对话。" @@ -3069,7 +3103,7 @@ async def _send_message_core( - else: + elif not external_handled: @@ -3077,7 +3111,43 @@ async def _send_message_core( - ai_content = "请选择一个已配置的AI模型进行对话。" + ai_content = "请先在后台管理-大模型库配置默认对话模型（或聊天中选择模型），Auto 模式下将自动使用平台默认模型。" + + + + + # ===== Post-check: 失败则切换备选，成功则跳出 ===== + _is_error = ( + ai_content and ( + "AI模型调用失败" in str(ai_content) + or "AI服务暂时不可用" in str(ai_content) + or "AI模型返回了空响应" in str(ai_content) + or "该模型缺少API密钥" in str(ai_content) + ) + ) + if _is_error: + _retry_log.append(f"{ai_model.name}: {str(ai_content)[:60]}") + logger.warning( + f"[Failover] 模型 {ai_model.name} 失败，切换至备选..." + ) + continue # try next model + + # Success — exit failover loop + break + else: + # Loop completed without break = all models failed + _all_failed = True + + # ===== All models failed? ===== + if _all_failed: + _names = ", ".join([m.name for m in _try_models]) + _details = "; ".join(_retry_log[-3:]) if _retry_log else "未知错误" + ai_content = ( + f"已尝试 {len(_try_models)} 个模型（{_names}）均调用失败。\n" + f"最后错误: {_details}\n" + f"请稍后重试或检查模型配置。" + ) + logger.error(f"[Failover] 全部 {len(_try_models)} 个模型均失败: {_details}") @@ -3085,17 +3155,6 @@ async def _send_message_core( - elif not external_handled: - - - - - - - - ai_content = "请先在后台管理-大模型库配置默认对话模型（或聊天中选择模型），Auto 模式下将自动使用平台默认模型。" - -