From 7dda635c9ac58803f29b958163465007c649c2cc Mon Sep 17 00:00:00 2001 From: ymq Date: Thu, 10 Sep 2026 18:36:54 +0800 Subject: [PATCH] =?UTF-8?q?feat(agent):=20=E4=BA=94=E7=BA=A7=E4=BD=9C?= =?UTF-8?q?=E7=94=A8=E5=9F=9F=E5=B7=A5=E5=85=B7=E9=97=A8=E7=A6=81=E6=8E=A5?= =?UTF-8?q?=E7=BA=BF+patch=5Ffile+=E5=8E=9F=E7=94=9F=E8=A7=86=E8=A7=89?= =?UTF-8?q?=E5=A4=9A=E6=A8=A1=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. _init_components 加第8步 _resolve_tool_scopes: 调 core.tool_sources 五级解析, registry 重建为解析后工具集 + capability 工具合成注册(路由 exec_capability_tool) 2. _execute_tool 加作用域执行门禁: LLM 幻觉调用作用域外工具名(generic喊create_task/ A产线喊B产线工具)直接拒绝并回可行动提示,不只 schema 不给看;解析失败降级不拦截 3. patch_file handler(_t_patch_file): 唯一性校验+replace_all+原子写+越界防护+二进制拒绝 4. 原生视觉: run(image_parts) 构造多模态用户消息; _call_llm 含图失败剥离重试一次 (_degrade_images_if_needed)+注入如实告知说明(绝不假装看过图); _estimate_tokens/ _summarize 兼容 list content(_content_as_text); gateway.run_message 加 image_paths 透传+build_image_parts+超限notes注入 --- pipeline_service/agent_loop_v2.py | 265 ++++++++++++++++++++++++++++-- pipeline_service/gateway.py | 20 ++- 2 files changed, 273 insertions(+), 12 deletions(-) diff --git a/pipeline_service/agent_loop_v2.py b/pipeline_service/agent_loop_v2.py index 04f4ce0..1d348f8 100644 --- a/pipeline_service/agent_loop_v2.py +++ b/pipeline_service/agent_loop_v2.py @@ -127,6 +127,11 @@ class AgentExecutor: # ── 委派运行时（2026-09-10）── self._delegate_depth: int = delegate_depth # 子agent不能再委派（subagents.MAX_DEPTH） self._subagent_id: str = "" # 本 executor 若是后台子agent，其 id（心跳/steer/stop 钩子用） + # ── 五级作用域工具解析结果（2026-09-10）── + self._has_images: bool = False # 本轮消息含图片 parts（LLM 失败降级重试用） + self._allowed_tools: Optional[set] = None # None=未解析(回退旧行为)；set=执行层门禁允许集 + self._capability_tool_names: set = set() # 技能 frontmatter 声明的 capability 工具名 + self._tool_scope_trace: List[str] = [] # 诊断轨迹（/tools slash 可展示） # 懒加载 self._tool_registry = None @@ -139,12 +144,18 @@ class AgentExecutor: # 主入口 # ═══════════════════════════════════════════════════════ - async def run(self, user_input: str, history: List[dict] = None) -> AsyncGenerator[str, None]: + async def run(self, user_input: str, history: List[dict] = None, + image_parts: List[dict] = None) -> AsyncGenerator[str, None]: """执行 agent 主循环。 Args: user_input: 用户输入 history: 历史消息列表 [{"role":"user"|"assistant","content":"..."}] + image_parts: 上传图片的 OpenAI 多模态 parts（upload_tools.build_image_parts + 产物）。非空时首条用户消息构造为多模态 content 数组 + [{"type":"text",...}, {"type":"image_url",...}]—— + llm_bridge → llm_v1 端点 → inference 全链原样透传 messages， + 视觉模型直接看图；非视觉模型由 _call_llm 的降级重试兜底。 Yields: NDJSON 行：{"type":"progress"|"tool_call"|"reply"|"error", ...} @@ -186,7 +197,14 @@ class AgentExecutor: self._msgs = [{"role": "system", "content": system_prompt}] if hist_msgs: self._msgs.extend(hist_msgs) - self._msgs.append({"role": "user", "content": user_input}) + if image_parts: + # 原生视觉（2026-09-10）：OpenAI 多模态 content 数组。 + # 文本部分放最前，图片 parts 依次跟随；_has_images 标记供降级重试用。 + self._msgs.append({"role": "user", "content": + [{"type": "text", "text": user_input}] + list(image_parts)}) + self._has_images = True + else: + self._msgs.append({"role": "user", "content": user_input}) # Step 4.5: 危险命令确认回复短路（pending_confirm + 确认/全部确认/取消） confirm_decision = self._detect_confirm_decision(user_input) @@ -525,6 +543,88 @@ class AgentExecutor: except Exception: pass + # 8. 五级作用域工具解析（2026-09-10）：global(GENERAL_TOOLS) → org(策略+机构技能包) + # → pipeline(能力包+产线策略) → role(RoleSpec 白名单+角色策略) → project(策略微调)。 + # 产出：registry 重建为解析后工具集 + _allowed_tools 执行层门禁 + # （LLM 幻觉调用作用域外工具名 → 直接拒绝，schema 防线挡不住幻觉）。 + try: + await self._resolve_tool_scopes() + except Exception as e: + # 解析失败不阻断会话：回退旧行为（registry 保持 config.tools 全量，无门禁） + logger.warning(f"tool scope resolve failed, fallback to base tools: {e}") + self._allowed_tools = None + + async def _resolve_tool_scopes(self): + """五级作用域工具解析（pipeline_core.tool_sources）。 + + - merged_skills 用与技能注入完全相同的六级可见性参数（generic 只 global）， + 技能作用域即工具作用域：机构装的技能包声明的 capability 工具只对该机构可见。 + - capability 工具注册进本会话 registry（合成 ToolDefinition + 路由 handler + → capability_tools.exec_capability_tool），native FC schema 自动带上。 + - 策略表 pipeline_tool_policies 缺表/查询失败 → 按无策略继续（load_policies 内降级）。 + """ + from pipeline_core.tool_sources import resolve_scoped_tools + + # 与 _build_system_prompt 技能注入同款的可见性参数 + pipeline_id = "" if self.generic else self.pipeline_id + role = "" if self.generic else (self.role or "") + project_id = "" if self.generic else self.project_id + org_id = "" if self.generic else (self.org_id or "") + user_id = "" if self.generic else (self.user_id or "") + + merged = None + if self.config.skills.enabled and self._skill_loader: + try: + merged = self._skill_loader.get_merged( + pipeline_id=pipeline_id, role=role, project_id=project_id, + org_id=org_id, user_id=user_id) + except Exception as e: + logger.warning(f"get_merged for tool scopes failed: {e}") + + from sqlor.dbpools import DBPools + db = DBPools() + async with db.sqlorContext("pipeline") as sor: + res = await resolve_scoped_tools( + sor, base_tools=self.config.tools, merged_skills=merged, + generic=self.generic, org_id=org_id, user_id=user_id, + pipeline_id=pipeline_id, role=role, project_id=project_id) + + self._tool_scope_trace = res.trace + self._capability_tool_names = set(res.capability_tools or []) + self._allowed_tools = set(res.allowed) + + # registry 重建：解析后的 ToolDefinition 集 + capability 工具合成注册 + if self._tool_registry is not None: + from pipeline_core.tool_registry import ToolRegistry + from pipeline_core.agent_config import ToolDefinition + reg = ToolRegistry() + for t in res.tools: + reg.register(t) + # capability 工具：schema 从 capability_tools.TOOL_SCHEMAS 取， + # handler 路由到 exec_capability_tool（系统上下文自动注入，不暴露给 LLM） + for name in sorted(self._capability_tool_names): + if reg.get(name) is not None: + continue + try: + from .capability_tools import TOOL_SCHEMAS + schema = TOOL_SCHEMAS.get(name) + if not schema: + continue # 声明了但无实现 schema：跳过（不注册幻觉入口） + td = ToolDefinition( + name=name, + description=schema["description"], + parameters=dict(schema.get("params") or {}), + category="capability", + required=list(schema.get("required") or []), + ) + reg.register(td) + except Exception as e: + logger.warning(f"capability tool {name} register failed: {e}") + self._tool_registry = reg + logger.info("tool scopes resolved: allowed=%d (capability=%d) trace=%s", + len(self._allowed_tools), len(self._capability_tool_names), + " | ".join(res.trace)) + async def _resolve_skills_base_dir(self): """技能根目录 = 全局 skills/（单一技能树，所有机构共享读，2026-08-21 重构）。""" try: @@ -670,18 +770,61 @@ class AgentExecutor: org_id=self.org_id, ) except Exception as e: + # 原生视觉降级（2026-09-10）：模型不支持多模态 content 数组时 + # 上游会报错——剥离图片 parts 重试一次，并注入说明让 LLM 如实 + # 告知用户「当前模型看不了图」（绝不静默丢图假装看过了）。 + degraded = self._degrade_images_if_needed(e) + if degraded: + try: + return await llm_call_msgs_native( + self._msgs, tools=tools_schema, model=self.model_name, + temperature=self.config.temperature, org_id=self.org_id) + except Exception: + pass logger.error(f"native function calling failed, fallback to text: {e}") # 回退：纯文本调用（content 为原始文本） from pipeline_service.llm_bridge import llm_call_msgs - content = await llm_call_msgs( - self._msgs, - model=self.model_name, - temperature=self.config.temperature, - org_id=self.org_id, - ) + try: + content = await llm_call_msgs( + self._msgs, + model=self.model_name, + temperature=self.config.temperature, + org_id=self.org_id, + ) + except Exception as e: + if self._degrade_images_if_needed(e): + content = await llm_call_msgs( + self._msgs, model=self.model_name, + temperature=self.config.temperature, org_id=self.org_id) + else: + raise return {"content": content or "", "tool_calls": []} + def _degrade_images_if_needed(self, err: Exception) -> bool: + """含图消息调用失败时剥离图片 parts 重试（一次性降级）。 + + 只对「消息里确实有图」的失败生效；剥离后就地改写 self._msgs + （多模态 content 数组 → 纯文本 + 说明），返回是否执行了降级。 + """ + if not self._has_images: + return False + self._has_images = False # 只降级一次 + n_imgs = 0 + for m in self._msgs: + c = m.get("content") + if isinstance(c, list): + texts = [pt.get("text", "") for pt in c if pt.get("type") == "text"] + n_imgs += sum(1 for pt in c if pt.get("type") == "image_url") + m["content"] = ("\n".join(t for t in texts if t) + + f"\n\n[系统说明：用户本次上传了 {n_imgs} 张图片，但当前模型" + f"（{self.model_name or '默认模型'}）调用图片失败" + f"（{str(err)[:120]}），图片已剥离。请如实告知用户当前模型" + f"无法查看图片，可建议切换支持视觉的模型，或改用 " + f"invoke_model 的 i2t 图像理解能力。不要假装看过图片。]") + logger.warning(f"image degrade: stripped {n_imgs} images after LLM failure: {err}") + return n_imgs > 0 + # ═══════════════════════════════════════════════════════ # 解析 LLM 输出 # ═══════════════════════════════════════════════════════ @@ -782,6 +925,20 @@ class AgentExecutor: # 别名归一化：LLM 会编造工具名（2026-08-31 项目管理的常见变体）， # 映射到真实工具，避免「未知工具」导致用户指令落空。 tool_name = _PROJECT_TOOL_ALIASES.get(tool_name, tool_name) + + # 0. 五级作用域执行门禁（2026-09-10）：schema 里不给看挡不住幻觉—— + # LLM 仍可能喊出作用域外工具名（generic 喊 create_task、A 产线喊 B 产线 + # 工具、他机构技能包的工具）。必须在分发前按解析出的允许集硬拦， + # 拒绝并回可行动提示（列出本会话真实可用工具），而不是「未知工具」让 LLM 瞎猜。 + # _allowed_tools 为 None 时（解析失败降级）不拦截，回退旧行为。 + if self._allowed_tools is not None and tool_name not in self._allowed_tools: + avail = sorted(self._allowed_tools) + shown = ", ".join(avail[:20]) + ("..." if len(avail) > 20 else "") + scope_desc = "通用会话" if self.generic else ( + f"产线 {self.pipeline_id}" + (f"/角色 {self.role}" if self.role else "")) + return (f"FAIL: 工具 {tool_name} 不在本会话（{scope_desc}）的可用范围内，已拒绝执行。" + f"本会话可用工具: {shown}") + # 1. 注册的 handler if self._tool_registry: handler = self._tool_registry.get_handler(tool_name) @@ -797,6 +954,14 @@ class AgentExecutor: logger.error(f"Tool {tool_name} handler error: {e}") return f"ERROR: {str(e)[:300]}" + # 1.5 capability 工具（技能 frontmatter 驱动，2026-09-10）： + # 路由到 capability_tools.exec_capability_tool（按 TOOL_SCHEMAS 分发到 + # 对应 capability 模块，系统上下文 project_id/iteration_id/who/agent_id + # 自动注入，不暴露给 LLM）。generic 会话不挂（作用域解析已排除）。 + if tool_name in self._capability_tool_names: + from .capability_tools import exec_capability_tool + return await exec_capability_tool(tool_name, params, self._build_ctx()) + # 2. 产线能力包工具（从 PipelineAbility 注册表按 pipeline_id 取 handler） result = await self._execute_ability_tool(tool_name, params) if result is not None: @@ -903,6 +1068,7 @@ class AgentExecutor: "process": self._t_process, # ── 通用工具集（Hermes CLI 能力子集）── "read_file": self._t_read_file, + "patch_file": self._t_patch_file, "load_skill": self._t_load_skill, "list_packs": self._t_list_packs, "install_pack": self._t_install_pack, @@ -1504,6 +1670,58 @@ class AgentExecutor: except Exception as e: return f"ERROR: {str(e)[:300]}" + async def _t_patch_file(self, sor, p, pid): + """定点替换（2026-09-10，对齐 Hermes patch 工具核心语义）。 + + - old_string 默认须在文件中唯一（0 次/多次都拒绝并给可行动提示）， + replace_all=true 时替换全部出现； + - 路径经 _resolve_ws_path 越界防护（与 read/write_file 同一边界）； + - 二进制/不可解码文件拒绝（文本文件专用）； + - 原子写（.tmp + os.replace），防并发读到半截。 + """ + path = (p.get("path") or "").strip() + old_string = p.get("old_string") + new_string = p.get("new_string") + if not path: + return "FAIL: 需要文件路径 path" + if not isinstance(old_string, str) or old_string == "": + return "FAIL: 需要 old_string（要替换的原文片段，删除时 new_string 传空字符串）" + if not isinstance(new_string, str): + return "FAIL: 需要 new_string（替换后的文本；删除片段传空字符串）" + if old_string == new_string: + return "FAIL: old_string 与 new_string 相同，无需修改" + full = self._resolve_ws_path(path) + if not full: + return f"FAIL: 路径越界 {path}" + if not os.path.isfile(full): + return f"FAIL: 文件不存在 {path}（新建文件用 write_file）" + replace_all = str(p.get("replace_all", "")).strip().lower() in ("true", "1", "yes") + try: + with open(full, "r", encoding="utf-8") as f: + text = f.read() + except UnicodeDecodeError: + return f"FAIL: {path} 不是 UTF-8 文本文件，patch_file 只支持文本" + except Exception as e: + return f"ERROR: 读取失败 {str(e)[:200]}" + n = text.count(old_string) + if n == 0: + return ("FAIL: old_string 在文件中未找到。先用 read_file 核对原文" + "（空白/缩进/换行必须完全一致）") + if n > 1 and not replace_all: + return (f"FAIL: old_string 在文件中出现 {n} 次（须唯一）。" + "请带更多上下文使其唯一，或 replace_all=true 替换全部") + new_text = text.replace(old_string, new_string) if replace_all \ + else text.replace(old_string, new_string, 1) + try: + tmp = full + ".patch.tmp" + with open(tmp, "w", encoding="utf-8") as f: + f.write(new_text) + os.replace(tmp, full) + except Exception as e: + return f"ERROR: 写入失败 {str(e)[:200]}" + return (f"OK: {path} 已替换 {n if replace_all else 1} 处" + f"（文件 {len(text)}→{len(new_text)} 字符）") + async def _t_list_files(self, sor, p, pid): path = p.get("path", "") or "." full = self._resolve_ws_path(path) @@ -1858,17 +2076,44 @@ class AgentExecutor: total = 0 for m in self._msgs: content = m.get("content") or "" + if isinstance(content, list): + # 多模态消息（2026-09-10）：文本部分按字符算，图片按固定 1000 token + # 估（base64 不进文本计数——图片实际 token 由视觉模型分块决定， + # 此处只需量级正确触发压缩阈值） + for pt in content: + if pt.get("type") == "text": + total += len(pt.get("text") or "") // 2 + 1 + elif pt.get("type") == "image_url": + total += 1000 + continue # 简单估算：平均每 2 字符 1 token total += len(content) // 2 + 1 return total + @staticmethod + def _content_as_text(content) -> str: + """消息 content 归一为纯文本（多模态 list 取文本部分 + 图片占位）。 + + 压缩/估算/日志统一走这里——list content 直接 len()/join 会 TypeError + （同 native FC content=None 的教训：所有碰 _msgs content 的地方都要防御）。 + """ + if isinstance(content, list): + parts = [] + for pt in content: + if pt.get("type") == "text": + parts.append(pt.get("text") or "") + elif pt.get("type") == "image_url": + parts.append("[图片]") + return "\n".join(parts) + return content or "" + async def _summarize(self, messages: list) -> str: """压缩消息为摘要""" if len(messages) <= 2: - return "\n".join((m.get("content") or "")[:200] for m in messages) + return "\n".join(self._content_as_text(m.get("content"))[:200] for m in messages) text = "\n".join( - f"[{m['role']}]: {(m.get('content') or '')[:500]}" + f"[{m['role']}]: {self._content_as_text(m.get('content'))[:500]}" for m in messages ) diff --git a/pipeline_service/gateway.py b/pipeline_service/gateway.py index 510d27d..b4598df 100644 --- a/pipeline_service/gateway.py +++ b/pipeline_service/gateway.py @@ -124,7 +124,8 @@ class Gateway: async def run_message(self, channel: str, user_id: str, content: str, history=None, role: str = "", generic: bool = False, base_url: str = "", session_id: str = "", - pipeline_id: str = "", model_id: str = "") -> AsyncGenerator[str, None]: + pipeline_id: str = "", model_id: str = "", + image_paths=None) -> AsyncGenerator[str, None]: """统一消息入口：解析上下文 → 加载产线能力 → AgentExecutor 执行 → yield 事件流。 Args: @@ -211,8 +212,23 @@ class Gateway: base_url=base_url, generic=generic, session=sess, session_id=session_id, default_pipeline_id=pipeline_id) + # 3.5 原生视觉（2026-09-10）：上传图片构造多模态 parts。 + # 上限/超限策略在 build_image_parts（8MB/张、4张/条，超限显式告知不静默丢）。 + # 构造失败不阻断会话（降级为纯文本 + 说明）。 + image_parts, image_notes = [], [] + if image_paths: + try: + from pipeline_core.upload_tools import build_image_parts + image_parts, image_notes = build_image_parts(image_paths) + except Exception as e: + logger.warning(f"build_image_parts failed: {e}") + image_notes = [f"图片处理失败（{str(e)[:100]}），本条消息按纯文本处理。"] + if image_notes: + content = content + "\n\n[系统说明·本次上传图片] " + "；".join(image_notes) + # 4. 转发标准化事件流 - async for chunk in executor.run(content, history=history): + async for chunk in executor.run(content, history=history, + image_parts=image_parts or None): yield chunk async def _resolve_and_persist_model(self, user_id: str, project_id: str,