""" 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 time from typing import AsyncGenerator, Dict, List, Optional logger = logging.getLogger("pipeline.agent_executor") # ── 默认工具定义（在 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. 会话持久化 """ def __init__( self, config, # AgentConfig (from pipeline_core) project_id: str = "", user_id: str = "", workspace_dir: str = "", model_name: str = None, ): self.config = config self.project_id = project_id self.user_id = user_id self.org_id = "" # loaded from project context self.workspace_dir = workspace_dir or "/tmp/pipeline_ws" self.model_name = model_name or config.model_name # 运行状态 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._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) -> AsyncGenerator[str, None]: """执行 agent 主循环。 Args: user_input: 用户输入 history: 历史消息列表 [{"role":"user"|"assistant","content":"..."}] Yields: NDJSON 行：{"type":"progress"|"tool_call"|"reply"|"error", ...} """ self._started_at = time.time() self._turn_count = 0 self._tool_call_count = 0 # Step 1: 初始化 await self._init_components() # 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) self._msgs.append({"role": "user", "content": user_input}) # 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 # 5a. 上下文压缩 if self.config.compression.enabled: await self._maybe_compress() # 5b. LLM 调用（native function calling，返回 dict） resp = await self._call_llm() # 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): 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 yield json.dumps({ "type": "tool_result", "tool": tool_name, "result": result[:500], }, ensure_ascii=False) + "\n" # 原生回填：role=tool + tool_call_id self._msgs.append({ "role": "tool", "tool_call_id": tc.get("id", "") if isinstance(tc, dict) else "", "content": str(result), }) 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 → 强制注入工具调用 if self._tool_call_count == 0: # 意图识别：新需求描述（长文本+需求特征词）→ create_task 推进 req_keywords = ["实现", "开发", "建设", "系统", "功能", "需求", "模块", "支持", "切换", "监控", "备份", "同步", "要做", "设计", "平台"] is_requirement = (len(user_input) > 40 and any(kw in user_input for kw in req_keywords)) if is_requirement: hint_tool = "create_task" elif any(kw in user_input.lower() for kw in ["任务", "task", "list"]): hint_tool = "list_tasks" elif any(kw in user_input.lower() for kw in ["问题", "question"]): hint_tool = "list_questions" else: hint_tool = "diagnose_project" yield json.dumps({ "type": "auto_tool", "message": f"未调工具，自动注入 {hint_tool}", }, ensure_ascii=False) + "\n" self._msgs.append({"role": "assistant", "content": f"已调用 {hint_tool}"}) if hint_tool == "create_task": self._msgs.append({"role": "user", "content": "用户描述了新需求，请用 create_task 创建任务推进（可先 role=requirement 做需求分析，再拆分到 develop），填好 title/role/description 参数，输出 tool_call JSON。"}) else: self._msgs.append({"role": "user", "content": f"请调用 {hint_tool} 工具，只输出 {{\"action\":\"tool_call\",\"tool\":\"{hint_tool}\",\"params\":{{}}}}"}) continue # 调了工具但太少（1-2次），LLM 想停 → 推它继续深入（最多3次） if self._tool_call_count <= 2 and self._auto_push_count < 3: self._auto_push_count += 1 hint = "请继续：查看失败任务的详情(task_detail)，查看待回答问题(list_questions)，或启动agent(start_agents)" yield json.dumps({ "type": "auto_tool", "message": f"已调{self._tool_call_count}次工具，继续深入({self._auto_push_count}/3)", }, ensure_ascii=False) + "\n" self._msgs.append({"role": "assistant", "content": "继续分析"}) self._msgs.append({"role": "user", "content": hint}) continue final_reply = act.get("message", "") yield json.dumps({ "type": "reply", "message": final_reply, }, ensure_ascii=False) + "\n" await self._save_turn(user_input, final_reply) return elif action_type == "tool_call": tool_name = act.get("tool", "") tool_params = act.get("params", {}) # 需要确认的工具 if self._needs_confirmation(tool_name): 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 yield json.dumps({ "type": "tool_result", "tool": tool_name, "result": result[:500], }, 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): """懒加载各组件""" # Tool Registry try: from pipeline_core.tool_registry import get_tool_registry self._tool_registry = get_tool_registry() # 把 config.tools 注册进 registry（registry 是全局单例，可能为空） if self._tool_registry and self.config.tools: existing = set(self._tool_registry.get_tool_names()) for t in self.config.tools: if t.name not in existing: self._tool_registry.register(t) except ImportError: self._tool_registry = None # Skill Loader — 三级隔离 try: from pipeline_core.skill_loader import get_skill_loader base_dir = self.config.skills.base_dir self._skill_loader = get_skill_loader(base_dir) except ImportError: self._skill_loader = None # Memory Store try: from pipeline_core.memory_store import get_memory_store self._memory_store = get_memory_store() except ImportError: self._memory_store = None # 加载 org_id（从项目上下文） 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 FROM sd_projects WHERE id=${pid}$", {"pid": self.project_id}) if recs: self.org_id = getattr(recs[0], 'org_id', '') or '' except Exception: pass 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 "" # 注入记忆 if self.config.memory.enabled and self._memory_store: mem_block = await self._memory_store.build_prompt_block(max_entries=15) if mem_block: prompt += f"\n\n## 持久记忆\n{mem_block}" # 注入技能（三级隔离：user > org > global） if self.config.skills.enabled and self._skill_loader: skill_block = self._skill_loader.build_prompt_block( user_input=user_input, org_id=self.org_id if self.config.skills.enable_org else "", user_id=self.user_id if self.config.skills.enable_user else "", 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) 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, ) except Exception as e: logger.error(f"native function calling failed, fallback to text: {e}") # 回退：纯文本调用（content 为原始文本） from pipeline_service.llm_bridge import llm_call_msgs content = await llm_call_msgs( self._msgs, model=self.model_name, temperature=self.config.temperature, ) return {"content": content or "", "tool_calls": []} # ═══════════════════════════════════════════════════════ # 解析 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（始终可用） """ # 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]}" # 2. SDLC 内建工具 result = await self._execute_sdlc_tool(tool_name, params) if result is not None: return result # 3. ask_user if tool_name == "ask_user": return f"QUESTION: {params.get('question', '')}" return f"未知工具: {tool_name}" async def _execute_sdlc_tool(self, tool_name: str, params: dict) -> Optional[str]: """SDLC 内建工具（兼容 cockpit_chat.dspy）""" # 导入现有 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 # 所有工具映射（含别名） handlers = { "switch_project": self._t_switch_project, "create_project": self._t_create_project, "create_task": self._t_create_task, "list_tasks": self._t_list_tasks, "get_task_list": self._t_list_tasks, "get_tasks": self._t_list_tasks, # 别名 "task_detail": self._t_task_detail, "get_task": self._t_task_detail, "get_task_detail": self._t_task_detail, # 别名 "start_agents": self._t_start_agents, "diagnose_project": self._t_diagnose_project, "list_deliverables": self._t_list_deliverables, "view_deliverable": self._t_view_deliverable, "get_deliverable": self._t_view_deliverable, # 别名 "list_questions": self._t_list_questions, "answer_question": self._t_answer_question, "add_repo": self._t_add_repo, "list_repos": self._t_list_repos, "clone_repo": self._t_clone_repo, "run_command": self._t_run_command, "check_progress": self._t_check_progress, "add_bug": self._t_add_bug, } handler = handlers.get(tool_name) if handler: return await handler(sor, p, pid) return None # ── 工具实现 ── async def _persist_project(self, sor, pid): """持久化当前项目到 pipeline_agent_settings（user_id 唯一键）。 只改 self.project_id 不够——AgentExecutor 每轮新建，run 结束即销毁， 下一轮 cockpit_chat_v2 又从 pipeline_agent_settings 读回旧项目。 """ 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}) 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}$", {"n": name}) if recs: r = recs[0] self.project_id = getattr(r, "id", "") await self._persist_project(sor, self.project_id) 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 ORDER BY created_at DESC LIMIT 20", {}) 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) matched = matched.strip().strip('"').strip("'") recs2 = await sor.sqlExe( "SELECT id, name FROM sd_projects WHERE name=${n}$", {"n": matched}) if recs2: r = recs2[0] self.project_id = getattr(r, "id", "") await self._persist_project(sor, self.project_id) return f"OK: 已切换到 {getattr(r, 'name', matched)}" except Exception: pass return f"未找到项目: {name}" async def _t_create_project(self, sor, p, pid): import os 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" # 项目专属工作空间目录（workspace_base 从 appbase params 表读，可动态配置） # 不设置 workspace_dir 会导致 _resolve_workspace 回退到 ~/pipeline_ws/default， # 所有项目共用同一目录，设计文档/代码互相覆盖。 from .workspace import get_workspace_base workspace_base = await get_workspace_base(sor) workspace_dir = os.path.join(workspace_base, str(org_id), name) os.makedirs(workspace_dir, exist_ok=True) pid_val = getID() await sor.C("sd_projects", { "id": pid_val, "name": name, "description": desc, "status": "active", "org_id": org_id, "workspace_dir": workspace_dir, }) await sor.C("sd_iterations", { "id": getID(), "project_id": pid_val, "iteration_name": f"{name}-初始迭代", "iteration_type": "default", "status": "active", "priority": 1, }) self.project_id = pid_val await self._persist_project(sor, pid_val) return f"OK: 已创建项目 {name}" async def _t_create_task(self, sor, p, pid): if not pid: return "请先切换到项目" from appPublic.uniqueID import getID title = p.get("title", "").strip() role = p.get("role", "develop") desc = p.get("description", "") if not title: return "需要任务标题" tid = getID() await sor.C("pipeline_tasks", { "id": tid, "tenant_id": pid, "pipeline_id": "role_task", "owner_id": self.user_id or "user", "title": title, "state": "submitted", "role": role, "params": json.dumps({"description": desc}, ensure_ascii=False), }) return f"OK: 已创建任务 {title}（角色: {role}）" async def _t_list_tasks(self, sor, p, pid): if not pid: return "请先切换到项目" role = p.get("role", "") state = p.get("state", "") sql = "SELECT id, title, role, state, created_at FROM pipeline_tasks WHERE tenant_id=${pid}$" params_dict = {"pid": pid} if role: sql += " AND role=${role}$" params_dict["role"] = role if state: sql += " AND state=${state}$" params_dict["state"] = state sql += " ORDER BY created_at DESC LIMIT 30" recs = await sor.sqlExe(sql, params_dict) if not recs: return "暂无任务" # v2.1 format: markdown table icons = { "submitted": "⏳", "running": "🔄", "review": "👀", "approved": "✅", "completed": "✔️", "failed": "❌", "waiting": "⏸️", } lines = ["| 状态 | 任务 | 角色 | ID |", "|------|------|------|----|"] for r in recs: st = getattr(r, 'state', '?') icon = icons.get(st, "❓") title = (getattr(r, 'title', '') or '')[:60] role = getattr(r, 'role', '') tid = getattr(r, 'id', '') lines.append(f"| {icon} {st} | {title} | {role} | {tid} |") return "\n".join(lines) async def _t_task_detail(self, sor, p, pid): tid = p.get("task_id", "") if not tid: return "需要任务ID" recs = await sor.sqlExe( "SELECT * FROM pipeline_tasks WHERE id=${tid}$", {"tid": tid}) if not recs and len(tid) >= 6: # 前缀匹配兜底（兼容 8 位截断 ID） recs = await sor.sqlExe( "SELECT * FROM pipeline_tasks WHERE id LIKE ${prefix}$ LIMIT 1", {"prefix": tid + "%"}) if not recs: return f"任务不存在: {tid}" r = recs[0] return json.dumps({ "id": getattr(r, "id", ""), "title": getattr(r, "title", ""), "state": getattr(r, "state", ""), "role": getattr(r, "role", ""), "params": getattr(r, "params", "{}"), }, ensure_ascii=False, indent=2) async def _t_start_agents(self, sor, p, pid): if not pid: return "请先切换到项目" # 找 submitted 任务 recs = await sor.sqlExe( "SELECT id, title, role FROM pipeline_tasks " "WHERE tenant_id=${pid}$ AND state='submitted' ORDER BY created_at ASC LIMIT 10", {"pid": pid}) if not recs: return "没有待执行的任务" results = [] for r in recs: tid = getattr(r, "id", "") role = getattr(r, "role", "develop") try: from pipeline_service.agent_loop import role_agent_run result = await role_agent_run(pid, role) results.append(f"- {getattr(r, 'title', '')}: {result.get('status', '?')}") except Exception as e: results.append(f"- {getattr(r, 'title', '')}: error={str(e)[:100]}") return "Agent 执行结果:\n" + "\n".join(results) async def _t_diagnose_project(self, sor, p, pid): if not pid: return "请先切换到项目" # 汇总各种状态 submitted = await sor.sqlExe( "SELECT COUNT(*) as c FROM pipeline_tasks WHERE tenant_id=${pid}$ AND state='submitted'", {"pid": pid}) running = await sor.sqlExe( "SELECT COUNT(*) as c FROM pipeline_tasks WHERE tenant_id=${pid}$ AND state='running'", {"pid": pid}) failed = await sor.sqlExe( "SELECT COUNT(*) as c FROM pipeline_tasks WHERE tenant_id=${pid}$ AND state='failed'", {"pid": pid}) review = await sor.sqlExe( "SELECT COUNT(*) as c FROM pipeline_tasks WHERE tenant_id=${pid}$ AND state='review'", {"pid": pid}) # questions 表可能没有 answered 列，用 status='pending' 兜底 qc = 0 try: questions = await sor.sqlExe( "SELECT COUNT(*) as c FROM pipeline_agent_questions WHERE tenant_id=${pid}$ AND status='pending'", {"pid": pid}) qc = getattr(questions[0], "c", 0) if questions else 0 except Exception: pass sc = getattr(submitted[0], "c", 0) if submitted else 0 rc = getattr(running[0], "c", 0) if running else 0 fc = getattr(failed[0], "c", 0) if failed else 0 vc = getattr(review[0], "c", 0) if review else 0 lines = [ "项目诊断:", f"- 待执行: {sc}", f"- 运行中: {rc}", f"- 待审核: {vc}", f"- 失败: {fc}", f"- 待回答问题: {qc}", ] # 列出失败/待执行任务完整 ID，方便 agent 直接 task_detail if fc: failed_rows = await sor.sqlExe( "SELECT id, title, role FROM pipeline_tasks WHERE tenant_id=${pid}$ AND state='failed' ORDER BY created_at DESC LIMIT 10", {"pid": pid}) lines.append("失败任务:") for fr in (failed_rows or []): lines.append(f" - [{getattr(fr, 'role', '?')}] {getattr(fr, 'title', '')} (id={getattr(fr, 'id', '')})") if sc: submitted_rows = await sor.sqlExe( "SELECT id, title, role FROM pipeline_tasks WHERE tenant_id=${pid}$ AND state='submitted' ORDER BY created_at ASC LIMIT 10", {"pid": pid}) lines.append("待执行任务:") for sr in (submitted_rows or []): lines.append(f" - [{getattr(sr, 'role', '?')}] {getattr(sr, 'title', '')} (id={getattr(sr, 'id', '')})") # 运行中任务明细：claimed_by + 心跳陈旧度，帮助 agent 直接识别僵尸任务 # （进程崩溃/协程挂起后心跳不再更新，超时即疑似僵尸，会被 poller 自动回收重跑） if rc: running_rows = await sor.sqlExe( "SELECT id, title, role, claimed_by, " "TIMESTAMPDIFF(MINUTE, updated_at, NOW()) AS mins " "FROM pipeline_tasks WHERE tenant_id=${pid}$ AND state='running' " "ORDER BY updated_at ASC LIMIT 10", {"pid": pid}) lines.append("运行中任务:") for rr in (running_rows or []): rrole = getattr(rr, 'role', '?') rtitle = getattr(rr, 'title', '') rid = getattr(rr, 'id', '') rcb = getattr(rr, 'claimed_by', '') or '' try: mins = int(getattr(rr, 'mins', 0) or 0) except (TypeError, ValueError): mins = 0 if mins >= 10: flag = f"⚠️心跳超时{mins}分钟(疑似僵尸,将被回收重跑)" else: flag = f"已运行{mins}分钟" lines.append(f" - [{rrole}] {rtitle} (id={rid}) | {flag} | claimed_by={rcb[:8]}") return "\n".join(lines) async def _t_list_deliverables(self, sor, p, pid): if not pid: return "请先切换到项目" role = p.get("role", "") sql = "SELECT id, title, deliverable_type, review_status FROM pipeline_deliverables WHERE project_id=${pid}$" params_dict = {"pid": pid} sql += " ORDER BY created_at DESC LIMIT 20" recs = await sor.sqlExe(sql, params_dict) if not recs: return "暂无交付件" lines = [] for r in recs: lines.append( f"- [{getattr(r, 'review_status', '?')}] {getattr(r, 'title', '')} " f"({getattr(r, 'deliverable_type', '')}) id={getattr(r, 'id', '')}" ) return "\n".join(lines) async def _t_view_deliverable(self, sor, p, pid): did = p.get("deliverable_id", "") if not did: return "需要交付件ID" recs = await sor.sqlExe( "SELECT * FROM pipeline_deliverables WHERE id=${did}$", {"did": did}) if not recs and len(did) >= 6: recs = await sor.sqlExe( "SELECT * FROM pipeline_deliverables WHERE id LIKE ${prefix}$ LIMIT 1", {"prefix": did + "%"}) if not recs: return f"交付件不存在: {did}" r = recs[0] return getattr(r, "content", "")[:4000] or "(空)" async def _t_list_questions(self, sor, p, pid): if not pid: return "请先切换到项目" try: recs = await sor.sqlExe( "SELECT id, question, answer_source FROM pipeline_agent_questions " "WHERE tenant_id=${pid}$ AND status='pending' ORDER BY created_at DESC LIMIT 10", {"pid": pid}) except Exception: # 兼容无 status 列的情况 try: recs = await sor.sqlExe( "SELECT id, question, answer_source FROM pipeline_agent_questions " "WHERE tenant_id=${pid}$ ORDER BY created_at DESC LIMIT 10", {"pid": pid}) except Exception: return "暂无问题数据" if not recs: return "没有待回答问题" lines = [] for r in recs: lines.append(f"- [{getattr(r, 'id', '')}] {getattr(r, 'question', '')[:200]}") return "\n".join(lines) async def _t_answer_question(self, sor, p, pid): qid = p.get("question_id", "") answer = p.get("answer", "") if not qid or not answer: return "需要问题ID和回答内容" # 先精确匹配，再前缀匹配兜底（兼容截断 ID） recs = await sor.sqlExe( "SELECT id, task_id FROM pipeline_agent_questions WHERE id=${qid}$", {"qid": qid}) if not recs and len(qid) >= 6: recs = await sor.sqlExe( "SELECT id, task_id FROM pipeline_agent_questions WHERE id LIKE ${prefix}$ LIMIT 1", {"prefix": qid + "%"}) if not recs: return f"问题不存在: {qid}" full_qid = getattr(recs[0], "id", qid) task_id = getattr(recs[0], "task_id", "") or "" await sor.sqlExe( "UPDATE pipeline_agent_questions SET answer=${a}$, answer_source='main_agent', " "answered_by='main_agent', status='answered' WHERE id=${qid}$", {"a": answer, "qid": full_qid}) # 恢复任务：waiting → submitted，并清 claimed_by。 # 否则任务永久卡在 waiting（poll 器要求 state='submitted' AND claimed_by IS NULL 才会重新认领）。 resumed = False if task_id: await sor.sqlExe( "UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL " "WHERE id=${tid}$ AND state='waiting'", {"tid": task_id}) resumed = True suffix = "，任务已恢复执行" if resumed else "" return "OK: 已回答" + suffix async def _t_add_repo(self, sor, p, pid): if not pid: return "请先切换到项目" from appPublic.uniqueID import getID url = p.get("repo_url", "").strip() name = p.get("repo_name", "").strip() branch = p.get("default_branch", "main") if not url: return "需要仓库URL" await sor.C("sd_project_repos", { "id": getID(), "project_id": pid, "repo_url": url, "repo_name": name or url.split("/")[-1].replace(".git", ""), "default_branch": branch, }) return f"OK: 已添加仓库 {name or url}" async def _t_list_repos(self, sor, p, pid): if not pid: return "请先切换到项目" recs = await sor.sqlExe( "SELECT repo_name, repo_url, default_branch FROM sd_project_repos WHERE project_id=${pid}$", {"pid": pid}) if not recs: return "暂无关联仓库" lines = [] for r in recs: lines.append(f"- {getattr(r, 'repo_name', '')}: {getattr(r, 'repo_url', '')}") return "\n".join(lines) async def _t_clone_repo(self, sor, p, pid): if not pid: return "请先切换到项目" import os from pipeline_service.agent_loop import ( _setup_repos, _get_workspace_dir, _git_clone) workspace_dir = await _get_workspace_dir(sor, pid) url = (p.get("repo_url", "") or "").strip() if url: name = url.rstrip("/").split("/")[-1].replace(".git", "") target = os.path.join(workspace_dir, "repos", name) r = await _git_clone(url, target, p.get("branch", "main")) return f"rc={r['rc']} {r['message']}" results = await _setup_repos(sor, workspace_dir, pid) if not results: return "无关联仓库可克隆" return "\n".join( f"- {x['repo']}: rc={x.get('rc', '?')} {x.get('message', '')}" for x in results) 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 r = await _run_shell(cmd, self.workspace_dir, timeout=60) return f"rc={r['rc']}\n{r['stdout'][:2000]}" except Exception as e: return f"ERROR: {str(e)[:300]}" async def _t_check_progress(self, sor, p, pid): if not pid: return "请先切换到项目" recs = await sor.sqlExe( "SELECT role, state, COUNT(*) as c FROM pipeline_tasks " "WHERE tenant_id=${pid}$ GROUP BY role, state ORDER BY role, state", {"pid": pid}) if not recs: return "暂无进度数据" lines = [] for r in recs: lines.append( f"- {getattr(r, 'role', '?')}: {getattr(r, 'state', '?')} x{getattr(r, 'c', 0)}" ) return "\n".join(lines) async def _t_add_bug(self, sor, p, pid): if not pid: return "请先切换到项目" from appPublic.uniqueID import getID title = p.get("title", "").strip() desc = p.get("description", "") severity = p.get("severity", "medium") if not title: return "需要Bug标题" # 找项目的迭代（iteration_id 必填） iteration_id = p.get("iteration_id", "") if not iteration_id: its = await sor.sqlExe( "SELECT id FROM sd_iterations WHERE project_id=${pid}$ ORDER BY created_at ASC LIMIT 1", {"pid": pid}) if its: iteration_id = getattr(its[0], "id", "") if not iteration_id: return "ERROR: 项目无迭代，无法提交Bug" await sor.C("sd_bugs", { "id": getID(), "title": title, "description": desc, "severity": severity, "status": "open", "iteration_id": iteration_id, }) return f"OK: 已提交Bug: {title}" # ═══════════════════════════════════════════════════════ # 上下文压缩 # ═══════════════════════════════════════════════════════ 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 "" # 简单估算：平均每 2 字符 1 token total += len(content) // 2 + 1 return total async def _summarize(self, messages: list) -> str: """压缩消息为摘要""" if len(messages) <= 2: return "\n".join((m.get("content") or "")[:200] for m in messages) text = "\n".join( f"[{m['role']}]: {(m.get('content') or '')[: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, ) return summary[:500] except Exception: return text[:500] # ═══════════════════════════════════════════════════════ # 会话管理 # ═══════════════════════════════════════════════════════ async def _load_history(self, external_history: List[dict] = None) -> List[dict]: """加载历史消息。隔离策略：project 级别。""" 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: # 项目隔离 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 ASC LIMIT ${lim}$", {"pid": pid, "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 ASC LIMIT ${lim}$", {"uid": self.user_id or "", "lim": self.config.history_limit}, ) if not recs: return [] # 过滤 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 try: from sqlor.dbpools import DBPools from appPublic.uniqueID import getID db = DBPools() async with db.sqlorContext("pipeline") as sor: # 用户消息 await sor.C("pipeline_conversations", { "id": getID(), "role": "user", "content": user_input, "created_by": self.user_id or "", "iteration_id": self.project_id or "", }) # 助手回复（不含 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 "", }) 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 _needs_confirmation(self, tool_name: str) -> bool: """检查工具是否需要用户确认""" if self._tool_registry: tool = self._tool_registry.get(tool_name) if tool and tool.requires_confirmation: 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