"""Agent loop — 角色agent循环认领执行任务（v3.2.0）。 架构： - 主agent（cockpit_chat）理解客户意图，开发类意图写入 pipeline_tasks（带 role）。 - 各角色agent（需求/设计/开发/测试/运维）调 role_agent_run/role_agent_loop 循环认领自己角色的任务并执行。 - 认领是原子的：UPDATE ... WHERE state='submitted' + claimed_by 令牌校验， 多agent并发不会重复认领。 - 角色agent缺信息时返回 need_info → 写入 pipeline_agent_questions， 任务置 waiting，等主agent/客户回答后回到 submitted 重新认领继续。 - 只认领没有步骤记录的任务（带步骤的 DAG 任务归 executor 管）。 """ import asyncio import json import logging logger = logging.getLogger("pipeline.agent_loop") # 角色别名归一化：前端/主agent传的角色名统一到认领用的规范名 ROLE_ALIASES = { 'developer': 'develop', 'dev': 'develop', 'development': 'develop', 'coding': 'develop', 'requirement': 'requirement', 'requirements': 'requirement', 'requirement_analysis': 'requirement', 'designer': 'design', 'ui': 'design', 'ux': 'design', 'testing': 'test', 'qa': 'test', 'tester': 'test', 'deployment': 'deploy', 'release': 'deploy', 'operation': 'ops', 'operations': 'ops', 'maintenance': 'ops', '运维': 'ops', } def _normalize_role(role): r = (role or '').strip().lower() return ROLE_ALIASES.get(r, r) ROLE_PROMPT = """你是软件开发产线项目中的「__ROLE__」角色agent。 当前认领到的任务： 标题：__TITLE__ 参数：__PARAMS__ __QNA__ 请站在你的角色立场完成这项工作。 输出要求——返回纯 JSON（不要 markdown 包裹）： - 能完成时：{"status":"done","deliverable_type":"...","result":"完整交付内容"} - 缺少关键信息无法继续时：{"status":"need_info","question":"要向主agent/客户提出的具体问题","partial":"已完成的部分"}""" def _get_db(): from sqlor.dbpools import DBPools db = DBPools() if not db.databases: from appPublic.jsonConfig import getConfig config = getConfig() if config.databases: db.databases = config.databases return db async def _claim_task(sor, tenant_id: str, role: str): """原子认领一条符合角色的 submitted 任务。返回 task record 或 None。 认领策略： 1. 选出最早一条 state='submitted' 且 role 匹配（或 role 为空）的任务， 且排除有步骤记录的 DAG 任务（那些归 executor）。 2. UPDATE ... SET state='running', claimed_by=令牌 WHERE id=? AND state='submitted' 3. 用令牌回查校验——别的agent抢先时校验失败，返回 None。 """ role = _normalize_role(role) recs = await sor.sqlExe( "SELECT id, title, params, pipeline_id, tenant_id, role FROM pipeline_tasks " "WHERE tenant_id=${tid}$ AND state='submitted' AND (role=${role}$ OR role='') " "AND NOT EXISTS (SELECT 1 FROM pipeline_task_steps s WHERE s.task_id=pipeline_tasks.id) " "ORDER BY created_at ASC LIMIT 1", {"tid": tenant_id, "role": role}) if not recs: return None task = recs[0] task_id = task.id from appPublic.uniqueID import getID claim_token = getID() await sor.sqlExe( "UPDATE pipeline_tasks SET state='running', claimed_by=${cb}$ " "WHERE id=${tid}$ AND state='submitted'", {"cb": claim_token, "tid": task_id}) # 令牌校验：确认是本agent抢到的 check = await sor.sqlExe( "SELECT id FROM pipeline_tasks WHERE id=${tid}$ AND state='running' AND claimed_by=${cb}$", {"tid": task_id, "cb": claim_token}) if not check: logger.info(f"claim lost race: task={task_id} role={role}") return None return task async def _build_qna_section(sor, task_id: str) -> str: """把该任务已回答的问答历史注入prompt，让角色agent带着答案继续。""" from .questions import get_task_qna qna = await get_task_qna(task_id) if not qna: return "" lines = ["历史问答（这些问题已得到回答，请结合答案继续工作）："] for item in qna: src = item.get('answer_source') or '' src_label = "客户" if src == "customer" else "主agent" lines.append(f"问：{item.get('question', '')}") lines.append(f"答（{src_label}）：{item.get('answer', '')}") return "\n".join(lines) def _parse_result(raw: str) -> dict: """解析角色agent的 LLM 返回。解析失败时按 done + 原文处理。""" raw = (raw or "").strip() if raw.startswith("```"): raw = raw.split("\n", 1)[1].rsplit("```", 1)[0] try: d = json.loads(raw) if isinstance(d, dict): return d except (json.JSONDecodeError, ValueError): pass return {"status": "done", "result": raw} async def role_agent_run(project_id: str, role: str, agent_id: str = None, model_name: str = None) -> dict: """角色agent单次迭代：认领一条任务 → 执行 → 交付或提问。 Returns: {"status": "idle"} 没有待办任务 {"status": "completed", "task_id", "deliverable_id"} 完成交付 {"status": "need_info", "task_id", "question_id", "question"} 缺信息已提问 {"status": "failed", "task_id", "error"} 执行失败 """ role = _normalize_role(role) db = _get_db() async with db.sqlorContext("pipeline") as sor: task = await _claim_task(sor, project_id, role) if not task: return {"status": "idle", "message": "没有待办任务"} task_id = task.id title = getattr(task, "title", "") or "" params_str = getattr(task, "params", "{}") or "{}" qna_section = await _build_qna_section(sor, task_id) prompt = (ROLE_PROMPT .replace('__ROLE__', role) .replace('__TITLE__', title) .replace('__PARAMS__', params_str) .replace('__QNA__', qna_section)) from .llm_bridge import llm_call try: raw = await llm_call(prompt, model=model_name, temperature=0.4) except Exception as e: await sor.sqlExe( "UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", {"tid": task_id}) logger.error(f"role_agent_run llm failed: task={task_id} err={e}") return {"status": "failed", "task_id": task_id, "error": str(e)[:200]} parsed = _parse_result(raw) # 缺信息 → 提问回路：任务置 waiting，等主agent/客户回答 if parsed.get("status") == "need_info" and parsed.get("question"): from .questions import agent_ask qid = await agent_ask(project_id, task_id, role, parsed["question"], context={"partial": parsed.get("partial", ""), "title": title}) return {"status": "need_info", "task_id": task_id, "question_id": qid, "question": parsed["question"]} # 完成 → 写交付件 from appPublic.uniqueID import getID did = getID() result_text = parsed.get("result") or raw await sor.C("pipeline_deliverables", { "id": did, "project_id": project_id, "task_id": task_id, "deliverable_type": parsed.get("deliverable_type") or role, "title": title, "content": result_text, "quality_score": 80, "review_status": "pending", "created_by": agent_id or role, }) await sor.sqlExe( "UPDATE pipeline_tasks SET state='completed' WHERE id=${tid}$", {"tid": task_id}) logger.info(f"role_agent_run completed: task={task_id} role={role} deliverable={did}") return {"status": "completed", "task_id": task_id, "deliverable_id": did} async def role_agent_loop(project_id: str, role: str, agent_id: str = None, model_name: str = None, max_iterations: int = 10) -> list: """角色agent循环：连续认领执行，直到没有任务、需要提问或达到上限。""" results = [] for _ in range(max_iterations): r = await role_agent_run(project_id, role, agent_id, model_name) results.append(r) if r["status"] in ("idle", "need_info", "failed"): break await asyncio.sleep(1) return results # ── 向后兼容别名（旧前端按 run_agent_loop/agent_loop 调用）── async def run_agent_loop(project_id, role_name, model_name=None): """旧接口：单次迭代。role_name 为空时按 'dev' 处理。""" return await role_agent_run(project_id, role_name or "dev", model_name=model_name) async def agent_loop(project_id, role_name, model_name=None, max_iterations=10): """旧接口：循环执行。role_name 为空时按 'dev' 处理。""" return await role_agent_loop(project_id, role_name or "dev", model_name=model_name, max_iterations=max_iterations)