2630 lines
133 KiB
Python
2630 lines
133 KiB
Python
"""
|
||
pipeline-service v2: Agent Executor — 智能执行引擎
|
||
|
||
对照 Hermes Agent 的 run_conversation() + tool dispatch：
|
||
- 多轮 tool-calling loop，max_turns 可配置
|
||
- Memory + Skills 注入每轮 system prompt
|
||
- 上下文压缩（近 token 上限时自动压缩）
|
||
- 并行 tool calls（同一轮内独立工具同时执行）
|
||
- 消息数组污染防护（tool_call JSON 不进入 LLM 上下文）
|
||
- 会话持久化（用户+项目隔离）
|
||
- ask_user 澄清机制
|
||
- delegate_subtask 子任务分发
|
||
|
||
使用方法：
|
||
executor = AgentExecutor(config, project_id=..., user_id=...)
|
||
async for chunk in executor.run("用户输入"):
|
||
yield chunk # NDJSON 流式输出
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import os
|
||
import time
|
||
from typing import AsyncGenerator, Dict, List, Optional
|
||
|
||
logger = logging.getLogger("pipeline.agent_executor")
|
||
|
||
# 项目管理工具别名表：LLM 常编造近义名（停止/终止/删除项目的各种说法），
|
||
# 归一到真实工具名。只加「语义等价」的别名，不扩大权限（权限校验在代码层）。
|
||
# 注意：不能映射既有真实工具名（如 list_projects 是 sdlc_ability 的产线项目列表
|
||
# 工具，语义与 list_my_projects 不同），否则会劫持既有功能。
|
||
_PROJECT_TOOL_ALIASES = {
|
||
"stop_project": "pause_project",
|
||
"terminate_project": "pause_project",
|
||
"suspend_project": "pause_project",
|
||
"halt_project": "pause_project",
|
||
"unpause_project": "resume_project",
|
||
"continue_project": "resume_project",
|
||
"remove_project": "delete_project",
|
||
"destroy_project": "delete_project",
|
||
"get_project_info": "project_info",
|
||
"show_project": "project_info",
|
||
"my_projects": "list_my_projects",
|
||
}
|
||
|
||
from .workspace import WORKSPACE_BASE, GENERAL_SPACE
|
||
|
||
# 产线/项目管理类工具（category=='project'）。产线隔离铁律：纯通用会话
|
||
# （无产线插件）绝不挂载——否则通用助手能看到并操作其他产线的项目
|
||
# （2026-09-05 修复）。过滤同时作用于：① schema（LLM 看不见）
|
||
# ② 执行层（看见也拒绝），纵深防御。
|
||
_PROJECT_TOOL_NAMES = {
|
||
"switch_project", "create_project", "project_info", "list_my_projects",
|
||
"pause_project", "resume_project", "delete_project",
|
||
}
|
||
|
||
# 产线耦合工具拦截（2026-09-08 更新）：run_command 已从 generic 拒绝名单移除——
|
||
# strict bwrap 档（不挂平台目录、可写根=用户 _general/{uid}）根治了「shell 枚举
|
||
# 全工作空间」的泄漏路径，_t_run_command 对 generic 强制 strict=True，无 bwrap 拒绝。
|
||
# 名单保留机制本身，未来若有无法沙箱化的工具仍可加入。
|
||
_GENERIC_DENIED_TOOLS = set()
|
||
|
||
# 用户敏感信息库工具名（2026-09-17）：与 pipeline_core.agent_config.GENERAL_TOOLS 里
|
||
# category="secret" 的工具一一对应，与 secret_vault.SECRET_HANDLERS 的键一致。
|
||
# 三处同步：工具定义(core) + handlers 映射(本文件 _dispatch_sdlc_tool) + handler 实现(secret_vault)。
|
||
_SECRET_TOOL_NAMES = (
|
||
"list_secrets", "save_secret", "use_secret",
|
||
"delete_secret", "set_secret_status", "detect_secret",
|
||
)
|
||
|
||
|
||
# ── 默认工具定义（在 pipeline-core 未加载时使用）──
|
||
|
||
_BUILTIN_ASK_USER_SCHEMA = {
|
||
"name": "ask_user",
|
||
"description": "向用户提问澄清意图。不确定用户想要什么时使用。",
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"question": {"type": "string", "description": "要问的问题"},
|
||
},
|
||
"required": ["question"],
|
||
},
|
||
}
|
||
|
||
|
||
class AgentExecutor:
|
||
"""Agent 执行引擎。
|
||
|
||
负责：
|
||
1. 组装 system prompt（配置 + 记忆 + 技能 + 工具）
|
||
2. 运行 tool-loop（LLM 决策 → 工具执行 → 回传结果）
|
||
3. 上下文压缩
|
||
4. 会话持久化
|
||
"""
|
||
|
||
# 收尾预留轮次（2026-09-11）：轮次预算剩这么多轮时注入「立即收尾」指令，
|
||
# 强制 LLM 停止调工具、基于已有材料输出最终结论。根因见 run() 内 5-pre0 段注释
|
||
#（v1 有 _FORCE_PRODUCE_HINT、v2 此前缺等价机制 → 长任务撞穿 max_turns 后
|
||
# 静默退出，用户只拿到中间进度句）。
|
||
_WRAP_UP_RESERVE = 3
|
||
|
||
# 辅助调用（压缩摘要/技能选择/意图分类）超时上限（2026-09-11）。
|
||
# 这些是「可降级」调用：失败有兜底（摘要退化为截断文本、技能选择退化为目录层）。
|
||
# 给短超时让它们快速失败走兜底，而不是占用端点默认 300s × 3 重试把整轮卡死
|
||
#（实测上游抖动窗口内 utility 调用连烧 15 分钟才放弃，会话全程无响应）。
|
||
_UTILITY_TIMEOUT = 60
|
||
|
||
# 续跑门禁（2026-09-11 商机产线实测）：模型输出纯文本「意图句」就结束 turn、
|
||
# 不调工具 → 循环误判为最终答案终止会话，用户拿到半截话（实测连续 4 个会话
|
||
# 在 tool_call_count 7~11 时这样夭折，报告/总结从未产出）。
|
||
# 门禁条件：已有工作推进(tool_call_count>0) 且回复短于 _CONTINUE_GATE_MIN_CHARS
|
||
#（真结论/报告远长于此）且催促未达上限 → 注入续跑指令继续循环；
|
||
# 达上限后尊重模型终止（防死循环）。
|
||
_CONTINUE_GATE_MIN_CHARS = 400
|
||
_CONTINUE_GATE_MAX = 2
|
||
|
||
# 主调用单次超时（2026-09-11）：端点缺省 330s 太长——上游超时窗内一次失败
|
||
# 要烧 330s×3 重试。实测最长成功响应 3212 token <60s，180s 留 3 倍余量足够，
|
||
# 失败快速返回交给会话级重试跨越窗口。
|
||
_MAIN_TIMEOUT = 180
|
||
# 会话级重试次数（含首次共 3 次尝试）+ 退避秒数：上游超时窗实测分钟级、
|
||
# 间歇出现；3 次尝试 × 180s + 2×15s 退避 ≈ 9.5 分钟跨度，可跨越多数窗口。
|
||
_MAIN_ATTEMPTS = 3
|
||
_MAIN_RETRY_BACKOFF = 15
|
||
|
||
def __init__(
|
||
self,
|
||
config, # AgentConfig (from pipeline_core)
|
||
project_id: str = "",
|
||
user_id: str = "",
|
||
workspace_dir: str = "",
|
||
model_name: str = None,
|
||
role: str = "", # 产线内角色（可选，如 develop/design；驾驶舱 agent 为空）
|
||
base_url: str = "", # 请求 base_url（scheme://host/），供 slash 命令生成 widget 绝对 URL
|
||
generic: bool = False, # True = 纯通用会话（不解析项目、不挂产线能力）
|
||
session=None, # GatewaySession（会话级状态：approve_all/pending_confirm）
|
||
session_id: str = "", # 会话内唯一标识（web 多 tab 独立会话；历史隔离键）
|
||
default_pipeline_id: str = "", # 无当前项目时的默认产线（入口指定，如 bidding_general）
|
||
delegate_depth: int = 0, # 委派深度（0=顶层会话；子agent=1，禁止再委派）
|
||
pii_session_key: str = "", # PII ephemeral 缓存键（gateway _session_key；空=无 PII 层）
|
||
):
|
||
self.config = config
|
||
self.project_id = project_id
|
||
self.user_id = user_id
|
||
self.pii_session_key = pii_session_key
|
||
self.org_id = "" # loaded from project context
|
||
self.pipeline_id = default_pipeline_id or "" # 默认产线（入口指定）；有项目时被项目 pipeline_id 覆盖
|
||
self.role = role # 产线内角色（技能/工具/prompt 按角色加载）
|
||
self.generic = generic # 是否纯通用会话
|
||
self._session = session # GatewaySession（approve_all/pending_confirm 会话级状态）
|
||
self.session_id = session_id # 会话内唯一标识（历史消息按此隔离）
|
||
self.space = GENERAL_SPACE # 项目空间键：generic→'general'，产线→真实 pipeline_id
|
||
self.workspace_dir = workspace_dir or WORKSPACE_BASE
|
||
self.model_name = model_name or config.model_name
|
||
self.base_url = base_url
|
||
|
||
# 运行状态
|
||
self._msgs: List[dict] = [] # LLM 消息数组
|
||
self._turn_count: int = 0
|
||
self._tool_call_count: int = 0
|
||
self._auto_push_count: int = 0 # 限制 auto-inject 次数
|
||
self._session_id: str = ""
|
||
self._started_at: float = 0.0
|
||
self._todos: List[dict] = [] # 会话内任务清单 [{done, content}]
|
||
# ── 委派运行时（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
|
||
self._skill_loader = None
|
||
self._memory_store = None
|
||
self._compressor = None
|
||
self._session_mgr = 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", ...}
|
||
"""
|
||
self._started_at = time.time()
|
||
self._turn_count = 0
|
||
self._tool_call_count = 0
|
||
self._continue_nudges = 0
|
||
|
||
# Step 0: 初始化（解析 pipeline_id / skill_loader 等，slash 命令也需要）
|
||
await self._init_components()
|
||
|
||
# Step 0.5: slash 命令短路（/xxx 直接执行，不进入 LLM 循环）
|
||
if user_input and user_input.strip().startswith("/"):
|
||
from pipeline_core import resolve_slash_command, parse_slash_args
|
||
cmd = resolve_slash_command(user_input.strip(), self.pipeline_id, self.role)
|
||
if cmd and cmd.handler:
|
||
args = parse_slash_args(user_input.strip())
|
||
try:
|
||
result = await cmd.handler(args, self._build_slash_ctx())
|
||
except Exception as e:
|
||
logger.error(f"slash /{cmd.name} error: {e}")
|
||
result = f"ERROR: {str(e)[:300]}"
|
||
yield json.dumps({"type": "reply", "message": result}, ensure_ascii=False) + "\n"
|
||
return
|
||
# 未知 slash 命令：提示并列出可用命令
|
||
from pipeline_core import get_slash_commands
|
||
visible = get_slash_commands(self.pipeline_id, self.role)
|
||
help_text = "未知命令。可用: " + ", ".join(f"/{n}" for n in sorted(visible))
|
||
yield json.dumps({"type": "reply", "message": help_text}, ensure_ascii=False) + "\n"
|
||
return
|
||
|
||
# Step 2: 加载历史
|
||
hist_msgs = await self._load_history(history)
|
||
|
||
# Step 3: 组装 system prompt
|
||
system_prompt = await self._build_system_prompt(user_input)
|
||
|
||
# Step 4: 构建消息数组
|
||
self._msgs = [{"role": "system", "content": system_prompt}]
|
||
if hist_msgs:
|
||
self._msgs.extend(hist_msgs)
|
||
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)
|
||
if confirm_decision and self._session and self._session.pending_confirm:
|
||
pending = self._session.pending_confirm
|
||
if confirm_decision == "cancel":
|
||
self._session.pending_confirm = None
|
||
yield json.dumps({"type": "reply", "message": "已取消该操作。"}, ensure_ascii=False) + "\n"
|
||
await self._save_turn(user_input, "已取消该操作。")
|
||
return
|
||
if confirm_decision == "approve_all":
|
||
self._session.approve_all = True
|
||
self._session.pending_confirm = None
|
||
pending_tool = pending.get("tool", "")
|
||
pending_params = pending.get("params") or {}
|
||
# 回填 assistant tool_call → 执行 → 回填 tool result，随后继续 Tool Loop
|
||
fake_id = f"call_{int(time.time() * 1000)}"
|
||
self._msgs.append({
|
||
"role": "assistant",
|
||
"content": None,
|
||
"tool_calls": [{
|
||
"id": fake_id,
|
||
"type": "function",
|
||
"function": {"name": pending_tool, "arguments": json.dumps(pending_params, ensure_ascii=False)},
|
||
}],
|
||
})
|
||
yield json.dumps({"type": "tool_call", "tool": pending_tool, "params": pending_params}, ensure_ascii=False) + "\n"
|
||
result = await self._execute_tool(pending_tool, pending_params)
|
||
if not result.startswith("未知工具") and not result.startswith("ERROR"):
|
||
self._tool_call_count += 1
|
||
# 显示层上限（2026-09-15 用户要求完整显示，替代旧硬截 [:500]）：
|
||
# 默认 50000 字符 = 正常工具输出完整显示；超限显式告知（绝不静默截断）。
|
||
from .result_cap import cap_tool_result, cap_display_result, resolve_max_chars, resolve_display_max_chars
|
||
yield json.dumps({"type": "tool_result", "tool": pending_tool,
|
||
"result": cap_display_result(result, await resolve_display_max_chars())}, ensure_ascii=False) + "\n"
|
||
# 回填上限（2026-09-11）：工具结果全文回填会撑爆上下文（商机产线实测
|
||
# 8 次数据工具 ~78K token → 压缩摘要自身超时 → 会话零产出）。
|
||
# 截断显式告知 + 给缩小范围指引，让 LLM 自己改策略。
|
||
_capped, _ = cap_tool_result(result, await resolve_max_chars())
|
||
self._msgs.append({"role": "tool", "tool_call_id": fake_id, "content": _capped})
|
||
# 不 return，继续进入 Tool Loop，让 LLM 基于工具结果继续后续步骤
|
||
|
||
# Step 5: Tool Loop
|
||
yield json.dumps({"type": "progress", "message": "思考中..."}, ensure_ascii=False) + "\n"
|
||
|
||
final_reply = ""
|
||
|
||
max_turns = self.config.max_turns
|
||
for turn in range(max_turns):
|
||
self._turn_count = turn + 1
|
||
|
||
# 5-pre0. 收尾门禁（2026-09-11 商机产线实测教训）：轮次预算接近耗尽时
|
||
# 注入「立即收尾」指令，强制 LLM 用已有材料给出最终结论。
|
||
# 根因：v1 角色 agent 有 _FORCE_PRODUCE_HINT 打断探索循环，v2 会话 agent
|
||
# 此前没有等价机制——多工具长任务（如商机调研→功能点估算→成本→总结）
|
||
# 会跑满 max_turns 后静默退出，用户只看到中间进度句（「功能点已算出…」），
|
||
# 拿不到要求的总结。收尾轮不再允许调工具，只准产出最终回答。
|
||
_wrap_turn = max_turns - self._WRAP_UP_RESERVE
|
||
if turn == _wrap_turn:
|
||
self._msgs.append({
|
||
"role": "user",
|
||
"content": (
|
||
"⚠️【收尾指令】你的对话轮次即将用尽（仅剩 %d 轮）。"
|
||
"立即停止调用任何工具，基于已经获取的材料直接给出**完整的最终回答**："
|
||
"把用户要求的每一项都答到（数据/结论/建议/总结），"
|
||
"信息不全的部分如实标注「数据未覆盖」并给出你的专业判断，"
|
||
"不要再做探索、核对或补数。现在就输出最终结论。" % (max_turns - turn)
|
||
),
|
||
})
|
||
logger.info(f"wrap-up hint injected at turn {turn+1}/{max_turns}")
|
||
|
||
# 5-pre. 子 agent 运行时钩子（2026-09-10）：心跳 + steer 消费 + stop 检测。
|
||
# 后台委派的子 executor 带 _subagent_id，每轮 tool-loop 边界：
|
||
# - heartbeat 刷新 meta.updated_at（stale 判活依据）
|
||
# - steer.txt 有内容 → 作为带外父指令注入消息（消费后删除）
|
||
# - stop.flag 存在 → 优雅停止（已有部分结果由 subagents._runner 保留）
|
||
if self._subagent_id:
|
||
try:
|
||
from . import subagents as _sa
|
||
steer_msgs, stop_req = _sa.heartbeat(
|
||
self.workspace_dir, self._subagent_id)
|
||
except Exception as e:
|
||
logger.warning(f"subagent hook failed: {e}")
|
||
steer_msgs, stop_req = [], False
|
||
if stop_req:
|
||
msg = "（子任务被父agent终止，已保留部分结果）"
|
||
yield json.dumps({"type": "reply", "message": msg}, ensure_ascii=False) + "\n"
|
||
return
|
||
for sm in steer_msgs:
|
||
self._msgs.append({
|
||
"role": "user",
|
||
"content": ("[带外指令——来自父agent的中途纠正，与用户消息同等效力，"
|
||
f"立即调整方向] {sm}"),
|
||
})
|
||
|
||
# 5a. 上下文压缩
|
||
if self.config.compression.enabled:
|
||
await self._maybe_compress()
|
||
|
||
# 5b. LLM 调用（native function calling，返回 dict）
|
||
# 模型不可用等配置错误必须真实报给用户（error 事件），
|
||
# 绝不能让异常打断流——前端会停在"思考中..."干等。
|
||
# 会话级重试（2026-09-11）：上游瞬时抖动（TimeoutError/ClientError）时
|
||
# 重试一次再放弃。实测上游存在分钟级抖动窗口（同模型同上下文直调成功、
|
||
# 产线调用 TimeoutError 3 连败），单次失败即终止会话代价过高（整轮工作丢失）。
|
||
# 重试前向用户透出进度，避免"卡死"观感。
|
||
resp = None
|
||
_last_err = None
|
||
for _attempt in range(self._MAIN_ATTEMPTS):
|
||
try:
|
||
resp = await self._call_llm()
|
||
break
|
||
except Exception as e:
|
||
_last_err = e
|
||
_transient = any(k in type(e).__name__ or k in str(e)
|
||
for k in ("Timeout", "ClientError", "ServerDisconnected",
|
||
"ConnectionReset"))
|
||
if not _transient or _attempt >= self._MAIN_ATTEMPTS - 1:
|
||
break
|
||
logger.warning(f"run: LLM 瞬时失败 turn={turn+1} 重试{_attempt+1}: {e}")
|
||
yield json.dumps({
|
||
"type": "progress",
|
||
"message": "模型上游瞬时超时，正在重试…\n",
|
||
}, ensure_ascii=False) + "\n"
|
||
await asyncio.sleep(self._MAIN_RETRY_BACKOFF)
|
||
if resp is None:
|
||
logger.error(f"run: LLM 调用失败 turn={turn+1}: {_last_err}")
|
||
yield json.dumps({"type": "error", "message": str(_last_err)}, ensure_ascii=False) + "\n"
|
||
return
|
||
|
||
# 5c. 原生 function calling：优先处理 tool_calls
|
||
native_calls = (resp.get("tool_calls") or []) if isinstance(resp, dict) else []
|
||
if native_calls:
|
||
# 回填 assistant 消息（含 tool_calls，OpenAI 原生格式要求）
|
||
self._msgs.append({
|
||
"role": "assistant",
|
||
"content": resp.get("content") or None,
|
||
"tool_calls": native_calls,
|
||
})
|
||
for tc in native_calls:
|
||
fn = tc.get("function", {}) if isinstance(tc, dict) else {}
|
||
tool_name = fn.get("name", "")
|
||
try:
|
||
tool_params = json.loads(fn.get("arguments") or "{}")
|
||
except Exception:
|
||
tool_params = {}
|
||
|
||
# 需要确认的工具：记录待确认并暂停，等待用户确认/全部确认/取消
|
||
if self._needs_confirmation(tool_name, tool_params):
|
||
if self._session:
|
||
self._session.pending_confirm = {"tool": tool_name, "params": tool_params}
|
||
yield json.dumps({
|
||
"type": "confirm", "tool": tool_name, "params": tool_params,
|
||
}, ensure_ascii=False) + "\n"
|
||
await self._save_turn(user_input, f"[需确认] {tool_name}")
|
||
return
|
||
|
||
yield json.dumps({
|
||
"type": "tool_call", "tool": tool_name, "params": tool_params,
|
||
}, ensure_ascii=False) + "\n"
|
||
|
||
result = await self._execute_tool(tool_name, tool_params)
|
||
if not result.startswith("未知工具") and not result.startswith("ERROR"):
|
||
self._tool_call_count += 1
|
||
|
||
# 完备性/澄清门禁硬拦截（2026-09-08 用户定夺：一C+二A）：
|
||
# 工具返回 QUESTION: 前缀 = 平台判定「缺完成任务的必备输入」，
|
||
# 已停在花钱动作之前（上游未调用、零费用）。必须把它当 ask_user
|
||
# 直接抛给用户等回答，禁止回填后靠 LLM 自觉转达（deepseek 类
|
||
# 模型会无视、继续空参硬试烧钱）。ask_user 工具同走此前缀，语义统一。
|
||
if isinstance(result, str) and result.startswith("QUESTION:"):
|
||
question = result[len("QUESTION:"):].strip()
|
||
yield json.dumps({
|
||
"type": "ask_user", "message": question,
|
||
}, ensure_ascii=False) + "\n"
|
||
await self._save_turn(user_input, f"[提问] {question}")
|
||
return
|
||
|
||
# 显示层上限（同 Step 4.5，默认 50000 = 完整显示；超限显式告知）
|
||
from .result_cap import cap_display_result, resolve_display_max_chars
|
||
yield json.dumps({
|
||
"type": "tool_result", "tool": tool_name,
|
||
"result": cap_display_result(result, await resolve_display_max_chars()),
|
||
}, ensure_ascii=False) + "\n"
|
||
|
||
# 原生回填：role=tool + tool_call_id
|
||
# 回填上限：防数据类工具全文回填撑爆上下文（见 result_cap 模块 docstring）
|
||
from .result_cap import cap_tool_result, resolve_max_chars
|
||
_capped, _trunc = cap_tool_result(result, await resolve_max_chars())
|
||
if _trunc:
|
||
logger.info(f"tool_result truncated before refill: {tool_name}")
|
||
self._msgs.append({
|
||
"role": "tool",
|
||
"tool_call_id": tc.get("id", "") if isinstance(tc, dict) else "",
|
||
"content": _capped,
|
||
})
|
||
continue
|
||
|
||
# 5d. 文本解析（回退路径，纯文本模型）
|
||
raw = resp.get("content", "") if isinstance(resp, dict) else str(resp)
|
||
actions = self._parse_actions(raw)
|
||
|
||
# 5e. 分发处理
|
||
for act in actions:
|
||
action_type = act.get("action", "")
|
||
|
||
if action_type == "reply":
|
||
# reply 是合法终止动作，直接输出，不再强制注入工具。
|
||
# （历史版本会在 tool_call_count==0 时用硬编码关键词强制路由，
|
||
# 那会剥夺 LLM 诚实降级/能力自省的能力，把 agent 变成路由器。）
|
||
final_reply = act.get("message", "")
|
||
|
||
# 续跑门禁（2026-09-11）：已有工作推进但回复是短意图句
|
||
#（"我再拉一次数据…"）= 模型提前结束 turn 而非真结论。
|
||
# 注入续跑指令继续循环，不让会话夭折在半截话上。
|
||
if (self._tool_call_count > 0
|
||
and len(final_reply) < self._CONTINUE_GATE_MIN_CHARS
|
||
and self._continue_nudges < self._CONTINUE_GATE_MAX):
|
||
self._continue_nudges += 1
|
||
logger.warning(
|
||
f"run: 续跑门禁 turn={turn+1} 短回复({len(final_reply)}字)"
|
||
f" tool_calls={self._tool_call_count} 催促{self._continue_nudges}次")
|
||
yield json.dumps({
|
||
"type": "progress",
|
||
"message": "检测到未完成的中途回复，继续推进任务…\n",
|
||
}, ensure_ascii=False) + "\n"
|
||
self._msgs.append({"role": "assistant", "content": final_reply})
|
||
self._msgs.append({"role": "user", "content": (
|
||
"【续跑指令】你刚才的回复是中途进度句，不是最终答案，"
|
||
"任务尚未完成（用户要求的产出还没交付）。"
|
||
"请继续执行：要么调用工具推进下一步，"
|
||
"要么直接输出完整的最终结论/报告（含全部要求的分析项）。"
|
||
"不要只说下一步打算做什么。")})
|
||
continue
|
||
|
||
yield json.dumps({
|
||
"type": "reply", "message": final_reply,
|
||
}, ensure_ascii=False) + "\n"
|
||
await self._save_turn(user_input, final_reply)
|
||
return
|
||
|
||
elif action_type == "ask_user":
|
||
# ask_user 是合法终止动作：把问题抛给用户，等待回答。
|
||
question = act.get("question", "")
|
||
yield json.dumps({
|
||
"type": "ask_user", "message": question,
|
||
}, ensure_ascii=False) + "\n"
|
||
await self._save_turn(user_input, f"[提问] {question}")
|
||
return
|
||
|
||
elif action_type == "tool_call":
|
||
tool_name = act.get("tool", "")
|
||
tool_params = act.get("params", {})
|
||
|
||
# 需要确认的工具：记录待确认并暂停，等待用户确认/全部确认/取消
|
||
if self._needs_confirmation(tool_name, tool_params):
|
||
if self._session:
|
||
self._session.pending_confirm = {"tool": tool_name, "params": tool_params}
|
||
yield json.dumps({
|
||
"type": "confirm",
|
||
"tool": tool_name,
|
||
"params": tool_params,
|
||
}, ensure_ascii=False) + "\n"
|
||
await self._save_turn(user_input, f"[需确认] {tool_name}")
|
||
return
|
||
|
||
# 执行工具
|
||
yield json.dumps({
|
||
"type": "tool_call", "tool": tool_name,
|
||
"params": tool_params,
|
||
}, ensure_ascii=False) + "\n"
|
||
|
||
result = await self._execute_tool(tool_name, tool_params)
|
||
if not result.startswith("未知工具") and not result.startswith("ERROR"):
|
||
self._tool_call_count += 1
|
||
|
||
# 完备性/澄清门禁硬拦截（同 native 路径，2026-09-08 一C+二A）
|
||
if isinstance(result, str) and result.startswith("QUESTION:"):
|
||
question = result[len("QUESTION:"):].strip()
|
||
yield json.dumps({
|
||
"type": "ask_user", "message": question,
|
||
}, ensure_ascii=False) + "\n"
|
||
await self._save_turn(user_input, f"[提问] {question}")
|
||
return
|
||
|
||
# 显示层上限（同 native 路径，默认 50000 = 完整显示；超限显式告知）
|
||
from .result_cap import cap_display_result, resolve_display_max_chars
|
||
yield json.dumps({
|
||
"type": "tool_result", "tool": tool_name,
|
||
"result": cap_display_result(result, await resolve_display_max_chars()),
|
||
}, ensure_ascii=False) + "\n"
|
||
|
||
# 反馈给 LLM（关键：不存原始 tool_call JSON）
|
||
self._msgs.append({
|
||
"role": "assistant",
|
||
"content": f"已调用 {tool_name}",
|
||
})
|
||
self._msgs.append({
|
||
"role": "user",
|
||
"content": f"工具 {tool_name} 结果：\n{result}",
|
||
})
|
||
|
||
elif action_type == "error":
|
||
final_reply = act.get("message", "处理出错")
|
||
yield json.dumps({
|
||
"type": "error", "message": final_reply,
|
||
}, ensure_ascii=False) + "\n"
|
||
return
|
||
|
||
# 超过 max_turns
|
||
final_reply = final_reply or "任务过于复杂，请简化描述后重试。"
|
||
yield json.dumps({
|
||
"type": "reply", "message": final_reply,
|
||
}, ensure_ascii=False) + "\n"
|
||
await self._save_turn(user_input, final_reply)
|
||
|
||
# ═══════════════════════════════════════════════════════
|
||
# 初始化
|
||
# ═══════════════════════════════════════════════════════
|
||
|
||
async def _init_components(self):
|
||
"""懒加载各组件"""
|
||
# 1. 加载 org_id（用户机构优先，fallback 项目机构）+ workspace_dir + pipeline_id
|
||
# 个人选择 llm 按「用户机构」隔离（default_llm_id 指向本机构 llm），故 org_id 须取 users.orgid；
|
||
# 项目机构仅作 fallback（无 user_id 的测试场景）。
|
||
if self.user_id:
|
||
try:
|
||
from sqlor.dbpools import DBPools
|
||
db = DBPools()
|
||
async with db.sqlorContext("pipeline") as sor:
|
||
recs = await sor.sqlExe(
|
||
"SELECT orgid FROM users WHERE id=${uid}$", {"uid": self.user_id})
|
||
if recs:
|
||
self.org_id = getattr(recs[0], 'orgid', '') or ''
|
||
except Exception:
|
||
pass
|
||
|
||
if self.project_id:
|
||
try:
|
||
from sqlor.dbpools import DBPools
|
||
db = DBPools()
|
||
async with db.sqlorContext("pipeline") as sor:
|
||
recs = await sor.sqlExe(
|
||
"SELECT org_id, workspace_dir, pipeline_id FROM sd_projects WHERE id=${pid}$",
|
||
{"pid": self.project_id})
|
||
if recs:
|
||
if not self.org_id:
|
||
self.org_id = getattr(recs[0], 'org_id', '') or ''
|
||
ws = getattr(recs[0], 'workspace_dir', '') or ''
|
||
if ws:
|
||
self.workspace_dir = ws
|
||
self.pipeline_id = getattr(recs[0], 'pipeline_id', '') or ''
|
||
if not ws:
|
||
# 存量项目无 workspace_dir：兜底解析项目根目录（与上传落盘同一函数，
|
||
# 保证「上传放的位置 = agent 读的位置」）
|
||
from .workspace import get_project_dir_by_id
|
||
_pdir, _ = await get_project_dir_by_id(sor, self.project_id)
|
||
if _pdir:
|
||
self.workspace_dir = _pdir
|
||
except Exception:
|
||
pass
|
||
|
||
# 2. Tool Registry
|
||
# 产线隔离（2026-09-05）：必须每个 executor 独立实例，禁止全局单例。
|
||
# 全局单例会让先启动的产线会话注册的工具（如 sdlc 的 list_projects）
|
||
# 永久残留，之后通用会话把全部工具塞给 LLM → 通用助手看到并调用
|
||
# 其他产线的工具。config.tools 已由 load_agent_config 按会话裁剪，
|
||
# 这里只注册本会话自己的工具。
|
||
try:
|
||
from pipeline_core.tool_registry import ToolRegistry
|
||
self._tool_registry = ToolRegistry()
|
||
if self.config.tools:
|
||
for t in self.config.tools:
|
||
self._tool_registry.register(t)
|
||
except ImportError:
|
||
self._tool_registry = None
|
||
|
||
# 3. Skill Loader — base_dir 动态解析到机构工作目录（多租户隔离）
|
||
try:
|
||
from pipeline_core.skill_loader import get_skill_loader
|
||
base_dir = await self._resolve_skills_base_dir()
|
||
self._skill_loader = get_skill_loader(base_dir)
|
||
if base_dir:
|
||
self._skill_loader.reload()
|
||
except Exception:
|
||
self._skill_loader = None
|
||
|
||
# 4. Memory Store
|
||
try:
|
||
from pipeline_core.memory_store import get_memory_store
|
||
self._memory_store = get_memory_store()
|
||
except ImportError:
|
||
self._memory_store = None
|
||
|
||
# 5. pipeline_id 为空时 fallback 到默认产线（保证 slash/ability/skill 统一按产线挂载）
|
||
# 入口已指定 default_pipeline_id 时优先用它（投标/开发产线独立入口），否则用引擎默认。
|
||
# 产线隔离（2026-09-05）：纯通用会话（generic）绝不回退默认产线——
|
||
# 回退会让 _execute_ability_tool 拿到 sdlc_general 能力包，通用助手
|
||
# 就能调用 list_projects 等产线工具。
|
||
if not self.pipeline_id and not self.generic:
|
||
try:
|
||
from pipeline_core import DEFAULT_ABILITY_ID
|
||
self.pipeline_id = DEFAULT_ABILITY_ID
|
||
except ImportError:
|
||
pass
|
||
|
||
# 6. 项目空间键：generic→'general'（通用助手），否则用真实 pipeline_id（产线之间隔离）
|
||
self.space = GENERAL_SPACE if self.generic else (self.pipeline_id or GENERAL_SPACE)
|
||
|
||
# 7. 通用会话专属工作目录（2026-09-05 第 6 层泄漏修复）：默认文件根是
|
||
# WORKSPACE_BASE 总根（{base}/{org}/{产线}/… 全量可见），LLM 用
|
||
# list_files/read_file 就能从目录名枚举所有产线的项目。通用会话改用
|
||
# 用户专属目录 _general/{user_id}，_resolve_ws_path 越界保护把它圈死。
|
||
if self.generic:
|
||
from .workspace import generic_workspace_dir
|
||
self.workspace_dir = generic_workspace_dir(self.user_id)
|
||
try:
|
||
os.makedirs(self.workspace_dir, exist_ok=True)
|
||
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:
|
||
from pipeline_core.skill_pack import get_skills_base
|
||
return get_skills_base()
|
||
except Exception:
|
||
return self.config.skills.base_dir
|
||
|
||
async def _llm_select_skills(self, user_input: str, catalog, max_skills: int) -> List[str]:
|
||
"""用 LLM 从技能目录选最相关的 N 个技能（语义匹配，非关键词硬编码）。"""
|
||
if not catalog:
|
||
return []
|
||
catalog_text = "\n".join(f"- {name}: {desc}" for name, desc in catalog)
|
||
prompt = (
|
||
f"以下是可用技能目录（技能名: 描述）：\n{catalog_text}\n\n"
|
||
f"用户需求：{user_input}\n\n"
|
||
f"请从上述目录中选出与用户需求最相关的至多 {max_skills} 个技能，"
|
||
f"只返回技能名的 JSON 数组，如 [\"a\", \"b\"]。没有相关技能返回 []。"
|
||
)
|
||
try:
|
||
from pipeline_service.llm_bridge import llm_call_msgs
|
||
import json as _json
|
||
import re as _re
|
||
content = await llm_call_msgs(
|
||
[{"role": "user", "content": prompt}],
|
||
model=self.model_name,
|
||
temperature=0,
|
||
org_id=self.org_id,
|
||
purpose='utility',
|
||
session_id=self.session_id,
|
||
timeout=self._UTILITY_TIMEOUT,
|
||
)
|
||
m = _re.search(r"\[[^\]]*\]", content or "")
|
||
if m:
|
||
names = _json.loads(m.group(0))
|
||
return [n for n in names if isinstance(n, str)]
|
||
return []
|
||
except Exception as e:
|
||
logger.error(f"_llm_select_skills failed: {e}")
|
||
return []
|
||
|
||
async def _build_system_prompt(self, user_input: str) -> str:
|
||
"""组装完整 system prompt = 基础 prompt + 记忆 + 技能 + 工具列表"""
|
||
prompt = self.config.system_prompt
|
||
|
||
# 注入当前项目上下文
|
||
proj_name = self.project_id
|
||
try:
|
||
ctx = await self._load_project_context()
|
||
if ctx:
|
||
proj_name = ctx.replace("项目: ", "")
|
||
except Exception:
|
||
pass
|
||
prompt += f"\n当前项目: {proj_name}\n" if proj_name else ""
|
||
|
||
# 注入记忆（分域：通用 + 产线 + 项目 叠加）
|
||
# 产线隔离（2026-09-05）：纯通用会话只读 global 通用记忆，绝不带
|
||
# 「产线+项目」叠加——否则通用助手会看到产线专属记忆。
|
||
if self.config.memory.enabled and self._memory_store:
|
||
# 多租户可见性版（2026-09-10）：传 org_id/user_id → store 按
|
||
# visible_to 过滤（平台种子 + 本机构 org/pipeline/project + 本人 user），
|
||
# 替代旧的「global + pipeline 叠加」——旧版会把其他机构的
|
||
# org/pipeline/project 归属记忆也注入本会话（跨机构泄漏）。
|
||
if self.generic:
|
||
# 通用会话：只注入种子 + 本人 user 域 + 本机构 org 域
|
||
mem_block = await self._memory_store.build_prompt_block(
|
||
max_entries=15, org_id=self.org_id, user_id=self.user_id)
|
||
else:
|
||
mem_block = await self._memory_store.build_prompt_block(
|
||
max_entries=20, org_id=self.org_id, user_id=self.user_id,
|
||
pipeline_id=self.pipeline_id, project_id=self.project_id)
|
||
if mem_block:
|
||
prompt += f"\n\n## 持久记忆\n{mem_block}"
|
||
|
||
# 注入技能（六级隔离：global→org→pipeline→role→project→user）
|
||
if self.config.skills.enabled and self._skill_loader:
|
||
# 产线隔离（2026-09-05）：纯通用会话只检索 global 技能，
|
||
# 不挂产线/角色/项目/机构技能。
|
||
pipeline_id = "" if self.generic else (self.pipeline_id if self.config.skills.enable_pipeline else "")
|
||
role = "" if self.generic else (self.role if self.config.skills.enable_role else "")
|
||
project_id = "" if self.generic else (self.project_id if self.config.skills.enable_project else "")
|
||
org_id = "" if self.generic else (self.org_id if self.config.skills.enable_org else "")
|
||
user_id = "" if self.generic else (self.user_id if self.config.skills.enable_user else "")
|
||
# 技能检索必须 LLM 做（语义匹配，非关键词硬编码）
|
||
catalog = self._skill_loader.get_skill_catalog(
|
||
pipeline_id=pipeline_id, role=role, project_id=project_id,
|
||
org_id=org_id, user_id=user_id)
|
||
names = await self._llm_select_skills(
|
||
user_input, catalog, self.config.skills.max_skills_per_turn)
|
||
if names:
|
||
skill_block = self._skill_loader.build_prompt_block_by_names(
|
||
names, pipeline_id=pipeline_id, role=role, project_id=project_id,
|
||
org_id=org_id, user_id=user_id)
|
||
else:
|
||
# LLM 无结果兜底：按 scope+名字列出前 N 个（不传 user_input，走 fallback 分支）
|
||
skill_block = self._skill_loader.build_prompt_block(
|
||
pipeline_id=pipeline_id, role=role, project_id=project_id,
|
||
org_id=org_id, user_id=user_id,
|
||
max_skills=self.config.skills.max_skills_per_turn)
|
||
if skill_block:
|
||
prompt += f"\n\n{skill_block}"
|
||
|
||
# 注入工具列表（先构建，再替换占位符）
|
||
tools_text = ""
|
||
if self._tool_registry:
|
||
tools_text = self._tool_registry.to_text_description()
|
||
else:
|
||
lines = []
|
||
for t in self.config.tools:
|
||
if t.enabled:
|
||
params = ", ".join(f"{k}: {v}" for k, v in (t.parameters or {}).items())
|
||
lines.append(f"- {t.name}({params}): {t.description}")
|
||
tools_text = "\n".join(lines)
|
||
prompt = prompt.replace("{project_name}", proj_name)
|
||
prompt = prompt.replace("{tools_description}", tools_text)
|
||
|
||
# 敏感信息使用铁律（2026-09-17）：仅在有真实用户身份时注入——
|
||
# 无人值守/内部 agent 不挂 secret 工具，也就不给凭据使用引导。
|
||
if self.user_id:
|
||
try:
|
||
from .secret_vault import SECRET_PROMPT_BLOCK
|
||
prompt += SECRET_PROMPT_BLOCK
|
||
except Exception as e:
|
||
logger.warning(f"inject secret prompt block failed: {e}")
|
||
|
||
return prompt
|
||
|
||
# ═══════════════════════════════════════════════════════
|
||
# LLM 调用
|
||
# ═══════════════════════════════════════════════════════
|
||
|
||
async def _call_llm(self) -> dict:
|
||
"""调用 LLM，返回 {"content": str, "tool_calls": [...]}。
|
||
|
||
优先使用 OpenAI native function calling。若 registry 不可用或调用失败，
|
||
回退到纯文本调用（content 为原始文本，tool_calls 为空）。
|
||
"""
|
||
from pipeline_service.llm_bridge import llm_call_msgs_native
|
||
|
||
tools_schema = None
|
||
if self._tool_registry:
|
||
try:
|
||
tools_schema = self._tool_registry.to_openai_schema()
|
||
except Exception as e:
|
||
logger.error(f"to_openai_schema failed: {e}")
|
||
|
||
if tools_schema:
|
||
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,
|
||
session_id=self.session_id,
|
||
timeout=self._MAIN_TIMEOUT,
|
||
)
|
||
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,
|
||
session_id=self.session_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
|
||
try:
|
||
content = await llm_call_msgs(
|
||
self._msgs,
|
||
model=self.model_name,
|
||
temperature=self.config.temperature,
|
||
org_id=self.org_id,
|
||
session_id=self.session_id,
|
||
timeout=self._MAIN_TIMEOUT,
|
||
)
|
||
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,
|
||
session_id=self.session_id, timeout=self._MAIN_TIMEOUT)
|
||
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 输出
|
||
# ═══════════════════════════════════════════════════════
|
||
|
||
def _parse_actions(self, raw: str) -> List[dict]:
|
||
"""解析 LLM 输出，支持单个 JSON 或连续多个 JSON。
|
||
|
||
对照 HA：解析器成熟处理多 JSON + 容错。
|
||
"""
|
||
raw = (raw or "").strip()
|
||
|
||
# 去除 markdown 代码块包裹
|
||
if raw.startswith("```"):
|
||
parts = raw.split("\n", 1)
|
||
if len(parts) > 1:
|
||
raw = parts[1]
|
||
if raw.endswith("```"):
|
||
raw = raw[:-3]
|
||
raw = raw.strip()
|
||
|
||
results = []
|
||
|
||
# 尝试多 JSON 解析
|
||
# LLM 可能输出: {"action":"tool_call",...}\n{"action":"tool_call",...}
|
||
lines = raw.split("\n")
|
||
buffer = ""
|
||
depth = 0
|
||
|
||
for line in lines:
|
||
line_stripped = line.strip()
|
||
if not line_stripped:
|
||
if buffer:
|
||
# 尝试解析 buffer
|
||
result = self._try_parse_json(buffer)
|
||
if result:
|
||
results.append(result)
|
||
buffer = ""
|
||
continue
|
||
|
||
buffer += line_stripped
|
||
|
||
# 简单大括号计数
|
||
depth += line_stripped.count("{") - line_stripped.count("}")
|
||
|
||
if depth == 0 and buffer:
|
||
result = self._try_parse_json(buffer)
|
||
if result:
|
||
results.append(result)
|
||
buffer = ""
|
||
|
||
# 处理残留
|
||
if buffer:
|
||
result = self._try_parse_json(buffer)
|
||
if result:
|
||
results.append(result)
|
||
|
||
if not results:
|
||
# 完全无法解析 JSON → 当作纯文本回复
|
||
results.append({"action": "reply", "message": raw})
|
||
|
||
return results
|
||
|
||
def _try_parse_json(self, text: str) -> Optional[dict]:
|
||
"""容错 JSON 解析"""
|
||
text = text.strip()
|
||
try:
|
||
d = json.loads(text)
|
||
if isinstance(d, dict):
|
||
return d
|
||
except (json.JSONDecodeError, ValueError):
|
||
# 尝试提取第一个 JSON 对象
|
||
for start_char in ["{", "["]:
|
||
if start_char in text:
|
||
idx = text.index(start_char)
|
||
end_char = "}" if start_char == "{" else "]"
|
||
end_idx = text.rfind(end_char)
|
||
if end_idx > idx:
|
||
try:
|
||
d = json.loads(text[idx:end_idx + 1])
|
||
if isinstance(d, dict):
|
||
return d
|
||
except (json.JSONDecodeError, ValueError):
|
||
pass
|
||
return None
|
||
|
||
# ═══════════════════════════════════════════════════════
|
||
# 工具执行
|
||
# ═══════════════════════════════════════════════════════
|
||
|
||
async def _execute_tool(self, tool_name: str, params: dict) -> str:
|
||
"""执行工具，返回结果字符串。
|
||
|
||
优先级：
|
||
1. ToolRegistry 注册的 handler
|
||
2. SDLC 内建工具（兼容 cockpit_chat.dspy 的 TOOLS）
|
||
3. ask_user（始终可用）
|
||
"""
|
||
# 别名归一化：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)
|
||
if handler:
|
||
try:
|
||
result = await handler(params, {
|
||
"project_id": self.project_id,
|
||
"user_id": self.user_id,
|
||
"workspace_dir": self.workspace_dir,
|
||
})
|
||
return str(result)
|
||
except Exception as e:
|
||
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 会话不挂（作用域解析已排除）。
|
||
# ⚠️ 必须「名字被声明 且 TOOL_SCHEMAS 有 schema」双守卫（对齐 v1
|
||
# _exec_agent_tool 的 `if tool in TOOL_SCHEMAS`）：技能 frontmatter 声明的
|
||
# capability 工具（task 技能的 claim_task/reset_task_retry 等）多数不在
|
||
# TOOL_SCHEMAS（那是概念工具 schema 表），真正实现是产线能力包 handler
|
||
# （第 2 步 _execute_ability_tool）。此前只查名字 → 整族任务状态机工具被
|
||
# 本步劫持，返回「FAIL: 未注册的能力工具」，能力包 handler 永远到不了
|
||
# （2026-09-14 pbls 项目实测：主 agent 反复调 reset_task_retry 全 FAIL）。
|
||
if tool_name in self._capability_tool_names:
|
||
from .capability_tools import TOOL_SCHEMAS, exec_capability_tool
|
||
if tool_name in TOOL_SCHEMAS:
|
||
return await exec_capability_tool(tool_name, params, self._build_ctx())
|
||
# 声明了但无概念工具 schema → 落到第 2 步产线能力包分发（真实 handler 所在）
|
||
|
||
# 2. 产线能力包工具（从 PipelineAbility 注册表按 pipeline_id 取 handler）
|
||
result = await self._execute_ability_tool(tool_name, params)
|
||
if result is not None:
|
||
return result
|
||
|
||
# 3. 通用内建工具（core 内核：项目管理/终端/文件/搜索/会话/规划）
|
||
result = await self._execute_sdlc_tool(tool_name, params)
|
||
if result is not None:
|
||
return result
|
||
|
||
# 4. ask_user
|
||
if tool_name == "ask_user":
|
||
return f"QUESTION: {params.get('question', '')}"
|
||
|
||
return f"未知工具: {tool_name}"
|
||
|
||
def _build_ctx(self) -> dict:
|
||
"""构造传给产线能力 handler / slash 命令的上下文。"""
|
||
return {
|
||
"project_id": self.project_id,
|
||
"user_id": self.user_id,
|
||
"org_id": self.org_id,
|
||
"pipeline_id": self.pipeline_id,
|
||
"space": self.space,
|
||
"workspace_dir": self.workspace_dir,
|
||
"model_name": self.model_name,
|
||
"config": self.config,
|
||
# 会话级隔离键（2026-09-05）：能力 handler 的会话态缓存（如 extract 规格锚定）
|
||
# 必须按 session_id 隔离，防跨会话串数据
|
||
"session_id": self.session_id,
|
||
}
|
||
|
||
def _build_slash_ctx(self) -> dict:
|
||
"""构造传给 slash 命令 handler 的上下文（含 executor 引用 + role + base_url）。"""
|
||
ctx = self._build_ctx()
|
||
ctx["executor"] = self
|
||
ctx["role"] = self.role
|
||
ctx["base_url"] = getattr(self, "base_url", "") or ""
|
||
return ctx
|
||
|
||
async def _execute_ability_tool(self, tool_name: str, params: dict) -> Optional[str]:
|
||
"""产线能力包工具：按 pipeline_id 从 PipelineAbility 注册表取 handler 执行。"""
|
||
# 产线隔离（2026-09-05）：纯通用会话不挂任何产线能力包，执行层直接拒绝。
|
||
if self.generic:
|
||
return None
|
||
try:
|
||
from pipeline_core import get_ability, DEFAULT_ABILITY_ID
|
||
|
||
ability = get_ability(self.pipeline_id) or get_ability(DEFAULT_ABILITY_ID)
|
||
if not ability or tool_name not in ability.handlers:
|
||
return None
|
||
|
||
from sqlor.dbpools import DBPools
|
||
db = DBPools()
|
||
async with db.sqlorContext("pipeline") as sor:
|
||
return await ability.handlers[tool_name](sor, params, self._build_ctx())
|
||
except Exception as e:
|
||
logger.error(f"ability tool {tool_name} error: {e}")
|
||
return f"ERROR: {str(e)[:300]}"
|
||
|
||
async def _execute_sdlc_tool(self, tool_name: str, params: dict) -> Optional[str]:
|
||
"""通用内建工具（core 内核，产线无关）。"""
|
||
# 导入现有 handler
|
||
try:
|
||
from sqlor.dbpools import DBPools
|
||
|
||
db = DBPools()
|
||
async with db.sqlorContext("pipeline") as sor:
|
||
return await self._dispatch_sdlc_tool(sor, tool_name, params)
|
||
except Exception as e:
|
||
logger.error(f"SDLC tool {tool_name} error: {e}")
|
||
return f"ERROR: {str(e)[:300]}"
|
||
|
||
async def _dispatch_sdlc_tool(self, sor, tool_name: str, params: dict) -> str:
|
||
"""SDL 工具分发（兼容 cockpit_chat.dspy 的 16 个工具）"""
|
||
p = params or {}
|
||
pid = self.project_id
|
||
|
||
# 产线隔离（2026-09-05）：纯通用会话拒绝一切项目管理工具——此时
|
||
# tool_name 已经过 _PROJECT_TOOL_ALIASES 别名归一，别名变体同样被拦。
|
||
# schema 层（load_agent_config 剔除）+ 执行层（这里）双保险。
|
||
if self.generic and tool_name in _PROJECT_TOOL_NAMES:
|
||
return ("FAIL: 当前是通用助手会话（未挂产线插件），不提供项目管理能力，"
|
||
"无法查看或操作任何产线的项目。")
|
||
# 产线耦合工具拦截（2026-09-05 第 6 层）：run_command 的 shell 无法圈禁
|
||
# 在工作目录内，通用会话一律拒绝。
|
||
if self.generic and tool_name in _GENERIC_DENIED_TOOLS:
|
||
return ("FAIL: 当前是通用助手会话，不提供命令执行能力。"
|
||
"如需处理文件，可直接上传文件，我会读取上传内容作答。")
|
||
|
||
# 通用内建工具映射（core 内核，产线无关）。
|
||
# 产线专属工具（create_task/diagnose_project 等）已迁移到 sdlc_ability 能力包，
|
||
# 由 _execute_ability_tool 按 pipeline_id 动态挂载。
|
||
handlers = {
|
||
"switch_project": self._t_switch_project,
|
||
"create_project": self._t_create_project,
|
||
# ── 项目管理（终止/删除类代码层校验人类 owner）──
|
||
"project_info": self._t_project_info,
|
||
"list_my_projects": self._t_list_my_projects,
|
||
"pause_project": self._t_pause_project,
|
||
"resume_project": self._t_resume_project,
|
||
"delete_project": self._t_delete_project,
|
||
"run_command": self._t_run_command,
|
||
"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,
|
||
"propose_skill": self._t_propose_skill,
|
||
"write_file": self._t_write_file,
|
||
"list_files": self._t_list_files,
|
||
"search_files": self._t_search_files,
|
||
"session_search": self._t_session_search,
|
||
"memory": self._t_memory,
|
||
"manage_skill": self._t_manage_skill,
|
||
"todo": self._t_todo,
|
||
"delegate_subtask": self._t_delegate_subtask,
|
||
"subagent": self._t_subagent,
|
||
# ── 平台模型（2026-09-07：按任务自动选型 + 全能力调用）──
|
||
"list_platform_models": self._t_list_platform_models,
|
||
"invoke_model": self._t_invoke_model,
|
||
# ── 联网检索/网页抓取（2026-09-08，甲类只读能力，SSRF 防护在 web_tools）──
|
||
"web_search": self._t_web_search,
|
||
"fetch_url": self._t_fetch_url,
|
||
# ── 项目数据只读查询（2026-09-08，白名单+强制项目过滤在 db_query）──
|
||
"query_project_data": self._t_query_project_data,
|
||
# ── 用户敏感信息库（2026-09-17）：签名适配 (sor,params,ctx) ← (sor,p,pid) ──
|
||
# ctx 用 _build_ctx()（含真实 self.user_id/self.org_id）——secret 归属必须
|
||
# 是真实身份，不能用作用域解析里被 generic 置空的那份。
|
||
**{n: (lambda _n: (lambda s, _p, _pid: self._t_secret_tool(s, _n, _p)))(n)
|
||
for n in _SECRET_TOOL_NAMES},
|
||
}
|
||
|
||
handler = handlers.get(tool_name)
|
||
if handler:
|
||
return await handler(sor, p, pid)
|
||
return None
|
||
|
||
async def _t_secret_tool(self, sor, tool_name: str, params: dict) -> str:
|
||
"""敏感信息工具统一入口（2026-09-17）。
|
||
|
||
身份强制来自会话真实上下文（self.user_id / self.org_id），忽略 LLM 传的任何
|
||
归属参数——防越权读写他人 secret。无 user_id（无人值守/内部 agent）一律拒绝。
|
||
"""
|
||
from . import secret_vault
|
||
if not self.user_id:
|
||
return "FAIL: 当前会话没有用户身份，不提供敏感信息能力（防无人值守场景泄露凭据）"
|
||
h = secret_vault.SECRET_HANDLERS.get(tool_name)
|
||
if not h:
|
||
return f"FAIL: 未知敏感信息工具 {tool_name}"
|
||
try:
|
||
_r = await h(sor, params or {}, {"user_id": self.user_id,
|
||
"org_id": self.org_id or "",
|
||
"project_id": self.project_id})
|
||
# 写操作必须显式 COMMIT（多 worker 架构：不提交则其他进程读不到刚存的凭据）。
|
||
# 与本文件其他写库工具（_persist_project 等 6 处）同款做法；只读工具多跑一次
|
||
# COMMIT 无害（sqlor 事务幂等），省去按工具名区分读写。
|
||
try:
|
||
await sor.sqlExe("COMMIT", {})
|
||
except Exception as _ce:
|
||
logger.warning(f"secret tool {tool_name} commit failed: {_ce}")
|
||
return _r
|
||
except Exception as e:
|
||
logger.error(f"secret tool {tool_name} error: {e}")
|
||
return f"ERROR: {str(e)[:300]}"
|
||
|
||
# ── 工具实现 ──
|
||
|
||
async def _persist_project(self, sor, pid):
|
||
"""持久化当前项目。
|
||
|
||
- 有 session_id：写 pipeline_session_settings（web 多 tab 各自项目上下文，互不覆盖），
|
||
同时写全局 pipeline_agent_settings.current_project_id 作为「最近项目」兜底（供无 session 消费者）。
|
||
- 无 session_id：只写全局（向后兼容，如微信通道 / run_agent 直连）。
|
||
只改 self.project_id 不够——AgentExecutor 每轮新建，run 结束即销毁，
|
||
下一轮又从持久层读回旧项目。
|
||
"""
|
||
if not self.user_id or not pid:
|
||
return
|
||
try:
|
||
from appPublic.uniqueID import getID
|
||
# 全局「最近项目」兜底
|
||
await sor.sqlExe(
|
||
"UPDATE pipeline_agent_settings SET current_project_id=${pid}$ "
|
||
"WHERE user_id=${uid}$",
|
||
{"pid": pid, "uid": self.user_id})
|
||
exists = await sor.sqlExe(
|
||
"SELECT 1 FROM pipeline_agent_settings WHERE user_id=${uid}$",
|
||
{"uid": self.user_id})
|
||
if not exists:
|
||
await sor.sqlExe(
|
||
"INSERT INTO pipeline_agent_settings (id, user_id, current_project_id) "
|
||
"VALUES (${id}$, ${uid}$, ${pid}$)",
|
||
{"id": getID(), "uid": self.user_id, "pid": pid})
|
||
# 会话级项目上下文（多 tab 隔离）
|
||
if self.session_id:
|
||
try:
|
||
await sor.sqlExe(
|
||
"UPDATE pipeline_session_settings SET current_project_id=${pid}$ "
|
||
"WHERE user_id=${uid}$ AND session_id=${sid}$",
|
||
{"pid": pid, "uid": self.user_id, "sid": self.session_id})
|
||
sexists = await sor.sqlExe(
|
||
"SELECT 1 FROM pipeline_session_settings "
|
||
"WHERE user_id=${uid}$ AND session_id=${sid}$",
|
||
{"uid": self.user_id, "sid": self.session_id})
|
||
if not sexists:
|
||
await sor.sqlExe(
|
||
"INSERT INTO pipeline_session_settings "
|
||
"(id, user_id, session_id, current_project_id) "
|
||
"VALUES (${id}$, ${uid}$, ${sid}$, ${pid}$)",
|
||
{"id": getID(), "uid": self.user_id,
|
||
"sid": self.session_id, "pid": pid})
|
||
except Exception as e:
|
||
logger.warning(f"persist session project failed: {e}")
|
||
except Exception as e:
|
||
logger.warning(f"persist project failed: {e}")
|
||
|
||
async def _t_switch_project(self, sor, p, pid):
|
||
name = p.get("project_name", "").strip()
|
||
if not name:
|
||
return "需要项目名称"
|
||
|
||
# 精确匹配（限定当前项目空间 + 机构）
|
||
recs = await sor.sqlExe(
|
||
"SELECT id, name FROM sd_projects WHERE name=${n}$ AND pipeline_id=${s}$",
|
||
{"n": name, "s": self.space})
|
||
if recs:
|
||
r = recs[0]
|
||
self.project_id = getattr(r, "id", "")
|
||
await self._persist_project(sor, self.project_id)
|
||
await self._refresh_workspace_dir(sor)
|
||
return f"OK: 已切换到 {getattr(r, 'name', name)}"
|
||
|
||
# LLM 分类匹配
|
||
try:
|
||
from pipeline_service.llm_bridge import llm_call
|
||
|
||
all_recs = await sor.sqlExe(
|
||
"SELECT name FROM sd_projects WHERE pipeline_id=${s}$ ORDER BY created_at DESC LIMIT 20",
|
||
{"s": self.space})
|
||
pnames = [getattr(r, "name", "") for r in (all_recs or [])]
|
||
classify_prompt = f"用户输入: {name}\n项目列表: {', '.join(pnames)}\n\n判断用户想要哪个项目。只回复项目名或\"不存在\"。"
|
||
matched = await llm_call(classify_prompt, temperature=0.0, org_id=self.org_id,
|
||
purpose='utility', session_id=self.session_id)
|
||
matched = matched.strip().strip('"').strip("'")
|
||
|
||
recs2 = await sor.sqlExe(
|
||
"SELECT id, name FROM sd_projects WHERE name=${n}$ AND pipeline_id=${s}$",
|
||
{"n": matched, "s": self.space})
|
||
if recs2:
|
||
r = recs2[0]
|
||
self.project_id = getattr(r, "id", "")
|
||
await self._persist_project(sor, self.project_id)
|
||
await self._refresh_workspace_dir(sor)
|
||
return f"OK: 已切换到 {getattr(r, 'name', matched)}"
|
||
except Exception:
|
||
pass
|
||
|
||
return f"未找到项目: {name}"
|
||
|
||
async def _refresh_workspace_dir(self, sor):
|
||
"""切换项目后同步刷新工作目录（与上传落盘/初始化同一解析逻辑）。
|
||
|
||
不刷新则 list_files/read_file 仍停在切换前的目录（或无项目时的总根）。
|
||
"""
|
||
try:
|
||
from .workspace import get_project_dir_by_id
|
||
_pdir, _ = await get_project_dir_by_id(sor, self.project_id)
|
||
if _pdir:
|
||
import os as _os
|
||
_os.makedirs(_pdir, exist_ok=True)
|
||
self.workspace_dir = _pdir
|
||
except Exception:
|
||
pass
|
||
|
||
async def _t_create_project(self, sor, p, pid):
|
||
import os
|
||
import re
|
||
from appPublic.uniqueID import getID
|
||
|
||
name = p.get("name", "").strip()
|
||
desc = p.get("description", "")
|
||
if not name:
|
||
return "需要项目名称"
|
||
|
||
# 查当前用户 org_id（工作空间按 org 隔离）
|
||
org_id = "0"
|
||
if self.user_id:
|
||
_u = await sor.sqlExe(
|
||
"SELECT orgid FROM users WHERE id=${u}$ LIMIT 1", {"u": self.user_id})
|
||
if _u:
|
||
org_id = getattr(_u[0], "orgid", "0") or "0"
|
||
|
||
# 项目专属工作空间目录（统一机制：{base}/{org}/{space}/projects/{项目名}/，
|
||
# 用名字不用 id；创建时定死并写库，一切解析只读库不推导）。
|
||
from .workspace import alloc_project_dir
|
||
pid_val = getID()
|
||
project_dir, slug = await alloc_project_dir(sor, org_id, self.space, name, pid_val)
|
||
|
||
await sor.C("sd_projects", {
|
||
"id": pid_val, "name": name, "description": desc,
|
||
"status": "active", "org_id": org_id,
|
||
"pipeline_id": self.space, "workspace_dir": project_dir,
|
||
"directory_name": slug,
|
||
"created_by": self.user_id or "",
|
||
"default_model": self.model_name or "",
|
||
})
|
||
await sor.C("sd_iterations", {
|
||
"id": getID(), "project_id": pid_val,
|
||
"iteration_name": f"{name}-初始迭代",
|
||
"iteration_type": "default", "status": "in_progress", "priority": 1,
|
||
"seq_no": 1,
|
||
})
|
||
self.project_id = pid_val
|
||
await self._persist_project(sor, pid_val)
|
||
return f"OK: 已创建项目 {name}"
|
||
|
||
# ── 项目管理工具（2026-08-31 新增：会话 agent 获得项目管理能力）──
|
||
# 安全模型：终止（暂停）/删除等指令必须来自项目的人类 owner。
|
||
# 校验在代码层做（_resolve_owned_project 内调 check_project_owner），
|
||
# 不依赖 prompt——LLM 可以被注入/诱导，但校验函数只认数据库事实。
|
||
# owner 定义：sd_projects.created_by == 当前登录用户（self.user_id）。
|
||
|
||
async def _resolve_owned_project(self, sor, p):
|
||
"""解析目标项目并校验操作者是否为人类 owner。
|
||
|
||
project_name 缺省时用当前项目（self.project_id）；给了名称则在当前产线
|
||
空间内精确匹配。返回 (project_id, project_name, owner_err)：
|
||
owner_err 非空 = 无权限或项目不存在，调用方必须原样拒绝。
|
||
"""
|
||
from .project_capability import check_project_owner
|
||
name = (p.get("project_name") or "").strip()
|
||
if not name:
|
||
pid = self.project_id
|
||
if not pid:
|
||
return "", "", "当前会话没有选中项目，请先说明要操作哪个项目（或先切换项目）"
|
||
recs = await sor.sqlExe(
|
||
"SELECT id, name, created_by FROM sd_projects WHERE id=${pid}$",
|
||
{"pid": pid})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not recs:
|
||
return "", "", "当前项目已不存在"
|
||
pname = getattr(recs[0], "name", "") or pid
|
||
else:
|
||
recs = await sor.sqlExe(
|
||
"SELECT id, name, created_by FROM sd_projects "
|
||
"WHERE name=${n}$ AND pipeline_id=${s}$",
|
||
{"n": name, "s": self.space})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not recs:
|
||
return "", "", f"项目不存在：{name}（当前产线空间 {self.space}）"
|
||
pid = getattr(recs[0], "id", "")
|
||
pname = getattr(recs[0], "name", "") or pid
|
||
# 代码层 owner 校验：指令必须来自该项目的人类 owner（创建者）
|
||
ok, err = await check_project_owner(pid, self.user_id or "", sor)
|
||
if not ok:
|
||
return "", "", f"权限不足：{err}（项目「{pname}」的 owner 才可执行此操作）"
|
||
return pid, pname, ""
|
||
|
||
async def _t_project_info(self, sor, p, pid):
|
||
tpid, pname, err = await self._resolve_owned_project(sor, p)
|
||
if err:
|
||
return f"FAIL: {err}"
|
||
recs = await sor.sqlExe(
|
||
"SELECT id, name, description, status, pipeline_id, org_id, "
|
||
"created_by, created_at, updated_at FROM sd_projects WHERE id=${pid}$",
|
||
{"pid": tpid})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not recs:
|
||
return "FAIL: 项目不存在"
|
||
r = recs[0]
|
||
lines = [
|
||
f"项目：{getattr(r, 'name', '')}",
|
||
f"状态：{getattr(r, 'status', '')}",
|
||
f"描述：{getattr(r, 'description', '') or '（无）'}",
|
||
f"产线：{getattr(r, 'pipeline_id', '') or '（未绑定）'}",
|
||
f"创建：{getattr(r, 'created_at', '')} ｜ 更新：{getattr(r, 'updated_at', '')}",
|
||
]
|
||
# 迭代概况
|
||
iters = await sor.sqlExe(
|
||
"SELECT iteration_name, status FROM sd_iterations WHERE project_id=${pid}$ ORDER BY seq_no",
|
||
{"pid": tpid})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if iters:
|
||
lines.append("迭代：" + "；".join(
|
||
f"{getattr(i, 'iteration_name', '')}({getattr(i, 'status', '')})" for i in iters[:10]))
|
||
return "\n".join(lines)
|
||
|
||
async def _t_list_my_projects(self, sor, p, pid):
|
||
# 只列当前用户创建的项目（不暴露他人项目）
|
||
uid = self.user_id or ""
|
||
if not uid:
|
||
return "FAIL: 未登录，无法列出你的项目"
|
||
status = (p.get("status") or "").strip()
|
||
sql = ("SELECT id, name, status, pipeline_id, created_at FROM sd_projects "
|
||
"WHERE created_by=${uid}$")
|
||
params = {"uid": uid}
|
||
if status:
|
||
sql += " AND status=${st}$"
|
||
params["st"] = status
|
||
sql += " ORDER BY created_at DESC LIMIT 30"
|
||
recs = await sor.sqlExe(sql, params)
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not recs:
|
||
return f"你名下没有项目{('（状态=' + status + '）') if status else ''}"
|
||
lines = []
|
||
for r in recs:
|
||
lines.append(f"- {getattr(r, 'name', '')} ｜ {getattr(r, 'status', '')} ｜ "
|
||
f"产线 {getattr(r, 'pipeline_id', '') or '—'} ｜ {getattr(r, 'created_at', '')}")
|
||
return f"你创建的项目（{len(lines)} 个）：\n" + "\n".join(lines)
|
||
|
||
async def _t_pause_project(self, sor, p, pid):
|
||
tpid, pname, err = await self._resolve_owned_project(sor, p)
|
||
if err:
|
||
return f"FAIL: {err}"
|
||
from .project_capability import pause_project
|
||
ok, msg = await pause_project(tpid, who=self.user_id, agent_id="cockpit")
|
||
if not ok:
|
||
return f"FAIL: {msg}"
|
||
logger.info("cockpit pause_project: %s by user=%s", tpid, self.user_id)
|
||
return f"OK: 项目「{pname}」已暂停（终止推进）。需要时可用 resume_project 恢复。"
|
||
|
||
async def _t_resume_project(self, sor, p, pid):
|
||
tpid, pname, err = await self._resolve_owned_project(sor, p)
|
||
if err:
|
||
return f"FAIL: {err}"
|
||
from .project_capability import resume_project
|
||
ok, msg = await resume_project(tpid, who=self.user_id, agent_id="cockpit")
|
||
if not ok:
|
||
return f"FAIL: {msg}"
|
||
logger.info("cockpit resume_project: %s by user=%s", tpid, self.user_id)
|
||
return f"OK: 项目「{pname}」已恢复推进。"
|
||
|
||
async def _t_delete_project(self, sor, p, pid):
|
||
confirm = p.get("confirm")
|
||
if str(confirm).lower() not in ("true", "1", "yes"):
|
||
return ("FAIL: 参数错误，请直接带 confirm=true 调用（平台确认门会弹窗向用户最终确认，"
|
||
"无需你先向用户复述确认）。")
|
||
tpid, pname, err = await self._resolve_owned_project(sor, p)
|
||
if err:
|
||
return f"FAIL: {err}"
|
||
# 删除后清掉本会话的项目上下文（否则留下悬空指针，下次打开报「请先切换项目」）
|
||
from .project_capability import delete_project
|
||
ok, msg = await delete_project(tpid, who=self.user_id, agent_id="cockpit", confirm=True)
|
||
if not ok:
|
||
return f"FAIL: {msg}"
|
||
if self.project_id == tpid:
|
||
self.project_id = ""
|
||
# 清会话级+全局「当前项目」指针（_persist_project 对空 pid 直接 return，
|
||
# 这里直接清，否则留下悬空指针，下次打开报「请先切换项目」）
|
||
try:
|
||
await sor.sqlExe(
|
||
"UPDATE pipeline_session_settings SET current_project_id='' "
|
||
"WHERE user_id=${uid}$ AND session_id=${sid}$",
|
||
{"uid": self.user_id or "", "sid": self.session_id or ""})
|
||
await sor.sqlExe(
|
||
"UPDATE pipeline_agent_settings SET current_project_id='' "
|
||
"WHERE user_id=${uid}$",
|
||
{"uid": self.user_id or ""})
|
||
await sor.sqlExe("COMMIT", {})
|
||
except Exception as e:
|
||
logger.warning(f"clear project pointer after delete failed: {e}")
|
||
logger.info("cockpit delete_project: %s by user=%s", tpid, self.user_id)
|
||
return f"OK: 项目「{pname}」已删除（含归档备份）。{msg}"
|
||
|
||
async def _t_run_command(self, sor, p, pid):
|
||
cmd = p.get("command", "")
|
||
if not cmd:
|
||
return "需要命令"
|
||
|
||
try:
|
||
from pipeline_service.agent_loop import _run_shell, _find_bwrap
|
||
|
||
# 通用会话（generic）strict 沙箱档（2026-09-08 乙类开放）：
|
||
# bwrap 圈死——平台目录（/d/pipeline、/d/doit）完全不挂载（不可见），
|
||
# 可写根 = 用户自己的 _general/{uid} 目录。旧禁令的理由「shell 无法
|
||
# 圈禁在工作目录内、可枚举全工作空间」已被 strict 档根治（连读都不可见）。
|
||
# 无 bwrap 时 generic 一律拒绝（绝不降级为裸 shell——安全优先）。
|
||
if self.generic and not _find_bwrap():
|
||
return ("FAIL: 通用会话的命令执行需要 bwrap 沙箱（当前服务器不可用），已拒绝执行。"
|
||
"文件类操作请改用 read_file/write_file/list_files/search_files。")
|
||
|
||
# 敏感信息执行边界（2026-09-17）：占位符 @@sec:NAME@@ → $PIPELINE_SEC_NAME，
|
||
# 真值经 env 注入子进程，**不进命令字符串**（命令可安全落库/进上下文）。
|
||
# 未知占位符原样保留（绝不替换成空串——空串会静默改变命令语义）。
|
||
# 必须在前台/后台分支**之前**处理，两条执行路径才都受保护。
|
||
secret_env = None
|
||
try:
|
||
from . import secret_vault
|
||
cmd, secret_env, unknown = await secret_vault.prepare_command(
|
||
sor, cmd, org_id=self.org_id or "", user_id=self.user_id or "")
|
||
if unknown:
|
||
return ("FAIL: 命令引用了不存在的敏感信息 " + ", ".join(unknown) +
|
||
"（已拒绝执行，防止空值改变命令语义）。用 list_secrets 核对名称。")
|
||
except Exception as e:
|
||
logger.warning(f"secret prepare_command failed (run without secrets): {e}")
|
||
|
||
# PII 执行边界（2026-09-17）：@@pii:占位符 → $PIPELINE_PII_*，真值经 env 注入，
|
||
# 与凭据同构；缓存缺失时引用保留（命令取空，不静默改语义——与凭据不同：PII
|
||
# 缺失是过期常态，拒绝执行会误伤，取空+输出可见更安全）。
|
||
pii_env = None
|
||
if self.pii_session_key and "@@pii:" in cmd:
|
||
try:
|
||
from . import pii_guard
|
||
cmd, pii_env = await pii_guard.prepare_command(
|
||
cmd, self.pii_session_key)
|
||
except Exception as e:
|
||
logger.warning(f"pii prepare_command failed (run without pii env): {e}")
|
||
if pii_env:
|
||
secret_env = {**(secret_env or {}), **pii_env}
|
||
|
||
# 后台执行（2026-09-10，对齐 Hermes terminal background）：
|
||
# 状态文件化到 workspace/.bg/，跨 worker 进程可 poll；
|
||
# 沙箱档位与前台完全一致（generic 强制 strict，无 bwrap 上面已拒）。
|
||
bg = str(p.get("background", "")).strip().lower()
|
||
if bg in ("true", "1", "yes"):
|
||
from . import bg_jobs
|
||
try:
|
||
job_id = await bg_jobs.start_bg_job(
|
||
cmd, self.workspace_dir, strict=self.generic,
|
||
secret_env=secret_env)
|
||
except bg_jobs.BgJobError as e:
|
||
return f"FAIL: {e}"
|
||
return (f"OK: 后台任务已启动 job_id={job_id}。"
|
||
f"用 process(action=poll|log|wait|kill, job_id={job_id}) 跟进输出与状态。")
|
||
|
||
try:
|
||
timeout = int(p.get("timeout") or 60)
|
||
except (TypeError, ValueError):
|
||
timeout = 60
|
||
timeout = max(5, min(timeout, 300))
|
||
|
||
r = await _run_shell(cmd, self.workspace_dir, timeout=timeout,
|
||
strict=self.generic, secret_env=secret_env)
|
||
out = f"rc={r['rc']}\n{r['stdout'][:2000]}"
|
||
if r.get('stderr'):
|
||
out += f"\nSTDERR: {r['stderr'][:500]}"
|
||
if self.generic and not r.get('sandbox'):
|
||
out += "\n（注意：本次未经过沙箱）"
|
||
# 出站净化（需求2的「执行返回当输入处理」）：工具输出里若回显了
|
||
# 已入库凭据明文（报错信息常见），擦洗成占位符再回填模型上下文。
|
||
try:
|
||
from . import secret_vault
|
||
out, scrubbed = await secret_vault.sanitize_tool_output(
|
||
sor, out, org_id=self.org_id or "", user_id=self.user_id or "")
|
||
if scrubbed:
|
||
out += "\n（输出中出现的敏感信息已替换为占位符：" + ", ".join(scrubbed) + "）"
|
||
except Exception as e:
|
||
logger.warning(f"sanitize_tool_output failed: {e}")
|
||
return out
|
||
except Exception as e:
|
||
return f"ERROR: {str(e)[:300]}"
|
||
|
||
async def _t_process(self, sor, p, pid):
|
||
"""后台任务管理入口：先执行，再对输出做出站净化（2026-09-17）。
|
||
|
||
后台命令可能把凭据回显进 output.log（curl -v、报错信息常见），
|
||
读回模型上下文前擦洗成占位符——与前台 run_command 同一条防线。
|
||
"""
|
||
out = await self._t_process_raw(sor, p, pid)
|
||
try:
|
||
from . import secret_vault
|
||
out, scrubbed = await secret_vault.sanitize_tool_output(
|
||
sor, out, org_id=self.org_id or "", user_id=self.user_id or "")
|
||
if scrubbed:
|
||
out += "\n（输出中出现的敏感信息已替换为占位符：" + ", ".join(scrubbed) + "）"
|
||
except Exception as e:
|
||
logger.warning(f"sanitize process output failed: {e}")
|
||
return out
|
||
|
||
async def _t_process_raw(self, sor, p, pid):
|
||
"""后台任务管理（poll/log/wait/kill，配套 run_command background=true）。
|
||
|
||
隔离：只解析 self.workspace_dir 下的 .bg/——通用会话 workspace 是
|
||
_general/{uid}、项目会话是项目目录，跨用户/跨项目天然不可达；
|
||
job_id 白名单正则防路径穿越（bg_jobs 内校验）。
|
||
"""
|
||
from . import bg_jobs
|
||
action = (p.get("action") or "").strip().lower()
|
||
job_id = (p.get("job_id") or "").strip()
|
||
if action not in ("poll", "log", "wait", "kill"):
|
||
return "FAIL: action 须为 poll|log|wait|kill"
|
||
if not job_id:
|
||
return "FAIL: 需要 job_id（run_command background=true 的返回值）"
|
||
try:
|
||
offset = int(p.get("offset") or 0)
|
||
except (TypeError, ValueError):
|
||
offset = 0
|
||
try:
|
||
if action == "poll":
|
||
r = bg_jobs.poll_job(self.workspace_dir, job_id, offset)
|
||
m = r["meta"]
|
||
out = (f"job_id={job_id} status={m['status']} rc={m.get('rc')} "
|
||
f"sandbox={m.get('sandbox')} log_size={r['log_size']}")
|
||
if r["new_output"]:
|
||
out += f"\n--- 新增输出 ---\n{r['new_output'][-4000:]}"
|
||
return out
|
||
if action == "log":
|
||
r = bg_jobs.read_log(self.workspace_dir, job_id, offset)
|
||
m = r["meta"]
|
||
more = f"（还有后续，offset={offset + len(r['content'])} 续读）" if r["truncated"] else ""
|
||
return (f"job_id={job_id} status={m['status']} rc={m.get('rc')}\n"
|
||
f"--- 输出 ---\n{r['content']}{more}")
|
||
if action == "wait":
|
||
r = await bg_jobs.wait_job(self.workspace_dir, job_id)
|
||
m = r["meta"]
|
||
out = f"job_id={job_id} status={m['status']} rc={m.get('rc')}"
|
||
if m["status"] == "running":
|
||
out += "（等待超时，任务仍在跑，可继续 wait 或 poll）"
|
||
if r["new_output"]:
|
||
out += f"\n--- 输出（尾部） ---\n{r['new_output'][-4000:]}"
|
||
return out
|
||
if action == "kill":
|
||
r = bg_jobs.kill_job(self.workspace_dir, job_id)
|
||
return f"OK: {r['message']}（job_id={job_id}）"
|
||
except bg_jobs.BgJobError as e:
|
||
return f"FAIL: {e}"
|
||
except Exception as e:
|
||
return f"ERROR: {str(e)[:300]}"
|
||
return "FAIL: 未知分支"
|
||
|
||
def _resolve_ws_path(self, path: str) -> str:
|
||
"""解析相对路径为工作空间内绝对路径（越界返回 ''）。"""
|
||
import os
|
||
ws = self.workspace_dir or WORKSPACE_BASE
|
||
full = os.path.abspath(os.path.join(ws, path or "."))
|
||
# 限制在 workspace 内
|
||
if full != ws and not full.startswith(ws.rstrip("/") + "/"):
|
||
return ""
|
||
return full
|
||
|
||
async def _t_read_file(self, sor, p, pid):
|
||
path = p.get("path", "")
|
||
if not path:
|
||
return "FAIL: 需要文件路径"
|
||
full = self._resolve_ws_path(path)
|
||
if not full:
|
||
return f"FAIL: 路径越界 {path}"
|
||
# 统一文件读取（file_read 共享模块，v1/v2 共用）：分页续读 + docx/pdf 解析 + 显式截断告知
|
||
from .file_read import read_text_file, DEFAULT_LIMIT
|
||
try:
|
||
offset = int(p.get("offset") or 0)
|
||
except (ValueError, TypeError):
|
||
offset = 0
|
||
r = read_text_file(full, offset=offset, limit=DEFAULT_LIMIT)
|
||
if r['kind'] == 'error':
|
||
return f"FAIL: {r['message']} {path}"
|
||
if r['kind'] == 'binary':
|
||
return r['message']
|
||
return r['content']
|
||
|
||
async def _t_load_skill(self, sor, p, pid):
|
||
name = (p.get("name") or "").strip()
|
||
if not name:
|
||
return "FAIL: 需要技能名称"
|
||
if not self._skill_loader:
|
||
return "FAIL: 技能系统未初始化"
|
||
merged = self._skill_loader.get_merged(
|
||
pipeline_id=self.pipeline_id or "",
|
||
role=self.role or "",
|
||
project_id=self.project_id or "",
|
||
org_id=self.org_id or "",
|
||
user_id=self.user_id or "")
|
||
from pipeline_core.skill_loader import resolve_skill, format_skill_names
|
||
skill = resolve_skill(merged, name)
|
||
if skill is None:
|
||
names = format_skill_names(merged)
|
||
return (f"FAIL: 技能 '{name}' 不存在。注意：name 只传技能裸名"
|
||
f"（如 bid-workflow），不要带 [产线] 前缀或描述文字。可用技能: {names}")
|
||
file_path = (p.get("file_path") or "").strip()
|
||
if file_path:
|
||
body = skill.read_linked_file(file_path)
|
||
if body.startswith("FAIL") or body.startswith("ERROR"):
|
||
linked = skill.list_linked_files()
|
||
return f"{body}\n可用子文件: {', '.join(linked) if linked else '（无）'}"
|
||
return f"## [{skill.name}] {file_path}\n\n{body}"
|
||
block = skill.to_prompt_block()
|
||
linked = skill.list_linked_files()
|
||
if linked:
|
||
block += f"\n（可用子文件，用 file_path 参数加载：{', '.join(linked)}）\n"
|
||
return block
|
||
|
||
async def _t_list_platform_models(self, sor, p, pid):
|
||
"""列出平台可用模型（薄壳委托 platform_model_tools 唯一实现）。"""
|
||
from .platform_model_tools import tool_list_platform_models
|
||
return await tool_list_platform_models(p, self.org_id or "0")
|
||
|
||
async def _t_invoke_model(self, sor, p, pid):
|
||
"""调用平台模型完成生成类任务（薄壳委托 platform_model_tools 唯一实现）。"""
|
||
from .platform_model_tools import tool_invoke_model
|
||
return await tool_invoke_model(
|
||
p, self.org_id or "0", user_id=self.user_id or "",
|
||
project_id=self.project_id or "", session_id=self.session_id or "")
|
||
|
||
async def _t_web_search(self, sor, p, pid):
|
||
"""联网检索（薄壳委托 web_tools 唯一实现，SSRF 防护在其中）。"""
|
||
from .web_tools import tool_web_search
|
||
return await tool_web_search(p.get("query", ""), p.get("limit", 8))
|
||
|
||
async def _t_fetch_url(self, sor, p, pid):
|
||
"""抓取网页正文（超长落盘工作空间 webcache/，read_file 续读）。"""
|
||
from .web_tools import tool_fetch_url
|
||
return await tool_fetch_url(p.get("url", ""), workspace_dir=self.workspace_dir or "")
|
||
|
||
async def _t_query_project_data(self, sor, p, pid):
|
||
"""查询项目关联表数据（薄壳委托 db_query 唯一实现，白名单/过滤/审计在其中）。"""
|
||
if self.generic or not pid:
|
||
return "FAIL: 当前会话无项目上下文，无法查询项目数据。"
|
||
from .db_query import tool_query_project_data
|
||
return await tool_query_project_data(
|
||
sor, p.get("table", ""), pid,
|
||
where=p.get("where", ""), order_by=p.get("order_by", ""),
|
||
limit=p.get("limit", 100), who="agent.main_agent")
|
||
|
||
async def _t_list_packs(self, sor, p, pid):
|
||
try:
|
||
from pipeline_core.skill_pack import list_packs
|
||
packs = list_packs()
|
||
if not packs:
|
||
return "（无可安装的技能集）"
|
||
lines = ["可安装的技能集:"]
|
||
for pk in packs:
|
||
lines.append(f" - {pk['name']}: {pk.get('title', '')}（{pk.get('skill_count', 0)} 个技能）")
|
||
return "\n".join(lines)
|
||
except Exception as e:
|
||
return f"ERROR: {str(e)[:300]}"
|
||
|
||
async def _t_install_pack(self, sor, p, pid):
|
||
pack = (p.get("pack") or "").strip()
|
||
if not pack:
|
||
return "FAIL: 需要技能集名（先用 list_packs 查看可用的技能集名）"
|
||
try:
|
||
from pipeline_core.skill_pack import install_pack
|
||
base_dir = self._skill_loader.base_dir if self._skill_loader else ""
|
||
org_id = self.org_id or "0"
|
||
r = install_pack(pack, base_dir, org_id)
|
||
if r.get("success"):
|
||
if self._skill_loader:
|
||
self._skill_loader.reload()
|
||
installed = r.get("installed", [])
|
||
return f"OK: 已安装技能集 {pack}（{len(installed)} 个技能）"
|
||
return f"FAIL: {r.get('error', '未知错误')}"
|
||
except Exception as e:
|
||
return f"ERROR: {str(e)[:300]}"
|
||
|
||
async def _t_propose_skill(self, sor, p, pid):
|
||
name = (p.get("name") or "").strip()
|
||
description = (p.get("description") or "").strip()
|
||
content = (p.get("content") or "").strip()
|
||
if not name or not content:
|
||
return "FAIL: 需要技能名和内容（content 为 SKILL.md 草稿正文）"
|
||
try:
|
||
from appPublic.uniqueID import getID
|
||
# 实时发布（2026-09-08 用户拍板）：写提议者所属机构技能目录
|
||
# skills/orgs/{org_id}/{name}/，org scope 同名覆盖 global，
|
||
# 机构内立即生效、其他机构不受影响；org 0 有用户身份时降级
|
||
# users/{user_id}/（只影响本人），org 0 无人值守拒绝（防影响全平台）。
|
||
from .skill_live import publish_skill_live
|
||
ok, msg, info = await publish_skill_live(
|
||
name, description, content,
|
||
org_id=self.org_id or "", user_id=self.user_id or "",
|
||
who=self.user_id or "")
|
||
await sor.C("skill_proposals", {
|
||
"id": getID(),
|
||
"name": name,
|
||
"description": description,
|
||
"content": content,
|
||
"source": "agent",
|
||
# 实时发布成功=published（审计痕迹）；被拒=保持 pending 人工审核链
|
||
"status": "published" if ok else "pending",
|
||
"feedback": msg if ok else "",
|
||
"org_id": self.org_id or "",
|
||
"pipeline_id": self.pipeline_id or "",
|
||
"created_by": self.user_id or "",
|
||
})
|
||
if ok:
|
||
return msg
|
||
# 实时发布被拒（org 0 无人值守等）→ 保持旧行为：提议待审核
|
||
return f"OK: 已提交技能提议 '{name}'（{msg}，转人工审核，暂不生效）"
|
||
except Exception as e:
|
||
return f"ERROR: {str(e)[:300]}"
|
||
|
||
async def _t_write_file(self, sor, p, pid):
|
||
path = p.get("path", "")
|
||
content = p.get("content", "")
|
||
if not path:
|
||
return "FAIL: 需要文件路径"
|
||
full = self._resolve_ws_path(path)
|
||
if not full:
|
||
return f"FAIL: 路径越界 {path}"
|
||
try:
|
||
os.makedirs(os.path.dirname(full), exist_ok=True)
|
||
with open(full, "w", encoding="utf-8") as f:
|
||
f.write(content or "")
|
||
return f"OK: 已写入 {path} ({len(content)} 字符)"
|
||
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)
|
||
if not full:
|
||
return f"FAIL: 路径越界 {path}"
|
||
try:
|
||
if not os.path.isdir(full):
|
||
return f"FAIL: 目录不存在 {path}"
|
||
items = sorted(os.listdir(full))[:50]
|
||
lines = []
|
||
for name in items:
|
||
fp = os.path.join(full, name)
|
||
if name.startswith("."):
|
||
continue
|
||
t = "DIR" if os.path.isdir(fp) else "FILE"
|
||
size = os.path.getsize(fp) if os.path.isfile(fp) else 0
|
||
lines.append(f"[{t}] {name} ({size}B)")
|
||
return "\n".join(lines) if lines else "(空目录)"
|
||
except Exception as e:
|
||
return f"ERROR: {str(e)[:300]}"
|
||
|
||
async def _t_search_files(self, sor, p, pid):
|
||
pattern = p.get("pattern", "")
|
||
if not pattern:
|
||
return "FAIL: 需要搜索关键词"
|
||
path = p.get("path", "") or "."
|
||
full = self._resolve_ws_path(path)
|
||
if not full:
|
||
return f"FAIL: 路径越界 {path}"
|
||
try:
|
||
from pipeline_service.agent_loop import _run_shell
|
||
# grep -rn，排除 .git 和 __pycache__，限制输出
|
||
r = await _run_shell(
|
||
f"grep -rn --include='*.py' --include='*.md' --include='*.json' --include='*.txt' "
|
||
f"--exclude-dir=.git --exclude-dir=__pycache__ '{pattern}' . 2>/dev/null | head -50",
|
||
full, timeout=30)
|
||
out = r.get("stdout", "").strip()
|
||
return out[:4000] if out else f"未找到匹配 '{pattern}' 的内容"
|
||
except Exception as e:
|
||
return f"ERROR: {str(e)[:300]}"
|
||
|
||
async def _t_session_search(self, sor, p, pid):
|
||
query = p.get("query", "").strip()
|
||
if not query:
|
||
return "FAIL: 需要搜索关键词"
|
||
try:
|
||
# 产线隔离（2026-09-05 第7层）：通用会话只搜自己的历史
|
||
# （pipeline_id=''），与 _load_history 过滤对称；产线会话按
|
||
# iteration_id（项目）过滤不变。
|
||
if self.generic:
|
||
sql = ("SELECT role, content, created_at FROM pipeline_conversations "
|
||
"WHERE created_by=${u}$ AND (pipeline_id='' OR pipeline_id IS NULL) "
|
||
"AND content LIKE ${q}$ "
|
||
"ORDER BY created_at DESC LIMIT 10")
|
||
else:
|
||
sql = ("SELECT role, content, created_at FROM pipeline_conversations "
|
||
"WHERE created_by=${u}$ AND iteration_id=${pid}$ AND content LIKE ${q}$ "
|
||
"ORDER BY created_at DESC LIMIT 10")
|
||
recs = await sor.sqlExe(sql,
|
||
{"u": self.user_id, "pid": pid, "q": f"%{query}%"})
|
||
if not recs:
|
||
return f"未找到包含 '{query}' 的会话记录"
|
||
lines = []
|
||
for r in recs:
|
||
content = (getattr(r, "content", "") or "")[:300]
|
||
lines.append(f"[{getattr(r, 'role', '?')}] {content}")
|
||
return "\n".join(lines)
|
||
except Exception as e:
|
||
return f"ERROR: {str(e)[:300]}"
|
||
|
||
async def _t_memory(self, sor, p, pid):
|
||
"""持久记忆工具（add/list/remove，2026-09-10 对齐 Hermes memory）。
|
||
|
||
多租户写入门禁（服务多机构多用户，全部代码层强制，不靠 LLM 自觉）：
|
||
- org_id/user_id 一律用会话真实身份注入，忽略 LLM 传值（防伪造归属）
|
||
- scope 白名单：user/project/pipeline——global/org 是平台种子域，禁写
|
||
（global 域所有人可见，agent 写入=跨机构泄漏；org 域是平台运营配置）
|
||
- scope=project 必须有当前项目；scope=pipeline 的 scope_id 强制用
|
||
当前产线（防写进他产线域）
|
||
- remove 只能删「本机构 org_id + 本人或本机构名下」的条目
|
||
（store.remove 带 org/user 过滤），种子与他租户条目删不到
|
||
- 无 user_id（无人值守场景）拒绝写入——归属不明的记忆一律不收
|
||
"""
|
||
if not self._memory_store:
|
||
return "FAIL: 记忆系统未初始化"
|
||
from pipeline_core.memory_store import MemoryStore
|
||
action = (p.get("action") or "").strip().lower()
|
||
store = self._memory_store
|
||
org_id = self.org_id or ""
|
||
user_id = self.user_id or ""
|
||
|
||
if action == "add":
|
||
content = (p.get("content") or "").strip()
|
||
if not content:
|
||
return "FAIL: 需要 content（记忆内容）"
|
||
if not user_id:
|
||
return "FAIL: 无用户身份的会话不允许写记忆（归属不明）"
|
||
scope = (p.get("scope") or "user").strip().lower()
|
||
if scope not in MemoryStore.WRITABLE_SCOPES:
|
||
return ("FAIL: scope 只允许 user/project/pipeline"
|
||
"（global/org 是平台种子域，agent 禁写）")
|
||
category = (p.get("category") or "memory").strip().lower()
|
||
if category not in ("user", "memory"):
|
||
category = "memory"
|
||
scope_id = ""
|
||
if scope == "project":
|
||
if not self.project_id:
|
||
return "FAIL: 当前会话没有项目上下文，scope=project 不可用（改用 user 或 pipeline）"
|
||
scope_id = self.project_id
|
||
elif scope == "pipeline":
|
||
if self.generic or not self.pipeline_id:
|
||
return "FAIL: 通用会话没有产线上下文，scope=pipeline 不可用（改用 user）"
|
||
scope_id = self.pipeline_id
|
||
elif scope == "user":
|
||
scope_id = user_id
|
||
priority = MemoryStore.PRIORITY_HIGH if category == "user" else MemoryStore.PRIORITY_MEDIUM
|
||
# user_id 一律记创建者（所有 scope）：user 域据此仅本人可见；
|
||
# project/pipeline 域可见性仍由 org_id 控制（机构内共享），
|
||
# 但删除时按 (org_id+user_id) 过滤=只能删本人写入的，
|
||
# 防同机构他人误删共享条目
|
||
await store.add(content=content, category=category, priority=priority,
|
||
scope=scope, scope_id=scope_id,
|
||
org_id=org_id, user_id=user_id)
|
||
scope_cn = {"user": "个人偏好", "project": "当前项目", "pipeline": "当前产线"}[scope]
|
||
return f"OK: 已记住（{scope_cn}域，跨会话生效）"
|
||
|
||
if action == "list":
|
||
entries = await store.get_visible(org_id, user_id,
|
||
pipeline_id="" if self.generic else self.pipeline_id,
|
||
project_id=self.project_id)
|
||
if not entries:
|
||
return "(暂无可见记忆)"
|
||
lines = []
|
||
for e in entries[:30]:
|
||
lines.append(f"- [{e.scope}] key={e.key}: {e.content[:80]}")
|
||
if len(entries) > 30:
|
||
lines.append(f"（共 {len(entries)} 条，仅显示前 30）")
|
||
return "\n".join(lines)
|
||
|
||
if action == "remove":
|
||
key = (p.get("key") or "").strip()
|
||
if not key:
|
||
return "FAIL: remove 需要 key（先用 action=list 查）"
|
||
# 过滤 (org_id + 本人 user_id)：只能删本人写入的条目。
|
||
# 种子条目 org_id=''、他机构条目 org_id≠本机构、同机构他人条目
|
||
# user_id≠本人，全部够不着；key 掺了租户维度防跨租户误删。
|
||
removed = 0
|
||
for cat in ("memory", "user"):
|
||
removed += await store.remove(key, cat, org_id=org_id, user_id=user_id)
|
||
if removed:
|
||
return f"OK: 已删除记忆 {key}"
|
||
return f"FAIL: 未找到可删除的记忆 {key}（只能删本机构/本人的条目）"
|
||
|
||
return "FAIL: action 须为 add|list|remove"
|
||
|
||
async def _t_manage_skill(self, sor, p, pid):
|
||
"""技能管理（create/patch/write_file/remove_file/delete，2026-09-10）。
|
||
|
||
隔离全部由 skill_live.manage_skill_live 把关：只能落本机构 orgs/{org}/
|
||
或 users/{uid}/ 目录；global 原版走 fork-on-write 继承副本；产线/角色/
|
||
项目层技能拒绝修改；delete 只删本租户副本。org_id/user_id 用会话真实
|
||
身份注入，忽略 LLM 传值。
|
||
"""
|
||
from .skill_live import manage_skill_live
|
||
action = (p.get("action") or "").strip().lower()
|
||
name = (p.get("name") or "").strip()
|
||
if not action:
|
||
return "FAIL: 需要 action（create/patch/write_file/remove_file/delete）"
|
||
ok, msg, info = await manage_skill_live(
|
||
action=action, name=name,
|
||
org_id=self.org_id or "", user_id=self.user_id or "",
|
||
who=self.user_id or "",
|
||
description=(p.get("description") or "").strip(),
|
||
content=p.get("content") or "",
|
||
old_string=p.get("old_string") or "",
|
||
new_string=p.get("new_string") or "",
|
||
file_path=(p.get("file_path") or "").strip(),
|
||
file_content=p.get("file_content") or "",
|
||
pipeline_id="" if self.generic else (self.pipeline_id or ""),
|
||
role=self.role or "",
|
||
project_id=self.project_id or "",
|
||
)
|
||
if ok:
|
||
# 审计：技能增改删留痕（append-only）
|
||
try:
|
||
from .audit import record_audit
|
||
await record_audit(
|
||
tenant_id=self.org_id or "0",
|
||
entity="skills", entity_id=name,
|
||
action=f"skill_{action}", who=self.user_id or "agent",
|
||
detail=str(info)[:500], sor=sor)
|
||
except Exception as e:
|
||
logger.warning(f"manage_skill audit failed: {e}")
|
||
return msg if ok else f"FAIL: {msg}"
|
||
|
||
async def _t_todo(self, sor, p, pid):
|
||
action = p.get("action", "list")
|
||
content = (p.get("content", "") or "").strip()
|
||
if action == "add":
|
||
if not content:
|
||
return "FAIL: add 需要任务内容"
|
||
self._todos.append({"done": False, "content": content})
|
||
return f"OK: 已添加任务（共 {len(self._todos)} 项）"
|
||
if action == "done":
|
||
if not content:
|
||
return "FAIL: done 需要任务序号或内容"
|
||
for t in self._todos:
|
||
if content in t["content"] or content == str(self._todos.index(t) + 1):
|
||
t["done"] = True
|
||
return f"OK: 已完成任务 '{t['content']}'"
|
||
return f"未找到任务 '{content}'"
|
||
# list（默认）
|
||
if not self._todos:
|
||
return "任务清单为空"
|
||
lines = []
|
||
for i, t in enumerate(self._todos):
|
||
mark = "✅" if t["done"] else "⏳"
|
||
lines.append(f"{i + 1}. {mark} {t['content']}")
|
||
return "\n".join(lines)
|
||
|
||
async def _t_delegate_subtask(self, sor, p, pid):
|
||
"""委派子 agent（2026-09-10 升级，对齐 Hermes delegate_task）。
|
||
|
||
background=false（默认）：同步等待，行为同旧版（兼容存量用法）。
|
||
background=true：后台并行（同 workspace 上限 3 个），立即返回
|
||
subagent_id，用 subagent 工具 list/steer/stop/result 跟进。
|
||
|
||
隔离（两种模式一致）：
|
||
- 子 executor session_isolation='none'：不读父历史、不写
|
||
pipeline_conversations（子任务对话不污染父会话回放）
|
||
- 子 agent 不继承 session_id（多 tab 项目上下文不串）
|
||
- 深度限制：子 agent（_delegate_depth≥1）不能再委派，防递归爆炸
|
||
"""
|
||
goal = (p.get("goal", "") or "").strip()
|
||
context = (p.get("context", "") or "").strip()
|
||
if not goal:
|
||
return "FAIL: 需要子任务目标"
|
||
bg = str(p.get("background", "")).strip().lower() in ("true", "1", "yes")
|
||
|
||
if bg:
|
||
from . import subagents
|
||
try:
|
||
sid = await subagents.spawn_subagent(self, goal, context)
|
||
except subagents.SubagentError as e:
|
||
return f"FAIL: {e}"
|
||
return (f"OK: 子agent已后台启动 subagent_id={sid}。"
|
||
f"用 subagent(action=list) 查状态、steer 追加指示、"
|
||
f"stop 终止、result 取结果（完成前 result 是中间输出）。")
|
||
|
||
# 同步模式（深度限制同样生效）
|
||
from . import subagents
|
||
if self._delegate_depth >= subagents.MAX_DEPTH:
|
||
return "FAIL: 子agent不能再委派子任务（深度限制，防递归爆炸）"
|
||
try:
|
||
from dataclasses import replace as _dc_replace
|
||
child_config = _dc_replace(self.config, session_isolation="none")
|
||
sub = AgentExecutor(
|
||
config=child_config,
|
||
project_id=self.project_id,
|
||
user_id=self.user_id,
|
||
workspace_dir=self.workspace_dir,
|
||
model_name=self.model_name,
|
||
session_id="",
|
||
generic=self.generic,
|
||
default_pipeline_id=self.pipeline_id,
|
||
delegate_depth=self._delegate_depth + 1,
|
||
)
|
||
prompt = goal if not context else f"{goal}\n\n背景：{context}"
|
||
result_parts = []
|
||
async for chunk in sub.run(prompt):
|
||
try:
|
||
data = json.loads(chunk)
|
||
except Exception:
|
||
continue
|
||
t = data.get("type", "")
|
||
if t == "reply":
|
||
result_parts.append(data.get("message", ""))
|
||
elif t == "tool_result":
|
||
result_parts.append(data.get("result", ""))
|
||
out = "\n".join(x for x in result_parts if x).strip()
|
||
return out[:3000] or "(子任务无输出)"
|
||
except Exception as e:
|
||
return f"ERROR: {str(e)[:300]}"
|
||
|
||
async def _t_subagent(self, sor, p, pid):
|
||
"""后台子agent管理（list/steer/stop/result）。隔离靠 workspace 目录。"""
|
||
from . import subagents
|
||
action = (p.get("action") or "").strip().lower()
|
||
sid = (p.get("subagent_id") or "").strip()
|
||
try:
|
||
if action == "list":
|
||
items = subagents.list_subagents(self.workspace_dir)
|
||
if not items:
|
||
return "(本会话还没有派生过后台子agent)"
|
||
lines = []
|
||
for it in items[:10]:
|
||
lines.append(
|
||
f"- {it['subagent_id']} [{it['status']}] {it['goal']}"
|
||
+ (f"（错误: {it['error']}）" if it.get("error") else ""))
|
||
return "\n".join(lines)
|
||
if not sid:
|
||
return "FAIL: 需要 subagent_id（先用 action=list 查看）"
|
||
if action == "steer":
|
||
return subagents.steer_subagent(
|
||
self.workspace_dir, sid, (p.get("message") or "").strip())
|
||
if action == "stop":
|
||
return subagents.stop_subagent(self.workspace_dir, sid)
|
||
if action == "result":
|
||
r = subagents.get_result(self.workspace_dir, sid)
|
||
m = r["meta"]
|
||
return (f"subagent_id={sid} status={m['status']}\n"
|
||
f"--- 输出 ---\n{r['result']}")
|
||
except subagents.SubagentError as e:
|
||
return f"FAIL: {e}"
|
||
except Exception as e:
|
||
return f"ERROR: {str(e)[:300]}"
|
||
return "FAIL: action 须为 list|steer|stop|result"
|
||
|
||
# ═══════════════════════════════════════════════════════
|
||
# 上下文压缩
|
||
# ═══════════════════════════════════════════════════════
|
||
|
||
async def _maybe_compress(self):
|
||
"""检查是否需要压缩上下文"""
|
||
est_tokens = self._estimate_tokens()
|
||
limit = self.config.context_limit
|
||
threshold = int(limit * self.config.compression.threshold)
|
||
|
||
if est_tokens < threshold:
|
||
return
|
||
|
||
logger.info(f"Compressing: {est_tokens}/{limit} tokens (threshold={threshold})")
|
||
|
||
# 保留 system + 最近 N 轮
|
||
keep = self.config.compression.keep_recent * 2 # user + assistant 各一条
|
||
system_msg = self._msgs[0] if self._msgs else None
|
||
recent = self._msgs[-keep:] if keep < len(self._msgs) else self._msgs[1:]
|
||
|
||
# 压缩中间消息
|
||
middle = self._msgs[1:-keep] if keep < len(self._msgs) else []
|
||
if middle:
|
||
summary = await self._summarize(middle)
|
||
compressed = [
|
||
system_msg,
|
||
{"role": "user", "content": f"[对话摘要]\n{summary}"},
|
||
] if system_msg else []
|
||
compressed.extend(recent)
|
||
self._msgs = compressed
|
||
|
||
def _estimate_tokens(self) -> int:
|
||
"""简单 token 估算（中文 ~1.5 字符/token，英文 ~4 字符/token）"""
|
||
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(self._content_as_text(m.get("content"))[:200] for m in messages)
|
||
|
||
text = "\n".join(
|
||
f"[{m['role']}]: {self._content_as_text(m.get('content'))[:500]}"
|
||
for m in messages
|
||
)
|
||
|
||
try:
|
||
from pipeline_service.llm_bridge import llm_call
|
||
|
||
summary = await llm_call(
|
||
f"请用3-5句话总结以下对话的关键信息：\n\n{text[:4000]}",
|
||
temperature=0.1,
|
||
model=self.model_name,
|
||
org_id=self.org_id,
|
||
purpose='utility',
|
||
session_id=self.session_id,
|
||
timeout=self._UTILITY_TIMEOUT,
|
||
)
|
||
return summary[:500]
|
||
except Exception:
|
||
return text[:500]
|
||
|
||
# ═══════════════════════════════════════════════════════
|
||
# 会话管理
|
||
# ═══════════════════════════════════════════════════════
|
||
|
||
async def _load_history(self, external_history: List[dict] = None) -> List[dict]:
|
||
"""加载历史消息。隔离策略：session_id > project 级别。
|
||
|
||
多 tab 独立会话：session_id 非空时按 session_id 隔离（每个 tab 独立历史）；
|
||
否则退回 config.session_isolation（project/user/none）。
|
||
"""
|
||
if external_history:
|
||
return external_history
|
||
|
||
if self.config.session_isolation == "none":
|
||
return []
|
||
|
||
try:
|
||
from sqlor.dbpools import DBPools
|
||
|
||
db = DBPools()
|
||
async with db.sqlorContext("pipeline") as sor:
|
||
# 会话级隔离（web 多 tab 独立会话）
|
||
# 注意（2026-09-01 修复）：必须 ORDER BY created_at DESC LIMIT N 取「最新」N 条
|
||
# 再倒序回时间正序。历史用 ASC LIMIT 会取「最旧」N 条——会话消息超过
|
||
# history_limit 后，最近的提问（如 A/B/C 选项）被截掉，LLM 看不到刚问的
|
||
# 问题，按更旧的上下文答非所问（用户答 C，agent 复读 4 小时前的旧回复）。
|
||
if self.session_id:
|
||
# 产线隔离（2026-08-31 修复跨产线串扰）：各产线页面默认 tab 共用
|
||
# session_id='default'，历史若不按产线过滤，投标页会加载开发页的
|
||
# 对话（如"元景项目/新建demo项目"），LLM 跟着最后上下文答错产线。
|
||
# 2026-09-05 第7层泄漏修复：通用会话（存库时 pipeline_id=''）只回放
|
||
# 自己的历史，否则会加载该用户各产线的最近对话——项目名称/ID
|
||
# 等产线信息随历史泄漏进通用助手。
|
||
sql = ("SELECT role, content FROM pipeline_conversations "
|
||
"WHERE session_id=${sid}$ AND created_by=${uid}$ ")
|
||
params = {"sid": self.session_id, "uid": self.user_id or "",
|
||
"lim": self.config.history_limit}
|
||
if self.generic:
|
||
sql += "AND (pipeline_id='' OR pipeline_id IS NULL) "
|
||
elif self.pipeline_id:
|
||
sql += "AND pipeline_id=${pl}$ "
|
||
params["pl"] = self.pipeline_id
|
||
sql += "ORDER BY created_at DESC LIMIT ${lim}$"
|
||
recs = await sor.sqlExe(sql, params)
|
||
else:
|
||
# 项目隔离
|
||
pid = self.project_id
|
||
if self.config.session_isolation == "project" and pid:
|
||
recs = await sor.sqlExe(
|
||
"SELECT role, content FROM pipeline_conversations "
|
||
"WHERE iteration_id=${pid}$ AND created_by=${uid}$ "
|
||
"ORDER BY created_at DESC LIMIT ${lim}$",
|
||
{"pid": pid, "uid": self.user_id or "", "lim": self.config.history_limit},
|
||
)
|
||
elif self.generic:
|
||
# 2026-09-05 第7层泄漏修复：通用会话只回放自己的历史
|
||
# （存库时 generic 会话 pipeline_id=''），禁止按 user_id 全量
|
||
# 回放——否则用户各产线最近对话（含项目名称/ID）全部泄漏。
|
||
recs = await sor.sqlExe(
|
||
"SELECT role, content FROM pipeline_conversations "
|
||
"WHERE created_by=${uid}$ AND (pipeline_id='' OR pipeline_id IS NULL) "
|
||
"ORDER BY created_at DESC LIMIT ${lim}$",
|
||
{"uid": self.user_id or "", "lim": self.config.history_limit},
|
||
)
|
||
else:
|
||
recs = await sor.sqlExe(
|
||
"SELECT role, content FROM pipeline_conversations "
|
||
"WHERE created_by=${uid}$ "
|
||
"ORDER BY created_at DESC LIMIT ${lim}$",
|
||
{"uid": self.user_id or "", "lim": self.config.history_limit},
|
||
)
|
||
|
||
if not recs:
|
||
return []
|
||
recs = list(reversed(recs)) # DESC 取出 → 倒回时间正序
|
||
|
||
# 过滤 tool_call JSON 污染（关键！）
|
||
msgs = []
|
||
for r in recs:
|
||
role = getattr(r, "role", "")
|
||
content = getattr(r, "content", "") or ""
|
||
# 跳过 tool_call JSON
|
||
if content.startswith('{"action":"tool_call"'):
|
||
continue
|
||
if content.startswith("已调用 "):
|
||
continue
|
||
if role in ("user", "assistant"):
|
||
msgs.append({"role": role, "content": content})
|
||
return msgs
|
||
except Exception as e:
|
||
logger.warning(f"History load failed: {e}")
|
||
return []
|
||
|
||
async def _save_turn(self, user_input: str, reply: str):
|
||
"""持久化本轮对话"""
|
||
if self.config.session_isolation == "none":
|
||
return
|
||
|
||
# PII 落库掩码（2026-09-17）：会话表是长期留存+session_search 可捞，占位符
|
||
# 虽非明文但缓存 24h 内可还原 → 落库一律换掩码形态（138****5678），
|
||
# 缓存过期后掩码仍是唯一可见形态（合规：留存数据不含可还原 PII）。
|
||
if self.pii_session_key and "@@pii:" in (user_input + reply):
|
||
try:
|
||
from . import pii_guard
|
||
if "@@pii:" in user_input:
|
||
user_input = await pii_guard.restore_outbound(
|
||
user_input, self.pii_session_key, masked=True)
|
||
if "@@pii:" in reply:
|
||
reply = await pii_guard.restore_outbound(
|
||
reply, self.pii_session_key, masked=True)
|
||
except Exception as e:
|
||
logger.warning(f"pii mask on save failed: {e}")
|
||
|
||
try:
|
||
from sqlor.dbpools import DBPools
|
||
from appPublic.uniqueID import getID
|
||
|
||
db = DBPools()
|
||
async with db.sqlorContext("pipeline") as sor:
|
||
# 产线隔离（2026-08-31）：generic 会话不挂产线标签——generic 的
|
||
# pipeline_id 会回退 DEFAULT_ABILITY_ID(sdlc_general)，若落库会污染
|
||
# 开发产线历史。存空串 = 通用对话，各产线读取时按产线过滤自然排除。
|
||
store_pl = "" if self.generic else (self.pipeline_id or "")
|
||
# 用户消息
|
||
await sor.C("pipeline_conversations", {
|
||
"id": getID(),
|
||
"role": "user",
|
||
"content": user_input,
|
||
"created_by": self.user_id or "",
|
||
"iteration_id": self.project_id or "",
|
||
"session_id": self.session_id or "",
|
||
"pipeline_id": store_pl,
|
||
})
|
||
# 助手回复（不含 tool_call JSON！）
|
||
reply_clean = reply
|
||
if reply.startswith('{"action":"tool_call"'):
|
||
reply_clean = "[工具调用]"
|
||
await sor.C("pipeline_conversations", {
|
||
"id": getID(),
|
||
"role": "assistant",
|
||
"content": reply_clean[:4000],
|
||
"created_by": self.user_id or "",
|
||
"iteration_id": self.project_id or "",
|
||
"session_id": self.session_id or "",
|
||
"pipeline_id": store_pl,
|
||
})
|
||
except Exception as e:
|
||
logger.warning(f"Session save failed: {e}")
|
||
|
||
async def _load_project_context(self) -> str:
|
||
"""加载项目上下文（项目名、任务数等）"""
|
||
if not self.project_id:
|
||
return ""
|
||
|
||
try:
|
||
from sqlor.dbpools import DBPools
|
||
|
||
db = DBPools()
|
||
async with db.sqlorContext("pipeline") as sor:
|
||
recs = await sor.sqlExe(
|
||
"SELECT name, description FROM sd_projects WHERE id=${pid}$",
|
||
{"pid": self.project_id})
|
||
if recs:
|
||
name = getattr(recs[0], "name", "")
|
||
return f"项目: {name}"
|
||
except Exception:
|
||
pass
|
||
return ""
|
||
|
||
def _is_dangerous_command(self, cmd: str) -> bool:
|
||
"""判断 shell 命令是否危险（run_command 只对危险命令要求确认）。"""
|
||
c = (cmd or "").strip().lower()
|
||
dangerous = (
|
||
"rm -rf", "rm -r", "sudo", "su ", "drop table", "drop database",
|
||
"truncate", "delete from", "mkfs", "dd if", ":(){", "shutdown",
|
||
"reboot", "kill -9", "chmod 777", "chown", "> /dev/",
|
||
)
|
||
return any(d in c for d in dangerous)
|
||
|
||
def _detect_confirm_decision(self, user_input: str) -> str:
|
||
"""检测用户对危险命令确认的回复。返回 approve / approve_all / cancel / ''。"""
|
||
t = (user_input or "").strip().lower()
|
||
if not t:
|
||
return ""
|
||
for kw in ("全部确认", "确认全部", "全部同意", "都确认", "都同意", "一律确认", "approve all", "yes to all", "always"):
|
||
if kw in t:
|
||
return "approve_all"
|
||
for kw in ("取消", "放弃", "不执行", "别执行", "cancel", "abort", "不要"):
|
||
if kw in t:
|
||
return "cancel"
|
||
for kw in ("确认", "同意", "执行", "可以", "approve", "confirm", "yes", "ok", "继续"):
|
||
if kw in t:
|
||
return "approve"
|
||
return ""
|
||
|
||
def _needs_confirmation(self, tool_name: str, params: dict = None) -> bool:
|
||
"""检查工具是否需要用户确认。
|
||
|
||
会话级「全部确认」（approve_all）时直接放行；否则 run_command 只对
|
||
危险命令确认，普通命令自动执行，其他 requires_confirmation 工具一律确认。
|
||
|
||
别名先归一再查注册表（2026-08-31 修复）：否则 LLM 用别名（如 stop_project）
|
||
调需确认工具时，注册表查不到 → 误判无需确认 → 绕过确认门直接执行。
|
||
"""
|
||
tool_name = _PROJECT_TOOL_ALIASES.get(tool_name, tool_name)
|
||
if self._session and getattr(self._session, "approve_all", False):
|
||
return False
|
||
if self._tool_registry:
|
||
tool = self._tool_registry.get(tool_name)
|
||
if tool and tool.requires_confirmation:
|
||
if tool_name == "run_command":
|
||
cmd = (params or {}).get("command", "")
|
||
return self._is_dangerous_command(cmd)
|
||
return True
|
||
return False
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# 便捷函数
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
async def run_agent(
|
||
user_input: str,
|
||
project_id: str = "",
|
||
user_id: str = "",
|
||
config=None,
|
||
model_name: str = None,
|
||
workspace_dir: str = "",
|
||
) -> AsyncGenerator[str, None]:
|
||
"""便捷入口：运行 agent 并流式输出。
|
||
|
||
用法：
|
||
async for chunk in run_agent("创建一个新项目"):
|
||
print(chunk)
|
||
"""
|
||
if config is None:
|
||
try:
|
||
from pipeline_core.agent_config import load_agent_config, SDLC_DEFAULT_CONFIG
|
||
|
||
config = await load_agent_config(project_id=project_id)
|
||
except ImportError:
|
||
config = SDLC_DEFAULT_CONFIG
|
||
|
||
executor = AgentExecutor(
|
||
config=config,
|
||
project_id=project_id,
|
||
user_id=user_id,
|
||
workspace_dir=workspace_dir,
|
||
model_name=model_name,
|
||
)
|
||
async for chunk in executor.run(user_input):
|
||
yield chunk
|