feat(flow): 批2多流程模板机制+三机制修复——pipeline_flow_plans加base_flow_key列;flow_plan_capability:detect_flow_template(LLM判流,置信度+判据,失败不静默降级发人工)/propose_flow_plan支持flow_key+确认待办渲染判流结论/get_plan_flow_key供流转分支/_latest_plan_rec复用;agent_loop:_skip_trimmed_roles按base_flow_key解析阶段+human_task_qc检查对象不存在改cancel+fault冒泡(不再qc_reject计轮死循环,首医实测)+PM approve答结并入qc_reject(幻影角标,pbls实测);project_capability:pause_project不再删pending待办(暂停=可逆停摆,删待办致resume后QC死循环)

This commit is contained in:
ymq 2026-09-15 15:01:27 +08:00
parent d204ff4a9f
commit 5fe5c238ec
4 changed files with 215 additions and 35 deletions

View File

@ -31,6 +31,14 @@
"length": 64,
"nullable": "no"
},
{
"name": "base_flow_key",
"title": "基准流程模板key(多流程机制;空=产线单流程flow_stages)",
"type": "str",
"length": 64,
"nullable": "yes",
"default": ""
},
{
"name": "version",
"title": "版本(驳回重提+1)",

View File

@ -1486,18 +1486,25 @@ async def _skip_trimmed_roles(pipeline_id, project_id, next_role):
"""沿 next_role 链跳过被裁阶段的角色,返回第一个保留角色(或 "")。
角色 阶段归属能力包 flow_stages roles 含该角色即属该阶段
多流程机制2026-09-14已确认计划带 base_flow_key 时按该流程模板的阶段解析
阶段不在任何 flow_stage 声明里的角色 = 不受裁剪影响照常保留
防环最多跳 20 角色链本身有限异常声明不至于死循环
"""
from pipeline_core import get_role_spec, get_flow_stages
try:
from .flow_plan_capability import plan_stage_enabled
from .flow_plan_capability import plan_stage_enabled, get_active_plan
enabled = await plan_stage_enabled(project_id)
except Exception:
return next_role
if enabled is None:
return next_role # 无确认计划 = 标准全流程
stages = get_flow_stages(pipeline_id) or []
flow_key = ""
try:
plan = await get_active_plan(project_id)
flow_key = (plan.get("base_flow_key") or "").strip()
except Exception:
pass
stages = get_flow_stages(pipeline_id, flow_key=flow_key) or []
role_stage = {}
for s in stages:
for r in (s.roles or []):
@ -3322,11 +3329,14 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
await sor.sqlExe("UPDATE pipeline_deliverables SET review_status='approved', review_comment=${cm}$ WHERE task_id=${tid}$", {"cm": comment, "tid": task_id})
# 审核通过:该任务 review_reject 类的 pending 退回意见已由角色 agent 响应 → 解决,避免永久堆积。
# 只答结 review_reject归属角色 agent 的),不碰 need_info归属主 agent等其它 pending。
# 2026-09-15 补 qc_rejectQC 退回意见与 review_reject 语义对称——PM 审核通过
# 即代表 QC 退回意见已被响应QC 已通过才会进入 PM 审核),不并入则 qc_reject
# 问题永久 pending 挂幻影角标(首处理方角色 agent 无后续任务去答它pbls 实测)。
await sor.sqlExe(
"UPDATE pipeline_agent_questions SET status='answered', answer=${a}$, "
"answer_source='role_agent', answered_by=${role}$ "
"WHERE task_id=${tid}$ AND status='pending' "
"AND (problem_type='review_reject' "
"AND (problem_type IN ('review_reject','qc_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})
from .task_capability import approve_task
@ -3512,9 +3522,24 @@ async def qc_review_run(project_id, agent_id=None, model_name=None):
{"hid": human_task_id})
await sor.sqlExe("COMMIT", {})
if not ht_recs:
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": "人类任务不存在"}
# 2026-09-15 修复:检查对象不存在是「对象已被删除/归档」的系统性异常,
# 不是 QC 可退回重来的质量问题——qc_reject 会计轮直到上限转 failed
# 每轮都是对着不存在的对象空转(首医项目死循环实测)。
# 正解:取消本 QC 任务 + fault 冒泡一次性交人工,不进重试循环。
from .task_capability import cancel_task
from .communication import raise_problem
await cancel_task(task_id, project_id, who="agent.qc", agent_id=agent_id,
comment="人类任务不存在对象已被删除QC 无法检查")
await raise_problem(
"fault_report",
f"人类任务QC「{title}」的检查对象(人类任务 {human_task_id})已不存在,"
"无法继续 QC。请人工确认该事项是否已线下处理完毕已处理则答结本问题即可。",
"agent.qc", tenant_id=project_id, task_id=task_id,
first_handler_role="agent.main_agent",
suspend_task=False, # 任务已 cancelled终态不得再置回 waiting 复活
context={"fault": True, "reason": "human_task_missing",
"human_task_id": human_task_id})
return {"status": "fault", "task_id": task_id, "reason": "人类任务不存在已取消QC任务并冒泡人工"}
ht_title = getattr(ht_recs[0], 'title', '') or ''
ht_desc = getattr(ht_recs[0], 'description', '') or ''
ht_result = getattr(ht_recs[0], 'result_data', '') or ''

View File

@ -85,10 +85,13 @@ def _parse_trim_keys(raw):
return [x.strip() for x in s.replace("", ",").split(",") if x.strip()]
def _load_stages(pipeline_id):
"""从能力包取产线声明的阶段,返回 (stage_map, ordered_list)。"""
def _load_stages(pipeline_id, flow_key=""):
"""从能力包取产线声明的阶段,返回 (stage_map, ordered_list)。
flow_key 非空时取多流程模板FlowTemplate对应阶段 = 单流程 flow_stages
"""
from pipeline_core import get_flow_stages
stages = get_flow_stages(pipeline_id) or []
stages = get_flow_stages(pipeline_id, flow_key=flow_key or "") or []
return {s.key: s for s in stages}, list(stages)
@ -107,16 +110,99 @@ async def _resolve_pipeline(sor, project_id, pipeline_id=""):
# ══════════════════════ 展示预制流程 ══════════════════════
async def get_flow_template(project_id="", pipeline_id=""):
"""返回 (pid, [stage_dict,...]):产线预制流程阶段(含可裁性),供 show_flow_template 渲染。"""
async def get_flow_template(project_id="", pipeline_id="", flow_key=""):
"""返回 (pid, [stage_dict,...]):产线预制流程阶段(含可裁性),供 show_flow_template 渲染。
flow_key 空时项目已有确认/待确认计划则取计划的 base_flow_key展示实际将走的流程
多流程产线无计划时展示全部模板概览由调用方处理
"""
db, dbname = _get_db()
async with db.sqlorContext(dbname) as sor:
pid, _proj = await _resolve_pipeline(sor, project_id, pipeline_id)
fk = (flow_key or "").strip()
if not fk and project_id:
plan = await _latest_plan_rec(sor, project_id)
fk = (plan.get("base_flow_key") or "").strip()
from pipeline_core import flow_stage_to_dict
_map, stages = _load_stages(pid)
_map, stages = _load_stages(pid, fk)
return pid, [flow_stage_to_dict(s) for s in stages]
async def list_flow_templates(pipeline_id=""):
"""列出产线的多流程模板 [(key, name, description, n_stages)];单流程产线返回 []。"""
from pipeline_core import get_flow_templates
return [(t.key, t.name or t.key, t.description or "", len(t.stages or []))
for t in get_flow_templates(pipeline_id)]
# ══════════════════════ 文件自动判流多流程机制2026-09-14 批2 ══════════════════════
async def detect_flow_template(pipeline_id, file_name="", file_text="",
project_id="", org_id=""):
"""LLM 按各模板 detect_criteria 给导入文件判流(语义判断归 LLM2026-09-11 用户铁律)。
返回 (flow_key, detect_note)
- 单流程产线模板 <2 ("", "")调用方走现状不判流不确认
- 判流成功 (模板key, 识别结论文本置信度+判据摘录进确认待办给用户核对)
- LLM 失败/无法解析 ("", 错误说明)调用方**不得**默认选流程硬跑
应发人工介入待办判流是流程选择的语义判断禁词匹配兜底禁静默降级
错判由 flow_plan_confirm 确认门兜住用户在待办里看到识别结论驳回附意见重提
"""
from pipeline_core import get_flow_templates
tpls = get_flow_templates(pipeline_id)
if len(tpls) < 2:
return "", ""
snippet = (file_text or "").strip()[:6000]
if not snippet and not file_name:
return "", "判流失败:文件内容为空"
tpl_lines = []
for t in tpls:
tpl_lines.append("- key=%s 名称=%s\n 适用判据:%s" % (
t.key, t.name or t.key, (t.detect_criteria or t.description or "(未写判据)")))
prompt = (
"你是产线流程路由器。根据输入文件的特征,判断它应该走哪条预制流程。\n\n"
"【候选流程】\n%s\n\n"
"【文件名】%s\n\n【文件正文(截断)】\n%s\n\n"
"严格只输出一个 JSON 对象:\n"
"{\"flow_key\": \"候选之一的key\", \"confidence\": \"high|medium|low\",\n"
" \"evidence\": \"判据摘录文件里哪些特征原文引用≤3处支持这个判断\"}\n"
"无法归入任何候选流程时 flow_key 输出空字符串并在 evidence 说明原因。"
% ("\n".join(tpl_lines), file_name or "(无)", snippet or "(无正文)")
)
try:
from pipeline_service.llm_bridge import llm_call
raw = await llm_call(prompt, purpose="utility", org_id=org_id or "",
project_id=project_id or "", timeout=45)
except Exception as e:
return "", "判流 LLM 调用失败:%s" % str(e)[:200]
txt = (raw or "").strip()
if txt.startswith("```"):
parts = txt.split("\n", 1)
txt = parts[1] if len(parts) > 1 else txt
if "```" in txt:
txt = txt.rsplit("```", 1)[0]
txt = txt.strip()
try:
if not txt.startswith("{") and "{" in txt:
txt = txt[txt.find("{"):txt.rfind("}") + 1]
data = json.loads(txt)
except (json.JSONDecodeError, ValueError):
return "", "判流结果无法解析:%s" % (raw or "")[:200]
fk = str(data.get("flow_key") or "").strip()
conf = str(data.get("confidence") or "").strip()
ev = str(data.get("evidence") or "").strip()
valid = {t.key for t in tpls}
if fk and fk not in valid:
return "", "判流返回未知流程 key%s(候选:%s" % (fk, "".join(sorted(valid)))
if not fk:
return "", "判流无法归类:%s" % (ev or "LLM 未说明原因)")
note = "识别为「%s」流程(置信度 %s)。\n判据:%s" % (
fk, conf or "?", ev or "(未提供判据摘录)")
if conf == "low":
note += "\n⚠️ 识别置信度低,请重点核对——不对就驳回并说明应走哪条流程。"
return fk, note
def render_template_md(stages):
"""把阶段列表渲染成 markdown 表(展示给用户/待办)。"""
if not stages:
@ -134,11 +220,17 @@ def render_template_md(stages):
# ══════════════════════ 提案(裁剪草案 + 确认待办) ══════════════════════
async def propose_flow_plan(project_id, trim_keys=None, user_requirements="",
propose_note="", pipeline_id="", who=None, agent_id=None):
propose_note="", pipeline_id="", flow_key="",
detect_note="", who=None, agent_id=None):
"""按用户要求生成裁剪草案,落库 pending_confirm 并发确认待办。
flow_key多流程机制2026-09-14 批2指定基准流程模板 key
- = 单流程产线 flow_stages 声明或多流程产线沿用已有计划的模板
- 非空 = 按该模板的阶段链出草案base_flow_key 落库流转机制按它走分支
detect_note判流依据说明LLM 自动判流时写识别结论+判据摘录进确认待办给用户核对
代码硬校验任一不过即拒绝返回可行动原因绝不静默降级
1. 项目/产线存在且产线声明了 flow_stages
1. 项目/产线存在且产线声明了 flow_stages flow_key 对应模板
2. 裁剪时机项目流程已开跑有角色任务进入 started 拒绝
3. 已有 confirmed 计划 拒绝启动时裁剪一次确认后不可改
4. trim_keys 都是已知阶段
@ -149,6 +241,7 @@ async def propose_flow_plan(project_id, trim_keys=None, user_requirements="",
if not project_id:
return False, "缺少 project_id请先切换到项目"
trim_keys = _parse_trim_keys(trim_keys)
flow_key = (flow_key or "").strip()
db, dbname = _get_db()
async with db.sqlorContext(dbname) as sor:
pid, proj = await _resolve_pipeline(sor, project_id, pipeline_id)
@ -156,7 +249,18 @@ async def propose_flow_plan(project_id, trim_keys=None, user_requirements="",
return False, "项目不存在: %s" % project_id
if not pid:
return False, "项目未关联产线,无法裁剪流程"
stage_map, stages = _load_stages(pid)
# flow_key 校验:多流程产线必须是已声明的模板 key
if flow_key:
from pipeline_core import get_flow_template as _get_ftpl
if not _get_ftpl(pid, flow_key):
tpls = await list_flow_templates(pid)
avail = "".join(k for k, _n, _d, _c in tpls) or "(该产线未声明多流程模板)"
return False, ("未知流程模板 key%s。可用流程:%s" % (flow_key, avail))
else:
# 未指定:已有计划则沿用其 base_flow_key驳回重提场景
old_plan = await _latest_plan_rec(sor, project_id)
flow_key = (old_plan.get("base_flow_key") or "").strip()
stage_map, stages = _load_stages(pid, flow_key)
if not stages:
return False, ("产线「%s」未声明可裁剪流程flow_stages 为空),"
"不支持流程裁剪" % pid)
@ -250,6 +354,7 @@ async def propose_flow_plan(project_id, trim_keys=None, user_requirements="",
plan_id = getID()
await sor.C("pipeline_flow_plans", {
"id": plan_id, "project_id": project_id, "pipeline_id": pid,
"base_flow_key": flow_key,
"version": version, "status": S_PENDING,
"stages": json.dumps(stages_snap, ensure_ascii=False),
"user_requirements": (user_requirements or "")[:8000],
@ -260,7 +365,9 @@ async def propose_flow_plan(project_id, trim_keys=None, user_requirements="",
# 确认待办(平台待办中枢;专用确认端点 flow_plan_confirm.dspy 激活)
desc = _render_confirm_md(pname, stages_snap, trim_keys, warns,
user_requirements, propose_note, version)
user_requirements, propose_note, version,
flow_key=flow_key, pipeline_id=pid,
detect_note=detect_note)
ok_ht, ht = await create_human_task(
project_id,
"流程裁剪确认:%sv%d" % (pname, version),
@ -273,25 +380,42 @@ async def propose_flow_plan(project_id, trim_keys=None, user_requirements="",
await sor.sqlExe("COMMIT", {})
await record_audit(project_id, "pipeline_flow_plans", plan_id, "propose",
to_state=S_PENDING, who=who, agent_id=agent_id,
detail="version=%d trim=%s" % (version, ",".join(trim_keys) or "(无裁剪)"),
detail="version=%d flow_key=%s trim=%s" % (
version, flow_key or "(单流程)",
",".join(trim_keys) or "(无裁剪)"),
sor=sor)
await sor.sqlExe("COMMIT", {})
logger.info("flow_plan propose: project=%s plan=%s v=%d trim=%s",
project_id, plan_id, version, trim_keys)
logger.info("flow_plan propose: project=%s plan=%s v=%d flow_key=%s trim=%s",
project_id, plan_id, version, flow_key, trim_keys)
kept = [s["label"] or s["key"] for s in stages_snap if s["enabled"]]
trimmed = [s["label"] or s["key"] for s in stages_snap if not s["enabled"]]
msg = ("已生成流程裁剪草案plan %sv%d)并发确认待办。\n保留阶段:%s\n裁剪阶段:%s"
% (plan_id[:8], version, "".join(kept) or "(无)",
"".join(trimmed) or "(无)"))
if flow_key:
msg = "基准流程:%s\n%s" % (flow_key, msg)
if warns:
msg += "\n⚠️ 风险提示(请在待办里向用户说明):\n" + "\n".join("- " + w for w in warns)
msg += "\n用户需在「流程裁剪确认」待办里点击确认后流程才生效agent 不能代替确认。"
return True, msg
def _render_confirm_md(pname, stages_snap, trim_keys, warns, user_req, note, version):
def _render_confirm_md(pname, stages_snap, trim_keys, warns, user_req, note, version,
flow_key="", pipeline_id="", detect_note=""):
"""确认待办正文markdowntodo_detail 直接渲染)。"""
lines = ["## 项目「%s」流程裁剪方案v%d" % (pname, version), ""]
if flow_key:
# 多流程产线:展示判定的流程模板名 + 判流依据(用户核对识别结论)
try:
from pipeline_core import get_flow_template as _gft
tpl = _gft(pipeline_id, flow_key) if pipeline_id else None
tname = (tpl.name if tpl else "") or flow_key
except Exception:
tname = flow_key
lines += ["**将走的流程:** %s`%s`" % (tname, flow_key), ""]
if detect_note:
lines += ["### 文件识别结论(自动判流)", "", detect_note.strip(), "",
"> 识别不对?点「驳回」并说明应走哪条流程,助手会修订重提。", ""]
if user_req:
lines += ["**你的裁剪要求:**", "", user_req.strip(), ""]
if note:
@ -508,11 +632,24 @@ async def get_latest_plan(project_id):
return {}
db, dbname = _get_db()
async with db.sqlorContext(dbname) as sor:
recs = await sor.sqlExe(
"SELECT * FROM pipeline_flow_plans WHERE project_id=${p}$ "
"ORDER BY version DESC, created_at DESC LIMIT 1", {"p": project_id})
await sor.sqlExe("COMMIT", {})
return _rec(recs[0]) if recs else {}
return await _latest_plan_rec(sor, project_id)
async def _latest_plan_rec(sor, project_id):
"""取项目最新一条计划(已开 sor 上下文版,内部复用)。"""
if not project_id:
return {}
recs = await sor.sqlExe(
"SELECT * FROM pipeline_flow_plans WHERE project_id=${p}$ "
"ORDER BY version DESC, created_at DESC LIMIT 1", {"p": project_id})
await sor.sqlExe("COMMIT", {})
return _rec(recs[0]) if recs else {}
async def get_plan_flow_key(project_id):
"""项目已确认计划的 base_flow_key'' = 无计划/单流程)。流转机制按它选流程分支。"""
plan = await get_active_plan(project_id)
return (plan.get("base_flow_key") or "").strip() if plan else ""
# ══════════════════════ PM 复杂度拆解create_sub_tasks通用角色工具 ══════════════════════
@ -611,17 +748,18 @@ if ToolDefinition is not None:
FLOW_PLAN_TOOLS = [
ToolDefinition(
name="show_flow_template",
description="查看当前项目所属产线的预制流程(各阶段作用 + 可裁性:不可裁/可裁需警告/可自由裁剪)。用户提出裁剪流程前先用它了解可裁项",
description="查看当前项目所属产线的预制流程(各阶段作用 + 可裁性:不可裁/可裁需警告/可自由裁剪)。多流程产线会列出全部可选流程模板。用户提出裁剪流程前先用它了解可裁项",
parameters={}, required=[], category="project"),
ToolDefinition(
name="propose_flow_plan",
description=("按用户要求裁剪项目流程,生成裁剪草案并发「流程裁剪确认」待办给用户。"
description=("按用户要求裁剪项目流程(多流程产线可指定走哪条流程模板),生成草案并发「流程裁剪确认」待办给用户。"
"裁剪合法性由系统硬校验(不可裁阶段/依赖断裂/已开跑会被拒绝并说明原因)。"
"草案需用户在待办里确认才生效——你没有确认权,禁止声称已确认。"
"trim_keys 传要裁掉的阶段 key 数组(来自 show_flow_template"),
parameters={"trim_keys": "要裁剪的阶段key数组(如[\"analysis_cost_benefit\"]),空数组=不裁剪仅走确认",
"user_requirements": "用户裁剪要求原文(必填,写进确认待办)",
"propose_note": "你对裁剪方案的说明/风险提示(可选)"},
"propose_note": "你对裁剪方案的说明/风险提示(可选)",
"flow_key": "多流程产线指定流程模板key(可选,来自show_flow_template;单流程产线留空)"},
required=["user_requirements"], category="project"),
ToolDefinition(
name="get_flow_plan",
@ -633,13 +771,19 @@ if ToolDefinition is not None:
async def _h_show_flow_template(sor, params, ctx):
pid = (ctx.get("pipeline_id") or "").strip()
project_id = ctx.get("project_id") or ""
tpls = await list_flow_templates(pid)
head = ""
if len(tpls) >= 2:
head = ("本产线是多流程产线可选流程模板propose_flow_plan 用 flow_key 指定):\n"
+ "\n".join("- %s%s%s%d 阶段" % (k, n, d[:80], c)
for k, n, d, c in tpls) + "\n\n")
pid2, stages = await get_flow_template(project_id=project_id, pipeline_id=pid)
if not stages:
return ("当前产线(%s)未声明可裁剪流程,按标准全流程执行,无需裁剪。"
% (pid2 or "未知"))
return ("产线「%s」预制流程(共 %d 阶段):\n%s\n\n"
"用户要裁剪时,把要裁掉的阶段 key 传给 propose_flow_plan。"
% (pid2 or "?", len(stages), render_template_md(stages)))
return head + ("当前产线(%s)未声明可裁剪流程,按标准全流程执行,无需裁剪。"
% (pid2 or "未知"))
return head + ("产线「%s」预制流程(共 %d 阶段):\n%s\n\n"
"用户要裁剪时,把要裁掉的阶段 key 传给 propose_flow_plan。"
% (pid2 or "?", len(stages), render_template_md(stages)))
async def _h_propose_flow_plan(sor, params, ctx):
@ -652,6 +796,7 @@ async def _h_propose_flow_plan(sor, params, ctx):
user_requirements=params.get("user_requirements", "") or "",
propose_note=params.get("propose_note", "") or "",
pipeline_id=ctx.get("pipeline_id", "") or "",
flow_key=params.get("flow_key", "") or "",
who="agent.main_agent",
agent_id=ctx.get("user_id", "") or "")
return ("OK: " if ok else "ERROR: ") + msg

View File

@ -307,8 +307,10 @@ async def pause_project(project_id, who=None, agent_id=None):
from_state=cur, to_state=S_PAUSED,
who=_normalize_role(who), agent_id=agent_id,
detail=f"prev_state={cur}", sor=sor)
# 停止推进后清空项目全部待办2026-09-01项目不再推进遗留待办无意义且占角标
await _clear_project_todos(sor, project_id)
# 2026-09-15 修复:暂停不再删除待办。暂停语义是「可逆停摆」,删 pending 人类
# 任务/冒泡问题是破坏性动作——实测首医项目pause 删掉待办后 resume
# human_task_qc 的检查对象消失 → qc_reject 计轮死循环 → 项目再次 failed 暂停。
# 待办清理只保留给 archive/delete项目终态遗留待办确属无意义
return True, S_PAUSED