diff --git a/pipeline_service/__init__.py b/pipeline_service/__init__.py index 698b2ed..2b74018 100644 --- a/pipeline_service/__init__.py +++ b/pipeline_service/__init__.py @@ -1,4 +1,4 @@ -"""pipeline_service - 通用产线执行引擎模块""" +"""pipeline_service - 通用产线执行引擎模块。v3.5.0""" from .init import ( load_pipeline_service, @@ -25,4 +25,7 @@ from .state import ( HUMAN_PENDING, HUMAN_SUBMITTED, HUMAN_APPROVED, HUMAN_REJECTED, HUMAN_EXPIRED, ) -__version__ = "3.1.0" +# v2: Agent 执行引擎（对照 Hermes Agent） +from .agent_loop_v2 import AgentExecutor, run_agent + +__version__ = "3.5.0" diff --git a/pipeline_service/agent_loop_v2.py b/pipeline_service/agent_loop_v2.py new file mode 100644 index 0000000..002bf1a --- /dev/null +++ b/pipeline_service/agent_loop_v2.py @@ -0,0 +1,1001 @@ +""" +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.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._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 调用 + raw = await self._call_llm() + + # 5c. 解析 LLM 输出 + actions = self._parse_actions(raw) + + # 5d. 分发处理 + for act in actions: + action_type = act.get("action", "") + + if action_type == "reply": + 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) + 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() + except ImportError: + self._tool_registry = None + + # Skill Loader + try: + from pipeline_core.skill_loader import get_skill_loader + + self._skill_loader = get_skill_loader() + 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 + + async def _build_system_prompt(self, user_input: str) -> str: + """组装完整 system prompt = 基础 prompt + 记忆 + 技能 + 工具列表""" + prompt = self.config.system_prompt + + # 注入当前项目 + prompt = prompt.replace("{current_project}", self.project_id or "未选择") + + # 注入记忆 + 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}" + + # 注入技能 + if self.config.skills.enabled and self._skill_loader: + skill_block = self._skill_loader.build_prompt_block( + user_input=user_input, + max_skills=self.config.skills.max_skills_per_turn, + ) + if skill_block: + prompt += f"\n\n{skill_block}" + + # 注入工具列表 + prompt += "\n\n## 可用工具\n" + if self._tool_registry: + prompt += self._tool_registry.to_text_description() + else: + for t in self.config.tools: + if t.enabled: + params = ", ".join(f"{k}: {v}" for k, v in (t.parameters or {}).items()) + prompt += f"- {t.name}({params}): {t.description}\n" + + # 注入项目上下文 + try: + ctx = await self._load_project_context() + prompt = prompt.replace("{project_context}", ctx) + except Exception: + prompt = prompt.replace("{project_context}", "") + + return prompt + + # ═══════════════════════════════════════════════════════ + # LLM 调用 + # ═══════════════════════════════════════════════════════ + + async def _call_llm(self) -> str: + """调用 LLM，返回原始文本""" + from pipeline_service.llm_bridge import llm_call_msgs + + # 使用 OpenAI native function calling（如果模型支持） + use_native = False + tools_schema = None + if self._tool_registry and use_native: + tools_schema = self._tool_registry.to_openai_schema() + + # 简单消息调用（兼容所有模型） + return await llm_call_msgs( + self._msgs, + model=self.model_name, + temperature=self.config.temperature, + ) + + # ═══════════════════════════════════════════════════════ + # 解析 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, + "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, + "list_questions": self._t_list_questions, + "answer_question": self._t_answer_question, + "add_repo": self._t_add_repo, + "list_repos": self._t_list_repos, + "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 _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", "") + 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", "") + return f"OK: 已切换到 {getattr(r, 'name', matched)}" + except Exception: + pass + + return f"未找到项目: {name}" + + async def _t_create_project(self, sor, p, pid): + from appPublic.uniqueID import getID + + name = p.get("name", "").strip() + desc = p.get("description", "") + if not name: + return "需要项目名称" + + pid_val = getID() + await sor.C("sd_projects", { + "id": pid_val, "name": name, "description": desc, + "status": "active", + }) + 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 + 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 "暂无任务" + + lines = [] + for r in recs: + lines.append( + f"- [{getattr(r, 'state', '?')}] {getattr(r, 'title', '')} " + f"(角色: {getattr(r, 'role', '')}) " + f"id={getattr(r, 'id', '')[:8]}" + ) + 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: + 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 = await sor.sqlExe( + "SELECT COUNT(*) as c FROM pipeline_agent_questions WHERE tenant_id=${pid}$ AND answered IS NULL", + {"pid": pid}) + + 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 + qc = getattr(questions[0], "c", 0) if questions else 0 + + return ( + f"项目诊断:\n" + f"- 待执行: {sc}\n" + f"- 运行中: {rc}\n" + f"- 待审核: {vc}\n" + f"- 失败: {fc}\n" + f"- 待回答问题: {qc}\n" + ) + + 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', '')})" + ) + 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: + 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 "请先切换到项目" + + recs = await sor.sqlExe( + "SELECT id, question, answer_source FROM pipeline_agent_questions " + "WHERE tenant_id=${pid}$ AND answered IS NULL ORDER BY created_at DESC LIMIT 10", + {"pid": pid}) + if not recs: + return "没有待回答问题" + + lines = [] + for r in recs: + lines.append(f"- Q{getattr(r, 'id', '')[:8]}: {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和回答内容" + + await sor.sqlExe( + "UPDATE pipeline_agent_questions SET answered=1, answer=${a}$ WHERE id=${qid}$", + {"a": answer, "qid": qid}) + return "OK: 已回答" + + 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_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标题" + + await sor.C("sd_bugs", { + "id": getID(), "title": title, + "description": desc, "severity": severity, + "status": "open", + }) + 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", "") + # 简单估算：平均每 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", "")[:200] for m in messages) + + text = "\n".join( + f"[{m['role']}]: {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, + ) + 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 diff --git a/pipeline_service/init.py b/pipeline_service/init.py index e3b54f4..4d2b24b 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -512,6 +512,11 @@ def load_pipeline_service(): env.agent_loop = agent_loop # 向后兼容 env.run_agent_loop = run_agent_loop # 向后兼容 + # v2 Agent Executor (v3.5.0: 对照 Hermes Agent 架构) + from .agent_loop_v2 import AgentExecutor, run_agent + env.AgentExecutor = AgentExecutor + env.run_agent = run_agent + # PM agent (v3.3.0: 项目经理审核 + 任务链) env.pm_review_run = pm_review_run env.pm_review_loop = pm_review_loop