From 0270922205cf75f1ce8b85016d67199b3c2f8701 Mon Sep 17 00:00:00 2001 From: ymq Date: Sat, 8 Aug 2026 10:36:04 +0800 Subject: [PATCH] v3.3.0: PM agent + task chain + workspace + file output --- pipeline_service/agent_loop.py | 376 +++++++++++++++++++++++++++++++-- pipeline_service/init.py | 38 +++- pipeline_service/state.py | 2 + 3 files changed, 394 insertions(+), 22 deletions(-) diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index c3f652d..a397b61 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -1,9 +1,17 @@ -"""Agent loop — 角色agent循环认领执行任务（v3.2.0）。 +"""Agent loop — 角色agent循环认领执行任务（v3.3.0）。 + +v3.3.0 新增： +- PM（项目经理）角色：审核交付件，决定通过/驳回 +- 任务链：requirement → design → develop → test → deploy + 前序角色完成后 → PM审核通过 → 自动创建下一角色任务 +- 产物文件输出：交付件写入项目工作目录 deliverables/{role}/ +- 工作目录自动创建：确保 workspace_dir 存在 架构： - 主agent（cockpit_chat）理解客户意图，开发类意图写入 pipeline_tasks（带 role）。 - 各角色agent（需求/设计/开发/测试/运维）调 role_agent_run/role_agent_loop 循环认领自己角色的任务并执行。 +- PM agent 调 pm_review_loop 循环审核处于 review 状态的任务。 - 认领是原子的：UPDATE ... WHERE state='submitted' + claimed_by 令牌校验， 多agent并发不会重复认领。 - 角色agent缺信息时返回 need_info → 写入 pipeline_agent_questions， @@ -13,6 +21,7 @@ import asyncio import json +import os import logging logger = logging.getLogger("pipeline.agent_loop") @@ -25,8 +34,23 @@ ROLE_ALIASES = { 'testing': 'test', 'qa': 'test', 'tester': 'test', 'deployment': 'deploy', 'release': 'deploy', 'operation': 'ops', 'operations': 'ops', 'maintenance': 'ops', '运维': 'ops', + 'pm': 'pm', 'project_manager': 'pm', '项目经理': 'pm', 'manager': 'pm', } +# 任务链：前序角色完成后，PM审核通过 → 自动创建下一角色任务 +# None 表示链末端 +ROLE_CHAIN = { + 'requirement': 'design', + 'design': 'develop', + 'develop': 'test', + 'test': 'deploy', + 'deploy': None, # 部署完成 = 链结束 +} + +# PM 审核后任务状态 +TASK_REVIEW = 'review' # 角色agent完成，待PM审核 +TASK_APPROVED = 'approved' # PM审核通过 + def _normalize_role(role): r = (role or '').strip().lower() @@ -39,12 +63,59 @@ ROLE_PROMPT = """你是软件开发产线项目中的「__ROLE__」角色agent 标题：__TITLE__ 参数：__PARAMS__ __QNA__ + +## 项目信息 +- 工作目录：__WORKSPACE__ +- 你的产出物将保存到：__DELIVERABLE_DIR__ + 请站在你的角色立场完成这项工作。 +## 产出规范 +1. 产出内容必须完整、可交付、可直接使用 +2. 产出格式：根据角色不同，产出对应格式： + - requirement（需求分析）：需求规格文档（Markdown格式） + - design（设计）：设计文档 / 架构图描述（Markdown格式） + - develop（开发）：可运行的代码 + 代码说明 + - test（测试）：测试用例 + 测试报告（Markdown格式） + - deploy（部署）：部署文档 / 配置 / 脚本 + 输出要求——返回纯 JSON（不要 markdown 包裹）： -- 能完成时：{"status":"done","deliverable_type":"...","result":"完整交付内容"} +- 能完成时：{"status":"done","deliverable_type":"文档类型(requirement_doc/design_doc/code/test_report/deploy_doc)","result":"完整交付内容","summary":"一段话概述产出的关键结论"} - 缺少关键信息无法继续时：{"status":"need_info","question":"要向主agent/客户提出的具体问题","partial":"已完成的部分"}""" +PM_PROMPT = """你是软件开发产线项目中的「项目经理（PM）」角色agent。 + +你的职责是审核其他角色agent产出的交付件，确保质量和完整性，并推动项目前进。 + +## 任务链 +项目按照以下阶段推进： +1. requirement（需求分析）→ 2. design（设计）→ 3. develop（开发）→ 4. test（测试）→ 5. deploy（部署） + +当前待审核的任务： +标题：__TITLE__ +角色：__ROLE__ +交付件内容： +__DELIVERABLE__ + +## 你的决策 +请基于交付件内容做出判断，输出纯JSON（不要markdown包裹）： + +1. 审核通过，启动下一阶段： + {"status":"approved","comment":"审核意见","next_task_title":"下一阶段任务标题","next_task_description":"下一阶段任务详细描述"} + +2. 审核不通过，需要修改： + {"status":"rejected","comment":"驳回原因和具体修改要求","questions":"需要角色agent澄清的具体问题"} + +3. 项目已完成（deploy阶段通过后）： + {"status":"completed","comment":"项目完成总结"} + +## 审核标准 +- requirement：需求是否清晰、完整、可量化 +- design：方案是否合理、技术选型是否恰当、是否覆盖需求 +- develop：代码是否可运行、是否遵循规范、是否实现设计 +- test：测试覆盖是否充分、是否发现关键问题 +- deploy：部署文档是否完整、配置是否正确""" + def _get_db(): from sqlor.dbpools import DBPools @@ -57,22 +128,40 @@ def _get_db(): return db -async def _claim_task(sor, tenant_id: str, role: str): - """原子认领一条符合角色的 submitted 任务。返回 task record 或 None。 +def _resolve_workspace(workspace_dir): + """解析工作目录，如果不可写则回退到 /tmp。确保目录存在。""" + if not workspace_dir: + workspace_dir = '/tmp/pipeline_workspaces/default' + # 检测是否可写 + try: + parent = os.path.dirname(workspace_dir) + if parent and not os.access(parent, os.W_OK): + # 回退到 /tmp + proj_name = os.path.basename(workspace_dir) + workspace_dir = f'/tmp/pipeline_workspaces/{proj_name}' + except Exception: + workspace_dir = f'/tmp/pipeline_workspaces/{os.path.basename(workspace_dir) or "default"}' + # 确保目录存在 + os.makedirs(workspace_dir, exist_ok=True) + return workspace_dir + + +async def _claim_task(sor, tenant_id: str, role: str, state: str = 'submitted'): + """原子认领一条符合角色的任务。返回 task record 或 None。 认领策略： - 1. 选出最早一条 state='submitted' 且 role 匹配（或 role 为空）的任务， + 1. 选出最早一条符合状态和角色的任务， 且排除有步骤记录的 DAG 任务（那些归 executor）。 - 2. UPDATE ... SET state='running', claimed_by=令牌 WHERE id=? AND state='submitted' + 2. UPDATE ... SET state='running', claimed_by=令牌 WHERE id=? AND state=原状态 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='') " + "WHERE tenant_id=${tid}$ AND state=${state}$ 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}) + {"tid": tenant_id, "state": state, "role": role}) if not recs: return None @@ -83,8 +172,8 @@ async def _claim_task(sor, tenant_id: str, role: str): 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}) + "WHERE id=${tid}$ AND state=${state}$", + {"cb": claim_token, "tid": task_id, "state": state}) # 令牌校验：确认是本agent抢到的 check = await sor.sqlExe( @@ -96,6 +185,15 @@ async def _claim_task(sor, tenant_id: str, role: str): return task +async def _get_workspace_dir(sor, project_id): + """获取项目的工作目录，确保可写并存在。""" + recs = await sor.sqlExe( + "SELECT workspace_dir FROM sd_projects WHERE id=${pid}$", + {"pid": project_id}) + ws = getattr(recs[0], 'workspace_dir', '') if recs else '' + return _resolve_workspace(ws) + + async def _build_qna_section(sor, task_id: str) -> str: """把该任务已回答的问答历史注入prompt，让角色agent带着答案继续。""" from .questions import get_task_qna @@ -125,16 +223,90 @@ def _parse_result(raw: str) -> dict: return {"status": "done", "result": raw} +async def _write_deliverable_file(workspace_dir, role, task_id, deliverable_type, content): + """将交付件写入项目工作目录的文件中。""" + role_dir = os.path.join(workspace_dir, 'deliverables', role) + os.makedirs(role_dir, exist_ok=True) + # 文件名：任务ID_类型.md + safe_type = (deliverable_type or 'deliverable').replace('/', '_') + filename = f"{task_id}_{safe_type}.md" + file_path = os.path.join(role_dir, filename) + try: + with open(file_path, 'w', encoding='utf-8') as f: + f.write(content or '') + logger.info(f"deliverable written: {file_path}") + return file_path + except Exception as e: + logger.error(f"failed to write deliverable file: {file_path} err={e}") + return '' + + +async def _get_deliverable_content(sor, task_id): + """获取任务的最新交付件内容。""" + recs = await sor.sqlExe( + "SELECT content, deliverable_type FROM pipeline_deliverables " + "WHERE task_id=${tid}$ ORDER BY created_at DESC LIMIT 1", + {"tid": task_id}) + if recs: + return getattr(recs[0], 'content', '') or '', getattr(recs[0], 'deliverable_type', '') or '' + return '', '' + + +async def _get_next_role(current_role): + """获取任务链中的下一个角色。""" + return ROLE_CHAIN.get(_normalize_role(current_role)) + + +async def _create_next_task(sor, project_id, task, next_role, pm_comment=''): + """为下一个角色创建任务。""" + from appPublic.uniqueID import getID + title = getattr(task, 'title', '') or '' + params_str = getattr(task, 'params', '{}') or '{}' + try: + params = json.loads(params_str) if isinstance(params_str, str) else params_str + except (json.JSONDecodeError, TypeError): + params = {} + + new_title = f"{title}（{next_role}阶段）" + new_params = { + **params, + 'previous_role': _normalize_role(getattr(task, 'role', '')), + 'previous_task_id': getattr(task, 'id', ''), + 'pm_comment': pm_comment, + } + + new_task_id = getID() + await sor.C('pipeline_tasks', { + 'id': new_task_id, + 'tenant_id': project_id, + 'pipeline_id': 'role_task', + 'title': new_title, + 'params': json.dumps(new_params, ensure_ascii=False), + 'role': next_role, + 'state': 'submitted', + 'claimed_by': None, + }) + logger.info(f"_create_next_task: {next_role} task={new_task_id} from {getattr(task, 'id', '')}") + return new_task_id, new_title + + async def role_agent_run(project_id: str, role: str, agent_id: str = None, model_name: str = None) -> dict: - """角色agent单次迭代：认领一条任务 → 执行 → 交付或提问。 + """角色agent单次迭代：认领一条任务 → 执行 → 交付。 + + 非PM角色完成 → 任务置 review（等PM审核） + PM角色 → 由 pm_review_run 处理 Returns: {"status": "idle"} 没有待办任务 - {"status": "completed", "task_id", "deliverable_id"} 完成交付 + {"status": "completed", "task_id", "deliverable_id"} 完成交付（已进入review） {"status": "need_info", "task_id", "question_id", "question"} 缺信息已提问 {"status": "failed", "task_id", "error"} 执行失败 """ role = _normalize_role(role) + if role == 'pm': + # PM不执行角色任务，走pm_review_run + return {"status": "idle", "message": "PM agent请使用 pm_review_run"} + db = _get_db() async with db.sqlorContext("pipeline") as sor: task = await _claim_task(sor, project_id, role) @@ -145,12 +317,19 @@ async def role_agent_run(project_id: str, role: str, agent_id: str = None, model title = getattr(task, "title", "") or "" params_str = getattr(task, "params", "{}") or "{}" + # 获取工作目录 + workspace_dir = await _get_workspace_dir(sor, project_id) + deliverable_dir = os.path.join(workspace_dir, 'deliverables', role) + os.makedirs(deliverable_dir, exist_ok=True) + 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)) + .replace('__QNA__', qna_section) + .replace('__WORKSPACE__', workspace_dir) + .replace('__DELIVERABLE_DIR__', deliverable_dir)) from .llm_bridge import llm_call try: @@ -172,22 +351,34 @@ async def role_agent_run(project_id: str, role: str, agent_id: str = None, model return {"status": "need_info", "task_id": task_id, "question_id": qid, "question": parsed["question"]} - # 完成 → 写交付件 + # 完成 → 写交付件到文件 + DB from appPublic.uniqueID import getID did = getID() result_text = parsed.get("result") or raw + deliverable_type = parsed.get("deliverable_type") or role + summary = parsed.get("summary", "") + + # 写文件 + file_path = await _write_deliverable_file( + workspace_dir, role, task_id, deliverable_type, result_text) + + # 写DB await sor.C("pipeline_deliverables", { "id": did, "project_id": project_id, "task_id": task_id, - "deliverable_type": parsed.get("deliverable_type") or role, + "deliverable_type": deliverable_type, "title": title, "content": result_text, + "file_path": file_path, "quality_score": 80, "review_status": "pending", "created_by": agent_id or role, }) + + # 置为 review 状态，等待PM审核 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} + "UPDATE pipeline_tasks SET state=${st}$ WHERE id=${tid}$", + {"st": TASK_REVIEW, "tid": task_id}) + logger.info(f"role_agent_run completed(review): task={task_id} role={role} deliverable={did} file={file_path}") + return {"status": "completed", "task_id": task_id, "deliverable_id": did, + "next_state": TASK_REVIEW, "file_path": file_path} async def role_agent_loop(project_id: str, role: str, agent_id: str = None, @@ -203,14 +394,157 @@ async def role_agent_loop(project_id: str, role: str, agent_id: str = None, return results +# ── PM 审核 ── + +async def pm_review_run(project_id: str, agent_id: str = None, model_name: str = None) -> dict: + """PM审核单次迭代：认领一条 review 状态的任务 → 审核交付件 → 批准/驳回。 + + Returns: + {"status": "idle"} 没有待审核任务 + {"status": "approved", "task_id", "next_task_id", "next_role"} 审核通过，已创建下一阶段任务 + {"status": "rejected", "task_id"} 驳回，任务回到 submitted + {"status": "completed", "task_id"} 项目完成（链末端） + """ + db = _get_db() + async with db.sqlorContext("pipeline") as sor: + # 认领 review 状态的任务（不限角色，PM审核所有角色产出） + task = await _claim_task(sor, project_id, 'pm', state=TASK_REVIEW) + if not task: + return {"status": "idle", "message": "没有待审核任务"} + + task_id = task.id + title = getattr(task, "title", "") or "" + task_role = _normalize_role(getattr(task, "role", "") or "") + + # 获取交付件内容 + deliverable_content, deliverable_type = await _get_deliverable_content(sor, task_id) + if not deliverable_content: + # 没有交付件 → 驳回 + await sor.sqlExe( + "UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$", + {"tid": task_id}) + return {"status": "rejected", "task_id": task_id, + "reason": "没有找到交付件内容，任务已退回"} + + # 截断过长内容（LLM有token上限） + content_preview = deliverable_content[:6000] + if len(deliverable_content) > 6000: + content_preview += "\n\n...（内容过长已截断）" + + prompt = (PM_PROMPT + .replace('__TITLE__', title) + .replace('__ROLE__', task_role) + .replace('__DELIVERABLE__', content_preview)) + + from .llm_bridge import llm_call + try: + raw = await llm_call(prompt, model=model_name, temperature=0.3) + except Exception as e: + logger.error(f"pm_review_run llm failed: task={task_id} err={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]} + + parsed = _parse_result(raw) + decision = parsed.get("status", "rejected") + comment = parsed.get("comment", "") + + if decision == "approved": + # 获取工作目录 + workspace_dir = await _get_workspace_dir(sor, project_id) + + # 写PM审核记录到交付件 + 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), + "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", + }) + + # 更新原交付件审核状态 + await sor.sqlExe( + "UPDATE pipeline_deliverables SET review_status='approved', " + "review_comment=${cm}$ WHERE task_id=${tid}$", + {"cm": comment, "tid": task_id}) + + # 任务置为 approved + 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) + logger.info(f"pm_review: approved task={task_id}, created next={next_tid} role={next_role}") + return {"status": "approved", "task_id": task_id, + "next_task_id": next_tid, "next_role": next_role, + "next_title": next_title, "comment": comment} + else: + # 链末端（deploy审核通过 = 项目完成） + logger.info(f"pm_review: project completed, task={task_id} role={task_role}") + return {"status": "completed", "task_id": task_id, + "comment": comment or "项目所有阶段已完成"} + + elif decision == "rejected": + # 驳回：任务回到 submitted，附带PM驳回原因作为question + from .questions import agent_ask + rejection_question = parsed.get("questions") or comment or "交付件不满足要求，请修改后重新提交" + await agent_ask(project_id, task_id, "pm", rejection_question, + context={"pm_comment": comment, "deliverable_type": deliverable_type}) + # 任务回到 submitted（通过回答问题的resume机制） + # 但agent_ask已经把任务置waiting，需要让主agent/客户确认后才能回到submitted + # 简化为：直接回 submitted + await sor.sqlExe( + "UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$", + {"tid": task_id}) + logger.info(f"pm_review: rejected task={task_id}") + return {"status": "rejected", "task_id": task_id, + "comment": comment, "question": rejection_question} + + else: + # completed（项目完成） + await sor.sqlExe( + "UPDATE pipeline_tasks SET state='completed' WHERE id=${tid}$", + {"tid": task_id}) + return {"status": "completed", "task_id": task_id, + "comment": comment or "项目完成"} + + +async def pm_review_loop(project_id: str, agent_id: str = None, + model_name: str = None, max_iterations: int = 20) -> list: + """PM审核循环：连续审核处于 review 状态的任务。""" + results = [] + for _ in range(max_iterations): + r = await pm_review_run(project_id, agent_id, model_name) + results.append(r) + if r["status"] in ("idle", "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) + role = _normalize_role(role_name or "develop") + if role == 'pm': + return await pm_review_run(project_id, model_name=model_name) + return await role_agent_run(project_id, role_name or "develop", 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", + role = _normalize_role(role_name or "develop") + if role == 'pm': + return await pm_review_loop(project_id, model_name=model_name, max_iterations=max_iterations) + return await role_agent_loop(project_id, role_name or "develop", model_name=model_name, max_iterations=max_iterations) diff --git a/pipeline_service/init.py b/pipeline_service/init.py index cc897ab..029e447 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -16,6 +16,7 @@ from appPublic.log import debug from .state import ( TASK_SUBMITTED, TASK_RUNNING, TASK_COMPLETED, TASK_FAILED, TASK_PAUSED, TASK_CANCELLED, TASK_WAITING, + TASK_REVIEW, TASK_APPROVED, build_step_graph, get_cascade_rerun_steps, get_rerun_from_next, ) from .storage import ( @@ -37,9 +38,10 @@ from .questions import ( get_question, list_questions, get_task_qna, ) from .agent_loop import role_agent_run, role_agent_loop, agent_loop, run_agent_loop +from .agent_loop import pm_review_run, pm_review_loop MODULE_NAME = "pipeline_service" -MODULE_VERSION = "3.2.0" +MODULE_VERSION = "3.3.0" async def pipeline_submit(tenant_id, pipeline_id, owner_id, title, params=None): @@ -508,6 +510,10 @@ def load_pipeline_service(): env.agent_loop = agent_loop # 向后兼容 env.run_agent_loop = run_agent_loop # 向后兼容 + # PM agent (v3.3.0: 项目经理审核 + 任务链) + env.pm_review_run = pm_review_run + env.pm_review_loop = pm_review_loop + # Question loop (角色agent ↔ 主agent ↔ 客户) env.agent_ask = agent_ask env.question_answer = answer_question @@ -554,5 +560,35 @@ def load_pipeline_service(): from ahserver.configuredServer import add_startup add_startup(_role_poller) + # PM review poller (v3.3.0): auto-dispatch review-state tasks to PM agent + async def _pm_poller(app): + from sqlor.dbpools import DBPools as _DBP + pm_db = _DBP() + + async def _pm_poll_loop(): + while True: + try: + async with pm_db.sqlorContext("pipeline") as sor: + recs = await sor.sqlExe( + "SELECT id, tenant_id, role FROM pipeline_tasks " + "WHERE state='review' AND claimed_by IS NULL " + "ORDER BY created_at ASC LIMIT 5", + {}) + for rec in (recs or []): + tid = getattr(rec, 'id', '') + pid = getattr(rec, 'tenant_id', '') + if tid and pid: + debug(f"pm_poller dispatching review task={tid}") + asyncio.ensure_future( + pm_review_run(pid, agent_id="pm-poller")) + except Exception as e: + debug(f"pm_poller error: {e}") + await asyncio.sleep(15) + + asyncio.create_task(_pm_poll_loop()) + debug("pm poller started") + + add_startup(_pm_poller) + debug(f"[{MODULE_NAME}] v{MODULE_VERSION} loaded — pipeline engine with role-agent + question loop") return True diff --git a/pipeline_service/state.py b/pipeline_service/state.py index 9bfaead..9cfca0c 100644 --- a/pipeline_service/state.py +++ b/pipeline_service/state.py @@ -20,6 +20,8 @@ TASK_FAILED = "failed" TASK_PAUSED = "paused" TASK_CANCELLED = "cancelled" TASK_WAITING = "waiting" # 等待人工步骤完成 +TASK_REVIEW = "review" # 角色agent完成，待PM审核 +TASK_APPROVED = "approved" # PM审核通过 # Human task statuses HUMAN_PENDING = "pending"