yumoqing/pipeline-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 问题回路修复——(1)角色agent注入PM退回意见(pending问题)而非只查answered (2)review_complete同步交付件approved (3)PM通过后标记pending问题answered避免堆积

Browse Source
This commit is contained in:
ymq 2026-08-16 20:47:36 +08:00
parent ae7ce87ee2
commit 1da53a0edd
1 changed files with 31 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
pipeline_service/agent_loop.py
View File

@ -441,14 +441,26 @@ async def _get_workspace_dir(sor, project_id):
async def _build_qna_section(sor, task_id):
from .questions import get_task_qna
qna = await get_task_qna(task_id)
if not qna:
qna = await get_task_qna(task_id) # 已回答问答（answered）
# 关键：额外查 pending 问题（PM 退回意见 / 待回答），角色 agent 重新认领时必须先响应这些再产出。
# 否则 PM review_reject 写的 pending 问题不会注入，角色反复产出同类不合格交付物，问题越堆越多。
pending = await sor.sqlExe(
"SELECT question, from_role FROM pipeline_agent_questions "
"WHERE task_id=${tid}$ AND status='pending' ORDER BY created_at ASC",
{"tid": task_id})
if not qna and not pending:
return ""
lines = ["历史问答："]
for item in qna:
src = "客户" if item.get('answer_source') == "customer" else "主agent"
lines.append(f"问：{item.get('question','')}")
lines.append(f"答({src})：{item.get('answer','')}")
lines = []
if pending:
lines.append("⚠️ 待处理问题（审核/PM 退回意见，你必须先逐条响应这些再产出交付件）：")
for p in pending:
lines.append(f"- [{getattr(p, 'from_role', '') or '审核'}] {getattr(p, 'question', '')}")
if qna:
lines.append("历史问答：")
for item in qna:
src = "客户" if item.get('answer_source') == "customer" else "主agent"
lines.append(f"问：{item.get('question','')}")
lines.append(f"答({src})：{item.get('answer','')}")
return "\n".join(lines)
@ -977,6 +989,12 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
"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})
# 审核通过：该任务的 pending 问题（PM 退回意见）已由角色 agent 响应，标记 answered，避免问题永久堆积
await sor.sqlExe(
"UPDATE pipeline_agent_questions SET status='answered', answer=${a}$, "
"answer_source='main_agent', answered_by='pm' "
"WHERE task_id=${tid}$ AND status='pending'",
{"a": comment or "已响应并审核通过", "tid": task_id})
await sor.sqlExe("UPDATE pipeline_tasks SET state=${st}$, claimed_by=NULL WHERE id=${tid}$", {"st": TASK_APPROVED, "tid": task_id})
next_role = await _get_next_role(task_role, project_id)
if next_role:
@ -993,6 +1011,12 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
return {"status": "rejected", "task_id": task_id, "comment": comment, "question": rejection_q}
else:
# review_complete（项目完成/无下一阶段）：同步把交付件标记 approved，
# 避免任务 state=completed 但交付件 review_status 仍 pending 的状态不一致。
await sor.sqlExe(
"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})
return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"}