From ab776d0c3749ba1611561cd14e3086d3d55f8969 Mon Sep 17 00:00:00 2001 From: ymq Date: Tue, 18 Aug 2026 15:16:43 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=20QC=20=E8=A7=92?= =?UTF-8?q?=E8=89=B2=EF=BC=88=E8=B4=A8=E9=87=8F=E9=97=A8=E7=A6=81=EF=BC=89?= =?UTF-8?q?+=20=E9=83=A8=E7=BD=B2=E6=8B=86=E5=88=86=E4=B8=BA=20deploy=5Fte?= =?UTF-8?q?st/deploy=5Fprod=20+=20requirement=20=E6=98=8E=E7=A1=AE?= =?UTF-8?q?=E9=83=A8=E7=BD=B2=E7=8E=AF=E5=A2=83=E9=9C=80=E6=B1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - SDL_ROLES 6 环新链：requirement→design→develop→deploy_test→test→deploy_prod，新增 agent.qc - 状态机新增 qc_review：角色交付→QC 合规/质量检查→PM 验收→下一环 - qc_review_run + qc poller；不合规退回重做 - requirement 职责增加部署环境需求明确 --- pipeline_service/agent_loop.py | 144 ++++++++++++++++++++++++++-- pipeline_service/init.py | 53 +++++++++- pipeline_service/sdlc_ability.py | 69 ++++++++++--- pipeline_service/task_capability.py | 19 +++- 4 files changed, 261 insertions(+), 24 deletions(-) diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 05d896f..7131a16 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -26,7 +26,9 @@ ROLE_ALIASES = { 'requirement': 'agent.requirement', 'requirements': 'agent.requirement', 'requirement_analysis': 'agent.requirement', 'designer': 'agent.design', 'ui': 'agent.design', 'ux': 'agent.design', 'testing': 'agent.test', 'qa': 'agent.test', 'tester': 'agent.test', - 'deployment': 'agent.deploy', 'release': 'agent.deploy', + 'deployment': 'agent.deploy_prod', 'release': 'agent.deploy_prod', 'production': 'agent.deploy_prod', + 'staging': 'agent.deploy_test', 'deploy_staging': 'agent.deploy_test', + 'qc': 'agent.qc', 'quality': 'agent.qc', 'quality_control': 'agent.qc', '质量控制': 'agent.qc', 'operation': 'agent.ops', 'operations': 'agent.ops', 'maintenance': 'agent.ops', '运维': 'agent.ops', 'pm': 'agent.pm', 'project_manager': 'agent.pm', '项目经理': 'agent.pm', 'manager': 'agent.pm', } @@ -34,9 +36,10 @@ ROLE_ALIASES = { ROLE_CHAIN = { 'agent.requirement': 'agent.design', 'agent.design': 'agent.develop', - 'agent.develop': 'agent.test', - 'agent.test': 'agent.deploy', - 'agent.deploy': None, + 'agent.develop': 'agent.deploy_test', + 'agent.deploy_test': 'agent.test', + 'agent.test': 'agent.deploy_prod', + 'agent.deploy_prod': None, } TASK_REVIEW = 'review' @@ -461,11 +464,42 @@ __REPO_STATE__ 完成：{"action":"review_complete","comment":"总结"} 审核标准： -- requirement：需求是否清晰完整可量化 +- requirement：需求是否清晰完整可量化（含部署环境需求是否明确） - design：方案合理、覆盖需求、技术可行 - develop：必须用 git_status/read_file 检查代码是否实际写入仓库 +- deploy_test：测试环境部署配置完整、服务可访问 - test：测试覆盖充分、发现问题记录完整 -- deploy：部署配置完整、可一键部署""" +- deploy_prod：生产部署配置完整、可一键部署、有回滚方案""" + + +QC_SYSTEM_PROMPT = """你是质量控制工程师（QC）。对交付件做合规检查和质量检查，不合规直接退回重做。 + +## 检查维度 +1. 项目规范检查：产出是否按 SDLC 仓库标准路径/命名/格式产出（docs/ 目录结构、apps/modules 描述文件、文件命名规范） +2. 项目过程规范：是否遵循各阶段流程规范（develop 是否实际 git 提交、test 是否覆盖充分、deploy 配置是否完整、requirement 是否明确部署环境需求） +3. 产出质量：内容是否完整、可量化、可验收、无重大缺陷、无空泛套话 + +## 待检查 +标题：__TITLE__ +角色：__ROLE__ +工作目录：__WORKSPACE__ + +## 工具 +- read_file(path) — 读工作空间文件 +- list_files(path) — 列目录 +- git_status() — 查看git状态 +- run_shell(command) — 执行命令（编译/测试验证） + +## 输出格式（每次一个JSON） +查看文件：{"action":"tool_call","tool":"read_file","params":{"path":"相对路径"}} +检查通过：{"action":"review_approve","comment":"合规，质量达标"} +不合规退回：{"action":"review_reject","comment":"退回原因","questions":"具体问题清单，逐条列出"} + +## 检查流程 +1. 读交付件内容，检查是否按 SDLC 规范产出（路径/命名/格式） +2. 检查过程合规（git 提交、目录结构、必填项） +3. 判断产出质量是否达标（完整、可量化、可验收） +4. 合规则 review_approve；不合规则 review_reject 并逐条列出问题清单""" # ── Agent 核心逻辑 ── @@ -1043,7 +1077,7 @@ async def _pm_list_tasks(sor, project_id, params): await sor.sqlExe("COMMIT", {}) if not recs: return "暂无任务" - icons = {"submitted": "⏳", "running": "🔄", "review": "👀", + icons = {"submitted": "⏳", "running": "🔄", "review": "👀", "qc_review": "🔍", "approved": "✅", "completed": "✔️", "failed": "❌", "waiting": "⏸️"} lines = ["| 状态 | 任务 | 角色 |", "|------|------|------|"] for r in recs: @@ -1218,6 +1252,102 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"} +async def qc_review_run(project_id, agent_id=None, model_name=None): + """QC 质量门禁：认领 qc_review 状态任务，做合规+质量检查，通过转 review，不合规退回重做。""" + from .task_capability import S_QC_REVIEW + db = _get_db() + async with db.sqlorContext("pipeline") as sor: + model_name, org_id = await _resolve_llm_context(sor, project_id, 'qc', model_name) + task = await _claim_task(sor, project_id, '', state=S_QC_REVIEW, match_role=False, set_state='qc_review') + if not task: + return {"status": "idle", "message": "没有待 QC 检查的任务"} + + task_id = task.id + title = getattr(task, "title", "") or "" + task_role = _normalize_role(getattr(task, "role", "") or "") + workspace_dir = await _get_workspace_dir(sor, project_id) + + try: + await sor.sqlExe("COMMIT", {}) + except Exception: + pass + + deliverable_content, deliverable_type = await _get_deliverable_content(sor, task_id) + if not deliverable_content: + from .task_capability import qc_reject_task + await qc_reject_task(task_id, project_id, who="agent.qc", agent_id=agent_id, comment="没有交付件") + return {"status": "rejected", "task_id": task_id, "reason": "没有交付件"} + + content_preview = deliverable_content[:6000] + + qc_system = QC_SYSTEM_PROMPT.replace('__TITLE__', title)\ + .replace('__ROLE__', task_role)\ + .replace('__WORKSPACE__', workspace_dir) + + msgs = [{"role": "system", "content": qc_system}] + msgs.append({"role": "user", "content": f"请检查以下交付件（类型:{deliverable_type}）：\n\n{content_preview}"}) + + from .llm_bridge import llm_call_msgs + + decision = None + for turn in range(4): + await sor.sqlExe( + "UPDATE pipeline_tasks SET updated_at=NOW() WHERE id=${tid}$ AND state='qc_review'", + {"tid": task_id}) + await sor.sqlExe("COMMIT", {}) + if turn >= 2: + msgs.append({"role": "user", "content": + "已检查足够信息。现在必须立即输出 review_approve 或 review_reject，禁止再调用其它工具。"}) + try: + raw = await llm_call_msgs(msgs, model=model_name, temperature=0.2, org_id=org_id) + except Exception as e: + err_msg = f"{type(e).__name__}: {str(e)[:400]}" + from .task_capability import mark_failed + await mark_failed(task_id, project_id, who="agent.qc", agent_id=agent_id, error=err_msg) + 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', '')} + break + elif act.get('action') == 'review_reject': + decision = {'status': 'rejected', 'comment': act.get('comment', ''), 'questions': act.get('questions', '')} + 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: + # 无明确决策 → 默认通过（避免 QC 卡死阻塞流程） + decision = {'status': 'approved', 'comment': raw[:200]} + break + + if not decision: + decision = {'status': 'approved', 'comment': 'QC 检查超时，默认通过'} + + if decision['status'] == 'approved': + from .task_capability import qc_approve_task + await qc_approve_task(task_id, project_id, who="agent.qc", agent_id=agent_id, comment=decision.get('comment', '')) + logger.info(f"qc_review_run approve: task={task_id} -> review") + return {"status": "approved", "task_id": task_id, "comment": decision.get('comment', '')} + else: + comment = decision.get('comment', '') + from .communication import raise_problem + rejection_q = decision.get("questions") or comment or "交付件不合规" + await raise_problem("qc_reject", rejection_q, "agent.qc", + tenant_id=project_id, task_id=task_id, + first_handler_role=task_role, + context={"qc_comment": comment, "deliverable_type": deliverable_type}, + suspend_task=False) + from .task_capability import qc_reject_task + await qc_reject_task(task_id, project_id, who="agent.qc", agent_id=agent_id, comment=comment) + logger.info(f"qc_review_run reject: task={task_id} -> submitted") + return {"status": "rejected", "task_id": task_id, "comment": comment, "question": rejection_q} + + # ── 失败任务处理（第二层：PM/cockpit 判定重跑，最多3次，超限报故障给用户）── _RETRY_HINTS = ( diff --git a/pipeline_service/init.py b/pipeline_service/init.py index 3b1dd94..bb78b56 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -48,7 +48,7 @@ from .task_capability import ( revive_task, set_task_state, ) 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, handle_failed_task +from .agent_loop import pm_review_run, pm_review_loop, qc_review_run, handle_failed_task MODULE_NAME = "pipeline_service" MODULE_VERSION = "3.4.0" @@ -540,6 +540,7 @@ def load_pipeline_service(): # PM agent (v3.3.0: 项目经理审核 + 任务链) env.pm_review_run = pm_review_run env.pm_review_loop = pm_review_loop + env.qc_review_run = qc_review_run # Question loop (角色agent ↔ 主agent ↔ 客户) env.agent_ask = agent_ask @@ -693,6 +694,56 @@ def load_pipeline_service(): add_startup(_pm_poller) + # QC review poller: auto-dispatch qc_review-state tasks to QC agent（合规/质量门禁） + async def _qc_poller(app): + from sqlor.dbpools import DBPools as _DBP2 + qc_db = _DBP2() + _qc_dispatched = set() + + async def _qc_poll_once(qc_db): + """执行一轮 QC poll。可能因连接获取/SQL 锁等待卡住，由外层 wait_for 超时保护。""" + async with qc_db.sqlorContext("pipeline") as sor: + # 回收僵尸 qc_review 任务：QC 检查进程崩溃后 claimed_by 残留，重新放回检查队列。 + await sor.sqlExe( + "UPDATE pipeline_tasks SET claimed_by=NULL, updated_at=NOW() " + "WHERE state='qc_review' AND claimed_by IS NOT NULL " + "AND updated_at < (NOW() - INTERVAL 20 MINUTE)", {}) + recs = await sor.sqlExe( + "SELECT id, tenant_id, role FROM pipeline_tasks " + "WHERE state='qc_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 and tid not in _qc_dispatched: + _qc_dispatched.add(tid) + debug(f"qc_poller dispatching qc_review task={tid}") + + async def _qc_dispatch(tid, pid): + try: + await qc_review_run(pid, agent_id="qc-poller") + except Exception as e: + debug(f"qc_poller dispatch error: task={tid} err={e}") + finally: + _qc_dispatched.discard(tid) + + asyncio.ensure_future(_qc_dispatch(tid, pid)) + + async def _qc_poll_loop(): + while True: + try: + # watchdog：每轮 QC poll 最多 60 秒，超时则跳过本轮，防止 poller 永久停摆 + await asyncio.wait_for(_qc_poll_once(qc_db), timeout=60) + except Exception as e: + debug(f"qc_poller error/timeout: {e}") + await asyncio.sleep(15) + + asyncio.create_task(_qc_poll_loop()) + debug("qc poller started") + + add_startup(_qc_poller) + # Failed task poller: 失败任务自动重跑(最多3次)，超限报故障给用户 async def _failed_poller(app): from sqlor.dbpools import DBPools as _DBP3 diff --git a/pipeline_service/sdlc_ability.py b/pipeline_service/sdlc_ability.py index 597e649..45a8e17 100644 --- a/pipeline_service/sdlc_ability.py +++ b/pipeline_service/sdlc_ability.py @@ -200,10 +200,14 @@ SDL_TOOLS = [ # ── prompt 片段（追加到 core 通用心智之后） ── -SDL_PROMPT = """你是「开发产线」的驾驶舱 agent，负责软件项目的全流程管理（需求 → 设计 → 开发 → 测试 → 部署）。 +SDL_PROMPT = """你是「开发产线」的驾驶舱 agent，负责软件项目的全流程管理（需求 → 设计 → 开发 → 部署测试环境 → 测试 → 部署生产环境）。 + +## 部署环境需求（项目开始阶段必须明确） +- 除功能需求外，需求阶段还要明确部署环境需求：测试环境 + 生产环境的资源规格、依赖服务、端口、环境变量、高可用/备份/域名证书/监控告警等。 +- 用户未提供时，在会话中逐步追问明确（用 ask_user / list_questions），不要跳过；明确后再创建 requirement 任务。 ## 典型场景（帮助判断何时用哪个工具，非强制） -- 用户提出新的开发需求/功能（"实现XX""XX系统要做XX"）→ 用 create_task 创建任务（可按 requirement→design→develop 拆分），再 start_agents 启动。 +- 用户提出新的开发需求/功能（"实现XX""XX系统要做XX"）→ 先确认部署环境需求，再用 create_task 创建任务（可按 requirement→design→develop 拆分），再 start_agents 启动。 - 用户问进展/状态 → list_tasks / diagnose_project / check_progress。 - 用户问待回答问题 → list_questions / answer_question。 - 用户报告异常/故障 → 先用 diagnose_project 定位根因，再用 task_detail 查详情。 @@ -234,7 +238,11 @@ SDL_ROLES = [ - apps/<应用名>.md: 应用描述、部署环境、端口 - modules/<模块名>.md: 模块功能、仓库URL(先填写规划地址)、技术栈、依赖 - 每个应用至少关联一个模块 -内容: 项目概述、用户角色及权限、功能列表(每个功能:输入/处理/输出/验收标准)、非功能需求、业务流程。""", +内容: 项目概述、用户角色及权限、功能列表(每个功能:输入/处理/输出/验收标准)、非功能需求、业务流程。 +部署环境需求（项目开始阶段必须明确，含测试环境与生产环境）: + - 测试环境: 资源规格(CPU/内存/磁盘)、依赖服务(DB/缓存/中间件)、端口、环境变量 + - 生产环境: 资源规格、高可用要求、备份策略、域名/证书、监控告警 + - 若需求未明确部署环境信息，在需求文档中标注「待明确」，并列出需向用户确认的问题清单（后续会话逐步明确）。""", next_role="agent.design", ), RoleSpec( @@ -269,32 +277,67 @@ SDL_ROLES = [ 8. write_file 写 docs/02-develop/dev-notes.md 记录开发内容 9. deliver 交付，files 列出所有产出文件路径 PM审核: git_status 检查模块仓库有提交记录。""", + next_role="agent.deploy_test", + ), + RoleSpec( + name="agent.deploy_test", + description="测试环境部署工程师", + aliases=["deploy_staging", "staging"], + system_prompt="""你是测试环境部署工程师。把开发完成的代码部署到测试环境，供测试工程师验证。 + +产出路径: + - config/test/ 下: 测试环境 Dockerfile、docker-compose.yml、nginx 配置、部署脚本 + - docs/04-deploy/deploy-test-env.md: 测试环境部署步骤、环境配置、访问地址 +流程: + 1. read_file 读 docs/00-requirement/ 的部署环境需求 + docs/01-design/ 的设计 + 2. 准备测试环境部署配置（容器/依赖服务/端口/环境变量） + 3. 部署到测试环境，run_shell 验证服务可访问 + 4. 产出部署文档，deliver 交付，files 列出配置文件路径 +PM审核: 检查测试环境部署配置完整、服务可访问。""", next_role="agent.test", ), RoleSpec( name="agent.test", description="测试工程师", aliases=["testing", "qa", "tester"], - system_prompt="""你是测试工程师。按 SDLC 仓库标准产出文档。 + system_prompt="""你是测试工程师。在测试环境上执行测试，按 SDLC 仓库标准产出文档。 产出路径: docs/03-test/ 目录下: - test-plan.md: 测试策略(单元/集成/端到端) - test-cases.md: 用例清单(编号/前置条件/步骤/预期结果) - test-report.md: 执行结果、Bug清单、覆盖率 测试脚本放到 files。""", - next_role="agent.deploy", + next_role="agent.deploy_prod", ), RoleSpec( - name="agent.deploy", - description="部署运维工程师", - aliases=["deployment", "release"], - system_prompt="""你是部署运维工程师。按 SDLC 仓库标准产出文档和配置。 + name="agent.deploy_prod", + description="生产环境部署工程师", + aliases=["deploy_production", "production", "release"], + system_prompt="""你是生产环境部署工程师。测试通过后，把代码部署到生产环境。 产出路径: - - docs/04-deploy/deploy-guide.md: 环境要求、部署步骤、回滚方案 + - config/prod/ 下: 生产环境 Dockerfile、docker-compose.yml、nginx 配置 + - docs/04-deploy/deploy-prod-env.md: 生产部署步骤、回滚方案 - docs/04-deploy/release-notes.md: 发布说明 - - config/ 下: Dockerfile、nginx配置、docker-compose.yml 等 -产出后用 result 输出部署说明，files 列出所有配置文件路径。""", +流程: + 1. 确认测试验证通过（read_file 读 docs/03-test/test-report.md） + 2. 准备生产配置（高可用、备份、监控告警、域名证书） + 3. 部署到生产环境，验证服务可用 + 4. 产出部署文档 + 发布说明，deliver 交付，files 列出配置文件 +PM审核: 检查生产部署配置完整、可一键部署、有回滚方案。""", + next_role="", + ), + RoleSpec( + name="agent.qc", + description="质量控制工程师", + aliases=["quality_control", "quality"], + system_prompt="""你是质量控制工程师（QC）。对每项产出做合规检查和质量检查，不合规直接退回重做。 + +检查维度: + 1. 项目规范检查: 产出是否按 SDLC 仓库标准路径/命名/格式产出 + 2. 项目过程规范: 是否遵循各阶段流程规范（develop 是否实际 git 提交、test 是否覆盖充分、deploy 配置是否完整） + 3. 产出质量: 内容是否完整、可量化、可验收、无重大缺陷 +不合规处理: review_reject 退回重做，questions 列出具体问题清单。""", next_role="", ), ] @@ -368,7 +411,7 @@ async def _h_list_tasks(sor, p, ctx): icons = { "submitted": "⏳", "running": "🔄", "review": "👀", "approved": "✅", "completed": "✔️", "failed": "❌", - "waiting": "⏸️", + "waiting": "⏸️", "qc_review": "🔍", } lines = ["| 状态 | 任务 | 角色 | ID |", "|------|------|------|----|"] for r in recs: diff --git a/pipeline_service/task_capability.py b/pipeline_service/task_capability.py index 5b4e52f..51c7ae5 100644 --- a/pipeline_service/task_capability.py +++ b/pipeline_service/task_capability.py @@ -19,7 +19,8 @@ logger = logging.getLogger("pipeline.task_capability") # 任务状态（SDLC 默认，状态机语义见 task skill） S_SUBMITTED = "submitted" # 待认领 S_RUNNING = "running" # 角色 agent 执行中 -S_REVIEW = "review" # 已提交待审核 +S_QC_REVIEW = "qc_review" # 已交付，待 QC 合规/质量检查 +S_REVIEW = "review" # QC 通过，待 PM 审核验收 S_APPROVED = "approved" # 审核通过 S_COMPLETED = "completed" # 全流程完成 S_WAITING = "waiting" # 挂起等回答 @@ -105,11 +106,23 @@ async def claim_task(tenant_id, role, agent_id, from_state=S_SUBMITTED, async def submit_task(task_id, tenant_id, who=None, agent_id=None): - """提交产出：running → review（清 claimed_by）。""" - return await _transition(task_id, tenant_id, S_RUNNING, S_REVIEW, 'submit', + """提交产出：running → qc_review（清 claimed_by，先过 QC 合规/质量门禁）。""" + return await _transition(task_id, tenant_id, S_RUNNING, S_QC_REVIEW, 'submit', who=who, agent_id=agent_id) +async def qc_approve_task(task_id, tenant_id, who=None, agent_id=None, comment=None): + """QC 通过：qc_review → review（进入 PM 审核验收）。""" + return await _transition(task_id, tenant_id, S_QC_REVIEW, S_REVIEW, 'qc_approve', + who=who, agent_id=agent_id, detail=comment) + + +async def qc_reject_task(task_id, tenant_id, who=None, agent_id=None, comment=None): + """QC 退回：qc_review → submitted（清 claimed_by，被退角色重新认领重做）。""" + return await _transition(task_id, tenant_id, S_QC_REVIEW, S_SUBMITTED, 'qc_reject', + who=who, agent_id=agent_id, detail=comment) + + async def approve_task(task_id, tenant_id, who=None, agent_id=None, comment=None): """审核通过：review → approved。""" return await _transition(task_id, tenant_id, S_REVIEW, S_APPROVED, 'approve',