From bf819f3b79af7daf2f8e7ff3788e2f684635b572 Mon Sep 17 00:00:00 2001 From: ymq Date: Sun, 16 Aug 2026 23:51:55 +0800 Subject: [PATCH] =?UTF-8?q?refactor(agent=5Floop):=20=E7=8A=B6=E6=80=81?= =?UTF-8?q?=E6=9C=BA=E8=BF=81=E7=A7=BB=E8=B5=B0=20task=20=E8=83=BD?= =?UTF-8?q?=E5=8A=9B=20+=20=E8=A7=92=E8=89=B2=E8=A7=84=E8=8C=83=E5=8C=96?= =?UTF-8?q?=20agent.{role}?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 角色规范: agent 角色 agent.{role}(_normalize_role 补前缀+ROLE_ALIASES/ROLE_CHAIN/SDL_ROLES 改规范名)， 人角色 {orgtype}.{role}(默认 owner.superuser) - role_agent_run/pm_review_run/handle_failed_task 硬编码状态 SQL 改 task_capability 工具 (submit/approve/reject/complete/mark_failed/retry，CAS+审计) - communication/questions/sdlc_ability 默认 handler 角色规范化(main_agent→agent.main_agent, customer→owner.superuser) - 通用助手路径(AgentExecutor)不动，保持 hermes 式自由对话 --- pipeline_service/agent_loop.py | 74 +++++++++++++++++-------------- pipeline_service/communication.py | 6 +-- pipeline_service/questions.py | 21 ++++----- pipeline_service/sdlc_ability.py | 34 +++++++------- 4 files changed, 71 insertions(+), 64 deletions(-) diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 5c93bfe..1416686 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -20,21 +20,21 @@ import logging logger = logging.getLogger("pipeline.agent_loop") 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', - 'pm': 'pm', 'project_manager': 'pm', '项目经理': 'pm', 'manager': 'pm', + 'developer': 'agent.develop', 'dev': 'agent.develop', 'development': 'agent.develop', 'coding': 'agent.develop', + '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', + 'operation': 'agent.ops', 'operations': 'agent.ops', 'maintenance': 'agent.ops', '运维': 'agent.ops', + 'pm': 'agent.pm', 'project_manager': 'agent.pm', '项目经理': 'agent.pm', 'manager': 'agent.pm', } ROLE_CHAIN = { - 'requirement': 'design', - 'design': 'develop', - 'develop': 'test', - 'test': 'deploy', - 'deploy': None, + 'agent.requirement': 'agent.design', + 'agent.design': 'agent.develop', + 'agent.develop': 'agent.test', + 'agent.test': 'agent.deploy', + 'agent.deploy': None, } TASK_REVIEW = 'review' @@ -50,8 +50,12 @@ _ALLOWED_WORKDIRS = [ def _normalize_role(role): + """角色规范化：别名映射 + 补 agent. 前缀（人角色 {orgtype}.{role} 保留原样）。""" r = (role or '').strip().lower() - return ROLE_ALIASES.get(r, r) + r = ROLE_ALIASES.get(r, r) + if r and '.' not in r: + r = f"agent.{r}" + return r async def _resolve_pipeline_id(project_id): @@ -455,7 +459,7 @@ async def _build_qna_section(sor, task_id, role, agent_id=None): if qna: lines.append("历史问答：") for item in qna: - src = "客户" if item.get('answer_source') == "customer" else "主agent" + src = "业主" if item.get('answer_source') == "owner.superuser" else "主agent" lines.append(f"问：{item.get('question','')}") lines.append(f"答({src})：{item.get('answer','')}") return "\n".join(lines) @@ -756,7 +760,8 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): resp = await llm_call_msgs_native(msgs, tools=tools_schema, model=model_name, temperature=0.4) except Exception as e: err_msg = f"{type(e).__name__}: {str(e)[:400]}" - await sor.sqlExe("UPDATE pipeline_tasks SET state='failed', last_error=${e}$ WHERE id=${tid}$", {"e": err_msg, "tid": task_id}) + from .task_capability import mark_failed + await mark_failed(task_id, project_id, who=role, agent_id=agent_id, error=err_msg) logger.error(f"role_agent_run llm failed: task={task_id} err={e}") return {"status": "failed", "task_id": task_id, "error": str(e)[:200]} @@ -813,7 +818,7 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): qid = await raise_problem("need_info", ask_question, role, from_agentid=agent_id, tenant_id=project_id, task_id=task_id, - first_handler_role="main_agent", + first_handler_role="agent.main_agent", context={"title": title}) return {"status": "need_info", "task_id": task_id, "question_id": qid, "question": ask_question} @@ -879,11 +884,10 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): logger.info(f"git_commit_push {repo_name}: rc={r.get('rc', -1)} {r.get('message', '')[:100]}") git_result = {"rc": 0, "message": "; ".join(git_results) if git_results else "无仓库可提交"} - await sor.sqlExe( - "UPDATE pipeline_tasks SET state=${st}$, claimed_by=NULL WHERE id=${tid}$", - {"st": TASK_REVIEW, "tid": task_id}) + from .task_capability import submit_task + ok, _ = await submit_task(task_id, project_id, who=role, agent_id=agent_id) logger.info(f"role_agent_run completed: task={task_id} role={role} " - f"deliverable={did} files={len(files_written)} git={git_result.get('rc',-1)}") + f"deliverable={did} files={len(files_written)} git={git_result.get('rc',-1)} submit={ok}") return {"status": "completed", "task_id": task_id, "deliverable_id": did, "files_written": len(files_written), "git_result": git_result} @@ -913,7 +917,8 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): 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}) + from .task_capability import reject_task + await reject_task(task_id, project_id, who="agent.pm", agent_id=agent_id, comment="没有交付件") return {"status": "rejected", "task_id": task_id, "reason": "没有交付件"} repo_state = await _get_repo_state(workspace_dir) @@ -949,7 +954,8 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): raw = await llm_call_msgs(msgs, model=model_name, temperature=0.3) except Exception as e: err_msg = f"{type(e).__name__}: {str(e)[:400]}" - await sor.sqlExe("UPDATE pipeline_tasks SET state='failed', last_error=${e}$ WHERE id=${tid}$", {"e": err_msg, "tid": task_id}) + from .task_capability import mark_failed + await mark_failed(task_id, project_id, who="agent.pm", agent_id=agent_id, error=err_msg) return {"status": "failed", "task_id": task_id, "error": str(e)[:200]} act = _parse_agent_action(raw) @@ -999,7 +1005,8 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): "AND (problem_type='review_reject' " " OR ((problem_type IS NULL OR problem_type='') AND from_role IN ('pm','cockpit')))", {"a": "角色已响应，PM审核通过", "role": task_role or "role_agent", "tid": task_id}) - await sor.sqlExe("UPDATE pipeline_tasks SET state=${st}$, claimed_by=NULL WHERE id=${tid}$", {"st": TASK_APPROVED, "tid": task_id}) + from .task_capability import approve_task + await approve_task(task_id, project_id, who="agent.pm", agent_id=agent_id, comment=comment) next_role = await _get_next_role(task_role, project_id) if next_role: next_tid, next_title = await _create_next_task(sor, project_id, task, next_role, comment) @@ -1012,12 +1019,13 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): rejection_q = decision.get("questions") or comment or "交付件不满足要求" # 审核退回：首处理方=被退角色（该角色任意 agent 重新认领响应）。 # suspend_task=False —— 任务回 submitted（重新认领）而非 waiting（挂起等回答）。 - await raise_problem("review_reject", rejection_q, "pm", + await raise_problem("review_reject", rejection_q, "agent.pm", tenant_id=project_id, task_id=task_id, first_handler_role=task_role, context={"pm_comment": comment, "deliverable_type": deliverable_type}, suspend_task=False) - await sor.sqlExe("UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL WHERE id=${tid}$", {"tid": task_id}) + from .task_capability import reject_task + await reject_task(task_id, project_id, who="agent.pm", agent_id=agent_id, comment=comment) return {"status": "rejected", "task_id": task_id, "comment": comment, "question": rejection_q} else: @@ -1027,7 +1035,8 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): "UPDATE pipeline_deliverables SET review_status='approved', review_comment=${cm}$ " "WHERE task_id=${tid}$ AND review_status='pending'", {"cm": comment, "tid": task_id}) - await sor.sqlExe("UPDATE pipeline_tasks SET state='completed', claimed_by=NULL WHERE id=${tid}$", {"tid": task_id}) + from .task_capability import complete_task + await complete_task(task_id, project_id, who="agent.pm", agent_id=agent_id) return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"} @@ -1089,22 +1098,19 @@ async def handle_failed_task(task_id: str, project_id: str) -> dict: decision = _classify_failure(last_error) if decision == "retry": - await sor.sqlExe( - "UPDATE pipeline_tasks SET retry_count=retry_count+1, state='submitted', " - "claimed_by=NULL, last_error=NULL, updated_at=NOW() " - "WHERE id=${tid}$ AND state='failed'", - {"tid": task_id}) - logger.info("failed task retry: task=%s role=%s attempt=%d", task_id, role, retry_count + 1) + from .task_capability import retry_task + ok, _ = await retry_task(task_id, project_id, who="agent.pm") + logger.info("failed task retry: task=%s role=%s attempt=%d ok=%s", task_id, role, retry_count + 1, ok) return {"status": "retry", "task_id": task_id, "attempt": retry_count + 1} # 报故障：建问题通知用户，任务置 waiting 等待人工介入 from .communication import raise_problem - reporter = "cockpit" if role == "pm" else "pm" + reporter = "agent.main_agent" if role == "agent.pm" else "agent.pm" qid = await raise_problem( "fault_report", f"任务「{title}」已失败 {retry_count} 次，自动重试无法解决，请人工介入。失败原因：{last_error or '未知'}", reporter, tenant_id=project_id, task_id=task_id, - first_handler_role="main_agent", + first_handler_role="agent.main_agent", context={"fault": True, "last_error": last_error, "retry_count": retry_count}) logger.info("failed task fault: task=%s role=%s qid=%s", task_id, role, qid) return {"status": "fault", "task_id": task_id, "question_id": qid} diff --git a/pipeline_service/communication.py b/pipeline_service/communication.py index e07e4bf..7b7e4e8 100644 --- a/pipeline_service/communication.py +++ b/pipeline_service/communication.py @@ -56,7 +56,7 @@ async def raise_problem(problem_type, question, from_role, from_agentid="", """ db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: - hr = first_handler_role or "main_agent" + hr = first_handler_role or "agent.main_agent" ha = first_handler_agentid or "" qid = getID() ctx_json = json.dumps(context, ensure_ascii=False, default=str) if context else None @@ -83,7 +83,7 @@ async def raise_problem(problem_type, question, from_role, from_agentid="", return qid -async def resolve_problem(question_id, answer, answered_by="", answer_source="main_agent", +async def resolve_problem(question_id, answer, answered_by="", answer_source="agent.main_agent", resume_task=True): """当前处理方解决了问题 → 停止冒泡(status=answered)，任务恢复 submitted。""" db, dbname = _get_db() @@ -98,7 +98,7 @@ async def resolve_problem(question_id, answer, answered_by="", answer_source="ma 'id': question_id, 'answer': answer or '', 'answered_by': answered_by or '', - 'answer_source': answer_source or 'main_agent', + 'answer_source': answer_source or 'agent.main_agent', 'status': S_ANSWERED, }) diff --git a/pipeline_service/questions.py b/pipeline_service/questions.py index 7e9aa18..78df0eb 100644 --- a/pipeline_service/questions.py +++ b/pipeline_service/questions.py @@ -38,29 +38,30 @@ async def agent_ask(tenant_id: str, task_id: str, from_role: str, question: str, 新代码请直接调 communication.raise_problem 显式指定首处理方。 """ - if from_role == "pm": - ptype, hr = "review_reject", "develop" - elif from_role == "cockpit": - ptype, hr = "fault_report", "main_agent" + r = (from_role or "").lower() + if "pm" in r: + ptype, hr = "review_reject", "agent.develop" + elif "main_agent" in r or "cockpit" in r: + ptype, hr = "fault_report", "agent.main_agent" else: - ptype, hr = "need_info", "main_agent" + ptype, hr = "need_info", "agent.main_agent" return await raise_problem(ptype, question, from_role, "", tenant_id=tenant_id, task_id=task_id, context=context, first_handler_role=hr) async def answer_question(question_id: str, answer: str, answered_by: str = "", - answer_source: str = "main_agent", resume: bool = True): + answer_source: str = "agent.main_agent", resume: bool = True): """回答问题（解决 → 停止冒泡，任务恢复 submitted）。""" return await resolve_problem(question_id, answer, answered_by, answer_source, resume_task=resume) -async def forward_question(question_id: str, next_handler_role: str = "customer", +async def forward_question(question_id: str, next_handler_role: str = "owner.superuser", next_handler_agentid: str = ""): - """答不了 → 沿冒泡路径转给下一个处理方（默认客户）。""" + """答不了 → 沿冒泡路径转给下一个处理方（默认业主/客户）。""" return await escalate_problem(question_id, next_handler_role, next_handler_agentid, - by_role="main_agent") + by_role="agent.main_agent") async def get_question(question_id: str): @@ -87,7 +88,7 @@ async def get_question(question_id: str): async def list_questions(tenant_id: str = None, task_id: str = None, status: str = None, limit: int = 50) -> list: """客户视角：列出当前该「客户(人)」处理的问题。""" - rows = await list_problems_for("customer", tenant_id=tenant_id, + rows = await list_problems_for("owner.superuser", tenant_id=tenant_id, task_id=task_id, limit=limit) if status: rows = [r for r in rows if (r.get('status') or '') == status] diff --git a/pipeline_service/sdlc_ability.py b/pipeline_service/sdlc_ability.py index 33ee383..5027ff9 100644 --- a/pipeline_service/sdlc_ability.py +++ b/pipeline_service/sdlc_ability.py @@ -142,7 +142,7 @@ SDL_PROMPT = """你是「开发产线」的驾驶舱 agent，负责软件项目 SDL_ROLES = [ RoleSpec( - name="requirement", + name="agent.requirement", description="需求分析师", aliases=["requirements", "requirement_analysis"], system_prompt="""你是需求分析师。按 SDLC 仓库标准产出文档。 @@ -153,10 +153,10 @@ SDL_ROLES = [ - modules/<模块名>.md: 模块功能、仓库URL(先填写规划地址)、技术栈、依赖 - 每个应用至少关联一个模块 内容: 项目概述、用户角色及权限、功能列表(每个功能:输入/处理/输出/验收标准)、非功能需求、业务流程。""", - next_role="design", + next_role="agent.design", ), RoleSpec( - name="design", + name="agent.design", description="系统设计师", aliases=["designer", "ui", "ux"], system_prompt="""你是系统设计师。按 SDLC 仓库标准产出文档。 @@ -167,10 +167,10 @@ SDL_ROLES = [ - api-design.md: 接口列表(method/path/request/response) - ui-design.md: 页面结构、组件树(如适用) 产出后用 result 输出文档，files 列出所有文件路径。""", - next_role="develop", + next_role="agent.develop", ), RoleSpec( - name="develop", + name="agent.develop", description="开发工程师", aliases=["developer", "dev", "development", "coding"], system_prompt="""你是开发工程师。源码写入模块独立仓库，不在项目仓库。 @@ -187,10 +187,10 @@ SDL_ROLES = [ 8. write_file 写 docs/02-develop/dev-notes.md 记录开发内容 9. deliver 交付，files 列出所有产出文件路径 PM审核: git_status 检查模块仓库有提交记录。""", - next_role="test", + next_role="agent.test", ), RoleSpec( - name="test", + name="agent.test", description="测试工程师", aliases=["testing", "qa", "tester"], system_prompt="""你是测试工程师。按 SDLC 仓库标准产出文档。 @@ -200,10 +200,10 @@ PM审核: git_status 检查模块仓库有提交记录。""", - test-cases.md: 用例清单(编号/前置条件/步骤/预期结果) - test-report.md: 执行结果、Bug清单、覆盖率 测试脚本放到 files。""", - next_role="deploy", + next_role="agent.deploy", ), RoleSpec( - name="deploy", + name="agent.deploy", description="部署运维工程师", aliases=["deployment", "release"], system_prompt="""你是部署运维工程师。按 SDLC 仓库标准产出文档和配置。 @@ -227,7 +227,7 @@ async def _h_create_task(sor, p, ctx): from appPublic.uniqueID import getID title = p.get("title", "").strip() - role = p.get("role", "develop") + role = p.get("role", "agent.develop") desc = p.get("description", "") if not title: return "需要任务标题" @@ -315,7 +315,7 @@ async def _h_start_agents(sor, p, ctx): results = [] for r in recs: tid = getattr(r, "id", "") - role = getattr(r, "role", "develop") + role = getattr(r, "role", "agent.develop") try: from pipeline_service.agent_loop import role_agent_run result = await role_agent_run(pid, role) @@ -345,7 +345,7 @@ async def _h_diagnose_project(sor, p, ctx): try: from .communication import list_problems_for # 待回答问题 = 当前该「主 agent」处理的问题（不混入 PM 退回给角色 agent 的） - mine = await list_problems_for("main_agent", tenant_id=pid, limit=200) + mine = await list_problems_for("agent.main_agent", tenant_id=pid, limit=200) qc = len(mine) except Exception: pass @@ -432,7 +432,7 @@ async def _h_list_questions(sor, p, ctx): return "请先切换到项目" from .communication import list_problems_for # 只列「当前该主 agent 处理」的问题（不混入 PM 退回给角色 agent 的退回意见） - recs = await list_problems_for("main_agent", tenant_id=pid, limit=10) + recs = await list_problems_for("agent.main_agent", tenant_id=pid, limit=10) if not recs: return "没有待回答问题" lines = [] @@ -459,10 +459,10 @@ async def _h_escalate_question(sor, p, ctx): if recs: full_qid = getattr(recs[0], "id", qid) # 主 agent 答不了 → 沿冒泡路径转给客户（下一个处理方由 team-communication skill 决定） - r = await escalate_problem(full_qid, "customer", by_role="main_agent") + r = await escalate_problem(full_qid, "owner.superuser", by_role="agent.main_agent") if r is None: return f"问题不存在: {qid}" - return f"OK: 已沿冒泡路径转给 {r.get('current_handler_role', 'customer')}" + return f"OK: 已沿冒泡路径转给 {r.get('current_handler_role', 'owner.superuser')}" async def _h_answer_question(sor, p, ctx): @@ -482,8 +482,8 @@ async def _h_answer_question(sor, p, ctx): task_id = getattr(recs[0], "task_id", "") or "" await sor.sqlExe( - "UPDATE pipeline_agent_questions SET answer=${a}$, answer_source='main_agent', " - "answered_by='main_agent', status='answered' WHERE id=${qid}$", + "UPDATE pipeline_agent_questions SET answer=${a}$, answer_source='agent.main_agent', " + "answered_by='agent.main_agent', status='answered' WHERE id=${qid}$", {"a": answer, "qid": full_qid}) resumed = False