1019 lines
38 KiB
Python
1019 lines
38 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 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._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
|
||
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
|
||
|
||
# 注入当前项目
|
||
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}"
|
||
|
||
# 注入技能（三级隔离：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}"
|
||
|
||
# 注入工具列表
|
||
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,
|
||
"get_task": 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,
|
||
"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
|