From 6707cabedcda48b5182eac94b986df6eb42bdf85 Mon Sep 17 00:00:00 2001 From: ymq Date: Sun, 16 Aug 2026 22:08:00 +0800 Subject: [PATCH] =?UTF-8?q?refactor(team-communication):=20=E9=97=AE?= =?UTF-8?q?=E9=A2=98=E5=86=92=E6=B3=A1=E6=94=B9=E4=B8=BA=E6=98=BE=E5=BC=8F?= =?UTF-8?q?=E8=B7=AF=E7=94=B1=EF=BC=8C=E8=A7=84=E8=8C=83=E7=A7=BB=E4=BA=A4?= =?UTF-8?q?=20skill?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 规范(团队角色/问题类型/冒泡路径)移交 team-communication skill，由 LLM 读 skill 决策； 系统只固化通用问题工具，不读配置表、不存路径快照： - communication.py: raise/resolve/escalate/list_for 显式路由， escalate 显式传 next_handler；current_handler_role/agentid 两列确定性过滤 - questions.py: 向后兼容层适配 - agent_loop.py: need_info/review_reject/fault_report 显式 first_handler_role - sdlc_ability.py: escalate_question 显式转 customer - models: 去 escalation_path/pos，加 current_handler_role/agentid --- models/pipeline_agent_questions.json | 11 +- pipeline_service/agent_loop.py | 25 +-- pipeline_service/communication.py | 239 +++++++-------------------- pipeline_service/init.py | 5 +- pipeline_service/questions.py | 43 ++--- pipeline_service/sdlc_ability.py | 8 +- 6 files changed, 111 insertions(+), 220 deletions(-) diff --git a/models/pipeline_agent_questions.json b/models/pipeline_agent_questions.json index a849524..a1453de 100644 --- a/models/pipeline_agent_questions.json +++ b/models/pipeline_agent_questions.json @@ -2,22 +2,22 @@ "summary": [ { "name": "pipeline_agent_questions", - "title": "角色agent问题回路表", + "title": "团队沟通问题表", "primary": ["id"], "catelog": "entity" } ], "fields": [ {"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no"}, - {"name": "tenant_id", "title": "租户ID", "type": "str", "length": 32, "nullable": "no"}, + {"name": "tenant_id", "title": "项目ID", "type": "str", "length": 32, "nullable": "no"}, {"name": "task_id", "title": "任务ID", "type": "str", "length": 32, "nullable": "no"}, {"name": "from_role", "title": "提问角色", "type": "str", "length": 32, "nullable": "no"}, {"name": "from_agentid", "title": "提问Agent标识", "type": "str", "length": 64, "nullable": "yes"}, {"name": "problem_type", "title": "问题类型", "type": "str", "length": 50, "nullable": "yes"}, {"name": "question", "title": "问题内容", "type": "text"}, {"name": "context", "title": "问题上下文(JSON)", "type": "text"}, - {"name": "escalation_path", "title": "冒泡路径(JSON角色/agent数组快照)", "type": "text"}, - {"name": "escalation_pos", "title": "当前冒泡位置(处理方下标)", "type": "int", "default": "0"}, + {"name": "current_handler_role", "title": "当前处理角色", "type": "str", "length": 32, "nullable": "yes"}, + {"name": "current_handler_agentid", "title": "当前处理Agent标识(空=该角色任意)", "type": "str", "length": 64, "nullable": "yes"}, {"name": "answer", "title": "回答内容", "type": "text"}, {"name": "answered_by", "title": "回答人", "type": "str", "length": 64, "nullable": "yes"}, {"name": "answer_source", "title": "回答来源", "type": "str", "length": 16, "nullable": "yes"}, @@ -28,7 +28,8 @@ "indexes": [ {"name": "idx_paq_tenant", "idxtype": "index", "idxfields": ["tenant_id"]}, {"name": "idx_paq_task", "idxtype": "index", "idxfields": ["task_id"]}, - {"name": "idx_paq_status", "idxtype": "index", "idxfields": ["status"]} + {"name": "idx_paq_status", "idxtype": "index", "idxfields": ["status"]}, + {"name": "idx_paq_handler", "idxtype": "index", "idxfields": ["current_handler_role", "current_handler_agentid"]} ], "codes": [ {"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='question_state'"} diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 2b0fa7d..5c93bfe 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -441,8 +441,7 @@ async def _get_workspace_dir(sor, project_id): async def _build_qna_section(sor, task_id, role, agent_id=None): from .communication import get_task_qa - qa = await get_task_qa(task_id, include_pending_for=role, - include_pending_agentid=agent_id) + qa = await get_task_qa(task_id, role=role, agentid=agent_id) qna = qa.get('answered', []) or [] if isinstance(qa, dict) else [] # pending = 当前该「本角色/本 agent」处理的退回意见，重新认领时必须先逐条响应再产出。 pending = qa.get('pending', []) or [] if isinstance(qa, dict) else [] @@ -810,9 +809,12 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): if ask_question: from .communication import raise_problem - qid = await raise_problem(project_id, "need_info", role, ask_question, - task_id=task_id, tenant_id=project_id, - from_agentid=agent_id, context={"title": title}) + # need_info：角色缺信息向上提问，首处理方=主 agent + 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", + context={"title": title}) return {"status": "need_info", "task_id": task_id, "question_id": qid, "question": ask_question} if not deliverable: @@ -1008,11 +1010,11 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): elif status == 'rejected': from .communication import raise_problem rejection_q = decision.get("questions") or comment or "交付件不满足要求" - # 审核退回：路由到被退角色（该角色任意 agent 重新认领响应）。 + # 审核退回：首处理方=被退角色（该角色任意 agent 重新认领响应）。 # suspend_task=False —— 任务回 submitted（重新认领）而非 waiting（挂起等回答）。 - await raise_problem(project_id, "review_reject", "pm", rejection_q, - task_id=task_id, tenant_id=project_id, - first_handler=task_role, + await raise_problem("review_reject", rejection_q, "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}) @@ -1099,9 +1101,10 @@ async def handle_failed_task(task_id: str, project_id: str) -> dict: from .communication import raise_problem reporter = "cockpit" if role == "pm" else "pm" qid = await raise_problem( - project_id, "fault_report", reporter, + "fault_report", f"任务「{title}」已失败 {retry_count} 次，自动重试无法解决，请人工介入。失败原因：{last_error or '未知'}", - task_id=task_id, tenant_id=project_id, + reporter, tenant_id=project_id, task_id=task_id, + first_handler_role="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 3c15e69..7b21559 100644 --- a/pipeline_service/communication.py +++ b/pipeline_service/communication.py @@ -1,18 +1,17 @@ -"""团队沟通引擎 — 通用问题冒泡（多 agent 感知）。 +"""团队沟通引擎 — 通用问题冒泡（显式路由版）。 -问题类型(pipeline_problem_types)定义冒泡路径(escalation_path，处理方序列)。 -处理方 = role + agentid（同一角色可能有多个 agent，必须精确到具体 agent； -人类角色 agentid 为空表示"该角色的任意人"）。 - -问题产生后从路径第一个处理方开始，每个处理方要么解决(resolve)要么未解决 -继续冒泡(escalate)到下一个： -- 解决了 → status=answered，不再冒泡 -- 没解决 → escalation_pos+1，沿路径继续走 -- 路径尽头通常是"人"(human 角色)兜底 - -冒泡路径条目格式： -- "role_name" → 该角色任意 agent/人 -- {"role":"x", "agentid":"y"} → 指定角色 + 指定 agent +核心思想（规范在 skill，状态在表，工具固化通用）： +- 团队角色、问题类型、冒泡路径 = 写在 team-communication skill 里（规范/知识）， + 由 LLM 读 skill 理解后决策「问题该转给谁」。 +- 本模块只固化「问题操作」的通用工具（产线无关，不读任何配置表）： + raise_problem 提问题，显式指定首处理方(role+agentid) + resolve_problem 解决 → 停止冒泡(status=answered) + escalate_problem 没解决 → 显式指定下一个处理方(role+agentid) + list_problems_for 查「当前该我处理」的问题（按 current_handler 两列确定性过滤） +- 问题实例落最小表 pipeline_agent_questions，只存状态、不存冒泡路径快照 + （路径是 skill 里的知识，改 skill 后历史数据不残留旧路径）。 +- 处理方 = role + agentid（多 agent 存在，须精确到具体 agent； + agentid 空 = 该角色任意 agent 可处理）。 """ import json @@ -26,8 +25,6 @@ logger = logging.getLogger("pipeline.communication") S_PENDING = "pending" # 冒泡中（未解决） S_ANSWERED = "answered" # 已解决，停止冒泡 -DEFAULT_PIPELINE_ID = "sdlc_general" - def _get_db(): db = DBPools() @@ -39,114 +36,28 @@ def _get_db(): return db, DBNAME -def _parse_path(raw): - """解析冒泡路径为 [{"role":..., "agentid":...}] 列表（统一归一化）。""" - if not raw: - return [] - if isinstance(raw, (list, tuple)): - items = raw - else: - try: - items = json.loads(raw) - except (json.JSONDecodeError, TypeError): - return [] - if not isinstance(items, list): - return [] - result = [] - for it in items: - if isinstance(it, str): - result.append({"role": it, "agentid": ""}) - elif isinstance(it, dict): - result.append({ - "role": str(it.get("role", "") or ""), - "agentid": str(it.get("agentid", "") or ""), - }) - return [x for x in result if x["role"]] - - -def current_handler_of(rec) -> dict: - """当前该处理这个问题的处理方 {"role", "agentid"}（由 escalation_path + escalation_pos 决定）。""" - path = _parse_path(getattr(rec, 'escalation_path', '')) - try: - pos = int(getattr(rec, 'escalation_pos', 0) or 0) - except (TypeError, ValueError): - pos = 0 - if not path: - # 旧数据/无路径兜底：待主 agent 处理 - return {"role": "main_agent", "agentid": ""} - if pos < 0 or pos >= len(path): - return path[-1] - return path[pos] - - -def _handler_matches(handler: dict, role: str, agentid: str = None) -> bool: - """处理方是否匹配目标(role, agentid)。agentid 为空=任意 agent。""" - if handler.get("role") != role: - return False - if agentid and handler.get("agentid") and handler.get("agentid") != agentid: - return False # 指定了具体 agent，但处理方是别的 agent - return True - - -async def _resolve_pipeline_id(sor, project_id): - """从项目解析产线 id；查不到用默认 sdlc_general。""" - if project_id: - recs = await sor.sqlExe( - "SELECT pipeline_id FROM sd_projects WHERE id=${p}$", {"p": project_id}) - if recs: - pid = getattr(recs[0], 'pipeline_id', '') or '' - if pid: - return pid - return DEFAULT_PIPELINE_ID - - -async def get_problem_type(sor, pipeline_id, name): - """查问题类型定义（含冒泡路径，已归一化）。找不到返回 None。""" - recs = await sor.sqlExe( - "SELECT * FROM pipeline_problem_types " - "WHERE pipeline_id=${p}$ AND name=${n}$ LIMIT 1", - {"p": pipeline_id, "n": name}) - if not recs: - return None - rec = recs[0] - return { - 'id': getattr(rec, 'id', ''), - 'pipeline_id': pipeline_id, - 'name': name, - 'title': getattr(rec, 'title', '') or name, - 'escalation_path': _parse_path(getattr(rec, 'escalation_path', '')), - } - - -async def raise_problem(pipeline_id, problem_type, from_role, question, - task_id=None, tenant_id=None, context=None, - from_agentid=None, first_handler=None, - first_handler_agentid=None, suspend_task=True) -> str: - """提出问题。按问题类型的冒泡路径路由到第一个处理方。 +async def raise_problem(problem_type, question, from_role, from_agentid="", + tenant_id=None, task_id=None, context=None, + first_handler_role=None, first_handler_agentid="", + suspend_task=True) -> str: + """提出问题。首处理方由调用方（读 skill 后）显式指定。 Args: - pipeline_id: 产线 id（为空则从 tenant_id/project 解析） - problem_type: 问题类型标识（pipeline_problem_types.name） + problem_type: 问题类型标识（skill 里定义的字符串，如 need_info/review_reject/fault_report） + question: 问题内容 from_role: 谁提的（角色名） - from_agentid: 提问的具体 agent 标识（多 agent 场景必传） - first_handler: 可选，指定第一个处理角色（如审核退回指定被退角色） - first_handler_agentid: 可选，第一个处理方具体 agent - suspend_task: 是否把任务置 waiting（角色 agent 提问时挂起任务等回答） + from_agentid: 提问的具体 agent 标识（多 agent 场景） + first_handler_role: 首处理角色（不传默认 main_agent） + first_handler_agentid: 首处理方具体 agent（空=该角色任意 agent） + suspend_task: 是否把任务置 waiting（角色提问挂起等回答） Returns: question_id """ db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: - if not pipeline_id: - pipeline_id = await _resolve_pipeline_id(sor, tenant_id) - pt = await get_problem_type(sor, pipeline_id, problem_type) - path = pt['escalation_path'] if pt else [] - if first_handler and first_handler not in [p["role"] for p in path]: - path = [{"role": first_handler, "agentid": first_handler_agentid or ""}] + path - if not path: - path = [{"role": from_role or "main_agent", "agentid": from_agentid or ""}] - + hr = first_handler_role or "main_agent" + ha = first_handler_agentid or "" qid = getID() ctx_json = json.dumps(context, ensure_ascii=False, default=str) if context else None await sor.C('pipeline_agent_questions', { @@ -158,8 +69,8 @@ async def raise_problem(pipeline_id, problem_type, from_role, question, 'problem_type': problem_type or '', 'question': question or '', 'context': ctx_json, - 'escalation_path': json.dumps(path, ensure_ascii=False), - 'escalation_pos': 0, + 'current_handler_role': hr, + 'current_handler_agentid': ha, 'status': S_PENDING, }) if suspend_task and task_id: @@ -167,12 +78,12 @@ async def raise_problem(pipeline_id, problem_type, from_role, question, await sor.sqlExe( "UPDATE pipeline_tasks SET state='waiting', claimed_by=NULL WHERE id=${tid}$", {"tid": task_id}) - logger.info("raise_problem: pipeline=%s type=%s from=%s/%s qid=%s path=%s", - pipeline_id, problem_type, from_role, from_agentid, qid, path) + logger.info("raise_problem: type=%s from=%s/%s qid=%s -> handler=%s/%s", + problem_type, from_role, from_agentid, qid, hr, ha) 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="main_agent", resume_task=True): """当前处理方解决了问题 → 停止冒泡(status=answered)，任务恢复 submitted。""" db, dbname = _get_db() @@ -206,51 +117,43 @@ async def resolve_problem(question_id, answer, answered_by='', answer_source='ma return {'id': question_id, 'status': S_ANSWERED, 'resumed': resumed} -async def escalate_problem(question_id, by_role='', by_agentid='', note=''): - """当前处理方没解决 → 冒泡到路径下一个处理方（pos+1）。""" +async def escalate_problem(question_id, next_handler_role, next_handler_agentid="", + by_role="", by_agentid=""): + """当前处理方没解决 → 显式转给下一个处理方(role+agentid)。 + + 下一个处理方由调用方读 skill（冒泡路径）后决定，不读 DB 里的路径快照。 + """ db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: recs = await sor.R('pipeline_agent_questions', {'id': question_id}) if not recs: return None - rec = recs[0] - path = _parse_path(getattr(rec, 'escalation_path', '')) - try: - pos = int(getattr(rec, 'escalation_pos', 0) or 0) - except (TypeError, ValueError): - pos = 0 - new_pos = pos + 1 - if path and new_pos >= len(path): - new_pos = len(path) - 1 # 尽头：停在最后一个处理方（通常是"人"兜底） - elif not path: - new_pos = 0 await sor.U('pipeline_agent_questions', { 'id': question_id, - 'escalation_pos': new_pos, + 'current_handler_role': next_handler_role or '', + 'current_handler_agentid': next_handler_agentid or '', }) - handler = path[new_pos] if path else {} - logger.info("escalate_problem: qid=%s by=%s/%s -> %s (pos %d)", - question_id, by_role, by_agentid, handler, new_pos) - return {'id': question_id, 'escalation_pos': new_pos, 'current_handler': handler} - - -async def forward_to_customer(question_id, by_role='main_agent', by_agentid=''): - """便捷封装：主 agent 答不了，直接冒泡给客户(人)。等价于 escalate。""" - return await escalate_problem(question_id, by_role=by_role, by_agentid=by_agentid) + logger.info("escalate_problem: qid=%s by=%s/%s -> %s/%s", + question_id, by_role, by_agentid, + next_handler_role, next_handler_agentid) + return {'id': question_id, 'current_handler_role': next_handler_role, + 'current_handler_agentid': next_handler_agentid} async def list_problems_for(role, agentid=None, tenant_id=None, task_id=None, limit=50) -> list: - """列出「当前该 (role, agentid) 处理」的 pending 问题。 + """列出「当前该 (role, agentid) 处理」的 pending 问题（确定性 SQL 过滤）。 - 多 agent 感知：agentid 指定时，只返回"该 agent 处理"或"该角色任意 agent 处理" - 的问题；不指定 agentid 时返回该角色所有待处理问题。 - - 替代旧版「按 status 无差别列全部 pending」——把归属不同的问题混给同一个处理者。 + agentid 指定时，返回「该 agent 处理」或「该角色任意 agent 处理」的问题； + 不指定 agentid 时返回该角色所有待处理问题。 """ db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: - conditions = ["status='pending'"] - params = {} + conditions = ["status='pending'", "current_handler_role=${role}$"] + params = {"role": role} + if agentid: + conditions.append( + "(current_handler_agentid='' OR current_handler_agentid=${agentid}$)") + params["agentid"] = agentid if tenant_id: conditions.append("tenant_id=${tenant_id}$") params["tenant_id"] = tenant_id @@ -262,19 +165,14 @@ async def list_problems_for(role, agentid=None, tenant_id=None, task_id=None, li limit = int(limit) except (TypeError, ValueError): limit = 50 - sql = f"SELECT * FROM pipeline_agent_questions WHERE {where} ORDER BY created_at DESC LIMIT {limit}" + sql = (f"SELECT * FROM pipeline_agent_questions WHERE {where} " + f"ORDER BY created_at DESC LIMIT {limit}") recs = await sor.sqlExe(sql, params) # 释放 SELECT 元数据锁 await sor.sqlExe("COMMIT", {}) - result = [] for rec in (recs or []): - h = current_handler_of(rec) - if not _handler_matches(h, role, agentid): - continue - d = _rec_to_dict(rec) - d['current_handler'] = h - result.append(d) + result.append(_rec_to_dict(rec)) return result @@ -285,13 +183,12 @@ def _rec_to_dict(rec): return dict(rec) -async def get_task_qa(task_id, include_pending_for=None, - include_pending_agentid=None) -> dict: - """取任务的历史问答（已回答）+ 可选「待某处理方处理」的 pending 退回意见。 +async def get_task_qa(task_id, role=None, agentid=None) -> dict: + """取任务的历史问答(answered) + 可选「待该处理方」的 pending 退回意见。 Args: - include_pending_for: 角色名，额外返回当前该角色应响应的 pending 问题 - include_pending_agentid: 具体 agent 标识（多 agent 场景） + role: 若给角色名，pending 只返回当前该角色应响应的（按 current_handler 过滤） + agentid: 具体 agent 标识 Returns: {"answered": [...], "pending": [...]} @@ -312,18 +209,6 @@ async def get_task_qa(task_id, include_pending_for=None, }) pending = [] - if include_pending_for: - precs = await sor.sqlExe( - "SELECT * FROM pipeline_agent_questions " - "WHERE task_id=${tid}$ AND status='pending' ORDER BY created_at ASC", - {"tid": task_id}) - for rec in (precs or []): - h = current_handler_of(rec) - if _handler_matches(h, include_pending_for, include_pending_agentid): - pending.append({ - "question": getattr(rec, 'question', '') or '', - "from_role": getattr(rec, 'from_role', '') or '', - "problem_type": getattr(rec, 'problem_type', '') or '', - "current_handler": h, - }) + if role: + pending = await list_problems_for(role, agentid=agentid, task_id=task_id) return {"answered": answered, "pending": pending} diff --git a/pipeline_service/init.py b/pipeline_service/init.py index a273e85..36669e0 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -39,7 +39,7 @@ from .questions import ( ) from .communication import ( raise_problem, resolve_problem, escalate_problem, - forward_to_customer, list_problems_for, get_task_qa, + list_problems_for, get_task_qa, ) 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 @@ -543,11 +543,10 @@ def load_pipeline_service(): env.question_list = list_questions env.question_qna = get_task_qna - # 团队沟通（通用问题冒泡引擎，多 agent 感知） + # 团队沟通（通用问题冒泡引擎，多 agent 感知；规范在 team-communication skill） env.raise_problem = raise_problem env.resolve_problem = resolve_problem env.escalate_problem = escalate_problem - env.problem_forward = forward_to_customer env.problem_list_for = list_problems_for env.problem_qa = get_task_qa diff --git a/pipeline_service/questions.py b/pipeline_service/questions.py index 67b4628..4e7a7c7 100644 --- a/pipeline_service/questions.py +++ b/pipeline_service/questions.py @@ -1,11 +1,9 @@ """问题回路 — 向后兼容层，委托 communication.py 通用冒泡引擎。 -历史状态 pending/forwarded/answered 收敛为通用模型： -- 问题类型(pipeline_problem_types)定义冒泡路径(escalation_path) -- 问题沿路径冒泡：解决→answered(停)，没解决→pos+1(沿路径走)，尽头=人兜底 -- 归属不再靠 status 无差别路由，而靠 current_handler(escalation_path+pos) +规范（团队角色/问题类型/冒泡路径）在 team-communication skill 里； +本模块只保留旧签名供 DSPY(env.question_*) 与既有调用方使用，内部走 communication.py。 -本文件保留旧签名供 DSPY(env.question_*) 与既有调用方使用，内部全部走 communication.py。 +状态只有 pending/answered；归属不再靠 status，而靠 current_handler_role/agentid 两列。 """ import json @@ -15,7 +13,7 @@ from sqlor.dbpools import DBPools from .communication import ( raise_problem, resolve_problem, - forward_to_customer, + escalate_problem, list_problems_for, get_task_qa, ) @@ -36,13 +34,19 @@ def _get_db(): async def agent_ask(tenant_id: str, task_id: str, from_role: str, question: str, context: dict = None) -> str: - """角色 agent / PM 提问。按 from_role 推断问题类型，走通用冒泡引擎。""" - # 审核退回(PM) vs 故障报告(cockpit/pm) vs 缺信息(角色)： - ptype = "review_reject" if from_role == "pm" else ( - "fault_report" if from_role == "cockpit" else "need_info") - return await raise_problem( - "", ptype, from_role, question, - task_id=task_id, tenant_id=tenant_id, context=context) + """旧签名：按 from_role 推断问题类型 + 首处理方，走通用冒泡引擎。 + + 新代码请直接调 communication.raise_problem 显式指定首处理方。 + """ + if from_role == "pm": + ptype, hr = "review_reject", "develop" + elif from_role == "cockpit": + ptype, hr = "fault_report", "main_agent" + else: + ptype, hr = "need_info", "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 = "", @@ -52,9 +56,11 @@ async def answer_question(question_id: str, answer: str, answered_by: str = "", resume_task=resume) -async def forward_question(question_id: str): - """主 agent 答不了 → 沿路径冒泡给客户(人)。""" - return await forward_to_customer(question_id) +async def forward_question(question_id: str, next_handler_role: str = "customer", + next_handler_agentid: str = ""): + """答不了 → 沿冒泡路径转给下一个处理方（默认客户）。""" + return await escalate_problem(question_id, next_handler_role, next_handler_agentid, + by_role="main_agent") async def get_question(question_id: str): @@ -80,10 +86,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: - """客户视角：列出当前该「客户(人)」处理的问题。 - - 旧签名兼容 status 过滤；无 status 时列客户待处理(pending)。 - """ + """客户视角：列出当前该「客户(人)」处理的问题。""" rows = await list_problems_for("customer", tenant_id=tenant_id, task_id=task_id, limit=limit) if status: diff --git a/pipeline_service/sdlc_ability.py b/pipeline_service/sdlc_ability.py index 7e6228f..33ee383 100644 --- a/pipeline_service/sdlc_ability.py +++ b/pipeline_service/sdlc_ability.py @@ -449,7 +449,7 @@ async def _h_escalate_question(sor, p, ctx): qid = p.get("question_id", "") if not qid: return "需要问题ID" - from .communication import forward_to_customer + from .communication import escalate_problem # 模糊匹配短 id full_qid = qid if len(qid) < 32: @@ -458,11 +458,11 @@ async def _h_escalate_question(sor, p, ctx): {"prefix": qid + "%"}) if recs: full_qid = getattr(recs[0], "id", qid) - r = await forward_to_customer(full_qid, by_role="main_agent") + # 主 agent 答不了 → 沿冒泡路径转给客户（下一个处理方由 team-communication skill 决定） + r = await escalate_problem(full_qid, "customer", by_role="main_agent") if r is None: return f"问题不存在: {qid}" - h = r.get("current_handler", {}) or {} - return f"OK: 已沿冒泡路径转给 {h.get('role', '?')}" + return f"OK: 已沿冒泡路径转给 {r.get('current_handler_role', 'customer')}" async def _h_answer_question(sor, p, ctx):