diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index c6abc2b..5bbce51 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -256,6 +256,44 @@ __REPO_STATE__ - test：测试覆盖充分、发现问题记录完整 - deploy：部署配置完整、可一键部署""" +PM_SYSTEM_PROMPT = """你是项目经理（PM）。审核交付件并推动项目前进。 + +## 项目信息 +工作目录：__WORKSPACE__ +关联仓库：__REPOS__ + +## 待审核 +标题：__TITLE__ +角色：__ROLE__ + +## 仓库状态 +__REPO_STATE__ + +## 工具 +- read_file(path) — 读工作空间文件 +- list_files(path) — 列目录 +- git_status() — 查看git状态 +- run_shell(command) — 执行命令（编译/测试验证） + +## 审核流程 +1. 先用工具检查代码是否实际产出（git_status/list_files/read_file） +2. 如需编译验证，用 run_shell +3. 最后给出决策 + +## 输出格式（每次一个JSON） +查看文件：{"action":"tool_call","tool":"read_file","params":{"path":"相对路径"}} +查看git：{"action":"tool_call","tool":"git_status","params":{}} +批准：{"action":"review_approve","comment":"审核意见","next_task_title":"下阶段标题","next_task_description":"描述"} +驳回：{"action":"review_reject","comment":"原因","questions":"修改要求"} +完成：{"action":"review_complete","comment":"总结"} + +审核标准： +- requirement：需求是否清晰完整可量化 +- design：方案合理、覆盖需求、技术可行 +- develop：必须用 git_status/read_file 检查代码是否实际写入仓库 +- test：测试覆盖充分、发现问题记录完整 +- deploy：部署配置完整、可一键部署""" + # ── Agent 核心逻辑 ── @@ -631,13 +669,11 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): task_role = _normalize_role(getattr(task, "role", "") or "") workspace_dir = await _get_workspace_dir(sor, project_id) - # 提交 claim 事务，释放行锁，避免后续长操作（LLM调用）阻塞其他写入 try: await sor.sqlExe("COMMIT", {}) except Exception: pass - # 首次审核时，确保仓库已 clone repos = await _get_project_repos(sor, project_id) if repos: await _setup_repos(sor, workspace_dir, project_id) @@ -647,66 +683,79 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): await sor.sqlExe("UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$", {"tid": task_id}) return {"status": "rejected", "task_id": task_id, "reason": "没有交付件"} - summary = deliverable_content[:200].replace('\n', ' ') - content_preview = deliverable_content[:6000] - if len(deliverable_content) > 6000: - content_preview += "\n...（截断）" - repo_state = await _get_repo_state(workspace_dir) repos_str = ", ".join([r['name'] for r in repos]) if repos else "无" + content_preview = deliverable_content[:6000] - prompt = (PM_PROMPT - .replace('__TITLE__', title) - .replace('__ROLE__', task_role) - .replace('__SUMMARY__', summary) - .replace('__DELIVERABLE__', content_preview) - .replace('__WORKSPACE__', workspace_dir) - .replace('__REPOS__', repos_str) - .replace('__REPO_STATE__', repo_state)) + pm_system = PM_SYSTEM_PROMPT.replace('__TITLE__', title)\ + .replace('__ROLE__', task_role)\ + .replace('__WORKSPACE__', workspace_dir)\ + .replace('__REPOS__', repos_str)\ + .replace('__REPO_STATE__', repo_state) - from .llm_bridge import llm_call - try: - raw = await llm_call(prompt, model=model_name, temperature=0.3) - except Exception as e: - await sor.sqlExe("UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", {"tid": task_id}) - return {"status": "failed", "task_id": task_id, "error": str(e)[:200]} + msgs = [{"role": "system", "content": pm_system}] + msgs.append({"role": "user", "content": f"请审核以下交付件（类型:{deliverable_type}）：\n\n{content_preview}"}) - parsed = _parse_result(raw) - decision = parsed.get("status", "rejected") - comment = parsed.get("comment", "") + from .llm_bridge import llm_call_msgs - if decision == "approved": + decision = None + for turn in range(5): + try: + raw = await llm_call_msgs(msgs, model=model_name, temperature=0.3) + except Exception as e: + await sor.sqlExe("UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", {"tid": task_id}) + return {"status": "failed", "task_id": task_id, "error": str(e)[:200]} + + act = _parse_agent_action(raw) + + if act.get('action') == 'review_approve': + decision = {'status': 'approved', 'comment': act.get('comment', ''), 'next_title': act.get('next_task_title', ''), 'next_desc': act.get('next_task_description', '')} + break + elif act.get('action') == 'review_reject': + decision = {'status': 'rejected', 'comment': act.get('comment', ''), 'questions': act.get('questions', '')} + break + elif act.get('action') == 'review_complete': + decision = {'status': 'completed', 'comment': act.get('comment', '')} + break + elif act.get('action') == 'tool_call': + tool = act.get('tool', '') + params = act.get('params', {}) + result = await _exec_agent_tool(tool, params, workspace_dir) + msgs.append({"role": "assistant", "content": raw}) + msgs.append({"role": "user", "content": f"工具 {tool} 结果：\n{result}"}) + else: + decision = {'status': 'completed', 'comment': raw[:200]} + break + + if not decision: + decision = {'status': 'rejected', 'comment': '审核超时'} + + status = decision['status'] + comment = decision.get('comment', '') + + if status == 'approved': from appPublic.uniqueID import getID pm_did = getID() await sor.C("pipeline_deliverables", { "id": pm_did, "project_id": project_id, "task_id": task_id, - "deliverable_type": "pm_review", - "title": f"PM审核：{title}", - "content": json.dumps(parsed, ensure_ascii=False), + "deliverable_type": "pm_review", "title": f"PM审核：{title}", + "content": json.dumps(decision, ensure_ascii=False), "file_path": os.path.join(workspace_dir, 'deliverables', 'pm', f"{task_id}_review.md"), - "quality_score": 100, "review_status": "approved", - "created_by": agent_id or "pm", + "quality_score": 100, "review_status": "approved", "created_by": agent_id or "pm", }) - await sor.sqlExe( - "UPDATE pipeline_deliverables SET review_status='approved', review_comment=${cm}$ " - "WHERE task_id=${tid}$", {"cm": comment, "tid": task_id}) - await sor.sqlExe( - "UPDATE pipeline_tasks SET state=${st}$ WHERE id=${tid}$", - {"st": TASK_APPROVED, "tid": task_id}) - + await sor.sqlExe("UPDATE pipeline_deliverables SET review_status='approved', review_comment=${cm}$ WHERE task_id=${tid}$", {"cm": comment, "tid": task_id}) + await sor.sqlExe("UPDATE pipeline_tasks SET state=${st}$ WHERE id=${tid}$", {"st": TASK_APPROVED, "tid": task_id}) next_role = await _get_next_role(task_role) if next_role: next_tid, next_title = await _create_next_task(sor, project_id, task, next_role, comment) - return {"status": "approved", "task_id": task_id, - "next_task_id": next_tid, "next_role": next_role, "comment": comment} + return {"status": "approved", "task_id": task_id, "next_task_id": next_tid, "next_role": next_role, "comment": comment} else: return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"} - elif decision == "rejected": + elif status == 'rejected': from .questions import agent_ask - rejection_q = parsed.get("questions") or comment or "交付件不满足要求" - await agent_ask(project_id, task_id, "pm", rejection_q, - context={"pm_comment": comment, "deliverable_type": deliverable_type}) + rejection_q = decision.get("questions") or comment or "交付件不满足要求" + await agent_ask(project_id, task_id, "pm", rejection_q, context={"pm_comment": comment, "deliverable_type": deliverable_type}) await sor.sqlExe("UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$", {"tid": task_id}) return {"status": "rejected", "task_id": task_id, "comment": comment, "question": rejection_q}