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

refactor(team-communication): 问题冒泡改为显式路由，规范移交 skill

Browse Source 
规范(团队角色/问题类型/冒泡路径)移交 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
This commit is contained in:
ymq 2026-08-16 22:08:00 +08:00
parent 4f2c6da6cf
commit 6707cabedc
6 changed files with 111 additions and 220 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
models/pipeline_agent_questions.json
View File

@ -2,22 +2,22 @@
"summary": [ "summary": [
{ {
"name": "pipeline_agent_questions", "name": "pipeline_agent_questions",
"title": "角色agent问题回路表", "title": "团队沟通问题表",
"primary": ["id"], "primary": ["id"],
"catelog": "entity" "catelog": "entity"
} }
], ],
"fields": [ "fields": [
{"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no"}, {"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": "task_id", "title": "任务ID", "type": "str", "length": 32, "nullable": "no"},
{"name": "from_role", "title": "提问角色", "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": "from_agentid", "title": "提问Agent标识", "type": "str", "length": 64, "nullable": "yes"},
{"name": "problem_type", "title": "问题类型", "type": "str", "length": 50, "nullable": "yes"}, {"name": "problem_type", "title": "问题类型", "type": "str", "length": 50, "nullable": "yes"},
{"name": "question", "title": "问题内容", "type": "text"}, {"name": "question", "title": "问题内容", "type": "text"},
{"name": "context", "title": "问题上下文(JSON)", "type": "text"}, {"name": "context", "title": "问题上下文(JSON)", "type": "text"},
{"name": "escalation_path", "title": "冒泡路径(JSON角色/agent数组快照)", "type": "text"}, {"name": "current_handler_role", "title": "当前处理角色", "type": "str", "length": 32, "nullable": "yes"},
{"name": "escalation_pos", "title": "当前冒泡位置(处理方下标)", "type": "int", "default": "0"}, {"name": "current_handler_agentid", "title": "当前处理Agent标识(空=该角色任意)", "type": "str", "length": 64, "nullable": "yes"},
{"name": "answer", "title": "回答内容", "type": "text"}, {"name": "answer", "title": "回答内容", "type": "text"},
{"name": "answered_by", "title": "回答人", "type": "str", "length": 64, "nullable": "yes"}, {"name": "answered_by", "title": "回答人", "type": "str", "length": 64, "nullable": "yes"},
{"name": "answer_source", "title": "回答来源", "type": "str", "length": 16, "nullable": "yes"}, {"name": "answer_source", "title": "回答来源", "type": "str", "length": 16, "nullable": "yes"},
@ -28,7 +28,8 @@
"indexes": [ "indexes": [
{"name": "idx_paq_tenant", "idxtype": "index", "idxfields": ["tenant_id"]}, {"name": "idx_paq_tenant", "idxtype": "index", "idxfields": ["tenant_id"]},
{"name": "idx_paq_task", "idxtype": "index", "idxfields": ["task_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": [ "codes": [
{"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='question_state'"} {"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='question_state'"}

25
pipeline_service/agent_loop.py
View File

@ -441,8 +441,7 @@ async def _get_workspace_dir(sor, project_id):
async def _build_qna_section(sor, task_id, role, agent_id=None): async def _build_qna_section(sor, task_id, role, agent_id=None):
from .communication import get_task_qa from .communication import get_task_qa
qa = await get_task_qa(task_id, include_pending_for=role, qa = await get_task_qa(task_id, role=role, agentid=agent_id)
include_pending_agentid=agent_id)
qna = qa.get('answered', []) or [] if isinstance(qa, dict) else [] qna = qa.get('answered', []) or [] if isinstance(qa, dict) else []
# pending = 当前该「本角色/本 agent」处理的退回意见，重新认领时必须先逐条响应再产出。 # pending = 当前该「本角色/本 agent」处理的退回意见，重新认领时必须先逐条响应再产出。
pending = qa.get('pending', []) or [] if isinstance(qa, dict) else [] 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: if ask_question:
from .communication import raise_problem from .communication import raise_problem
qid = await raise_problem(project_id, "need_info", role, ask_question, # need_info：角色缺信息向上提问，首处理方=主 agent
task_id=task_id, tenant_id=project_id, qid = await raise_problem("need_info", ask_question, role,
from_agentid=agent_id, context={"title": title}) 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} return {"status": "need_info", "task_id": task_id, "question_id": qid, "question": ask_question}
if not deliverable: if not deliverable:
@ -1008,11 +1010,11 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
elif status == 'rejected': elif status == 'rejected':
from .communication import raise_problem from .communication import raise_problem
rejection_q = decision.get("questions") or comment or "交付件不满足要求" rejection_q = decision.get("questions") or comment or "交付件不满足要求"
# 审核退回：路由到被退角色（该角色任意 agent 重新认领响应）。 # 审核退回：首处理方=被退角色（该角色任意 agent 重新认领响应）。
# suspend_task=False —— 任务回 submitted（重新认领）而非 waiting（挂起等回答）。 # suspend_task=False —— 任务回 submitted（重新认领）而非 waiting（挂起等回答）。
await raise_problem(project_id, "review_reject", "pm", rejection_q, await raise_problem("review_reject", rejection_q, "pm",
task_id=task_id, tenant_id=project_id, tenant_id=project_id, task_id=task_id,
first_handler=task_role, first_handler_role=task_role,
context={"pm_comment": comment, "deliverable_type": deliverable_type}, context={"pm_comment": comment, "deliverable_type": deliverable_type},
suspend_task=False) suspend_task=False)
await sor.sqlExe("UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL WHERE id=${tid}$", {"tid": task_id}) 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 from .communication import raise_problem
reporter = "cockpit" if role == "pm" else "pm" reporter = "cockpit" if role == "pm" else "pm"
qid = await raise_problem( qid = await raise_problem(
project_id, "fault_report", reporter, "fault_report",
f"任务「{title}」已失败 {retry_count} 次，自动重试无法解决，请人工介入。失败原因：{last_error or '未知'}", 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}) 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) 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} return {"status": "fault", "task_id": task_id, "question_id": qid}

239
pipeline_service/communication.py
View File

@ -1,18 +1,17 @@
"""团队沟通引擎 — 通用问题冒泡（多 agent 感知）。 """团队沟通引擎 — 通用问题冒泡（显式路由版）。
问题类型(pipeline_problem_types)定义冒泡路径(escalation_path，处理方序列)。 核心思想（规范在 skill，状态在表，工具固化通用）：
处理方 = role + agentid（同一角色可能有多个 agent，必须精确到具体 agent； - 团队角色、问题类型、冒泡路径 = 写在 team-communication skill 里（规范/知识），
人类角色 agentid 为空表示"该角色的任意人"）。 由 LLM 读 skill 理解后决策「问题该转给谁」。
- 本模块只固化「问题操作」的通用工具（产线无关，不读任何配置表）：
问题产生后从路径第一个处理方开始，每个处理方要么解决(resolve)要么未解决 raise_problem 提问题，显式指定首处理方(role+agentid)
继续冒泡(escalate)到下一个： resolve_problem 解决 → 停止冒泡(status=answered)
- 解决了 → status=answered，不再冒泡 escalate_problem 没解决 → 显式指定下一个处理方(role+agentid)
- 没解决 → escalation_pos+1，沿路径继续走 list_problems_for 查「当前该我处理」的问题（按 current_handler 两列确定性过滤）
- 路径尽头通常是"人"(human 角色)兜底 - 问题实例落最小表 pipeline_agent_questions，只存状态、不存冒泡路径快照
（路径是 skill 里的知识，改 skill 后历史数据不残留旧路径）。
冒泡路径条目格式： - 处理方 = role + agentid（多 agent 存在，须精确到具体 agent；
- "role_name" → 该角色任意 agent/人 agentid 空 = 该角色任意 agent 可处理）。
- {"role":"x", "agentid":"y"} → 指定角色 + 指定 agent
""" """
import json import json
@ -26,8 +25,6 @@ logger = logging.getLogger("pipeline.communication")
S_PENDING = "pending" # 冒泡中（未解决） S_PENDING = "pending" # 冒泡中（未解决）
S_ANSWERED = "answered" # 已解决，停止冒泡 S_ANSWERED = "answered" # 已解决，停止冒泡
DEFAULT_PIPELINE_ID = "sdlc_general"
def _get_db(): def _get_db():
db = DBPools() db = DBPools()
@ -39,114 +36,28 @@ def _get_db():
return db, DBNAME return db, DBNAME
def _parse_path(raw): async def raise_problem(problem_type, question, from_role, from_agentid="",
"""解析冒泡路径为 [{"role":..., "agentid":...}] 列表（统一归一化）。""" tenant_id=None, task_id=None, context=None,
if not raw: first_handler_role=None, first_handler_agentid="",
return [] suspend_task=True) -> str:
if isinstance(raw, (list, tuple)): """提出问题。首处理方由调用方（读 skill 后）显式指定。
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:
"""提出问题。按问题类型的冒泡路径路由到第一个处理方。
Args: Args:
pipeline_id: 产线 id（为空则从 tenant_id/project 解析） problem_type: 问题类型标识（skill 里定义的字符串，如 need_info/review_reject/fault_report）
problem_type: 问题类型标识（pipeline_problem_types.name） question: 问题内容
from_role: 谁提的（角色名） from_role: 谁提的（角色名）
from_agentid: 提问的具体 agent 标识（多 agent 场景必传） from_agentid: 提问的具体 agent 标识（多 agent 场景）
first_handler: 可选，指定第一个处理角色（如审核退回指定被退角色） first_handler_role: 首处理角色（不传默认 main_agent）
first_handler_agentid: 可选，第一个处理方具体 agent first_handler_agentid: 首处理方具体 agent（空=该角色任意 agent）
suspend_task: 是否把任务置 waiting（角色 agent 提问时挂起任务等回答） suspend_task: 是否把任务置 waiting（角色提问挂起等回答）
Returns: Returns:
question_id question_id
""" """
db, dbname = _get_db() db, dbname = _get_db()
async with db.sqlorContext(dbname) as sor: async with db.sqlorContext(dbname) as sor:
if not pipeline_id: hr = first_handler_role or "main_agent"
pipeline_id = await _resolve_pipeline_id(sor, tenant_id) ha = first_handler_agentid or ""
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 ""}]
qid = getID() qid = getID()
ctx_json = json.dumps(context, ensure_ascii=False, default=str) if context else None ctx_json = json.dumps(context, ensure_ascii=False, default=str) if context else None
await sor.C('pipeline_agent_questions', { 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 '', 'problem_type': problem_type or '',
'question': question or '', 'question': question or '',
'context': ctx_json, 'context': ctx_json,
'escalation_path': json.dumps(path, ensure_ascii=False), 'current_handler_role': hr,
'escalation_pos': 0, 'current_handler_agentid': ha,
'status': S_PENDING, 'status': S_PENDING,
}) })
if suspend_task and task_id: if suspend_task and task_id:
@ -167,12 +78,12 @@ async def raise_problem(pipeline_id, problem_type, from_role, question,
await sor.sqlExe( await sor.sqlExe(
"UPDATE pipeline_tasks SET state='waiting', claimed_by=NULL WHERE id=${tid}$", "UPDATE pipeline_tasks SET state='waiting', claimed_by=NULL WHERE id=${tid}$",
{"tid": task_id}) {"tid": task_id})
logger.info("raise_problem: pipeline=%s type=%s from=%s/%s qid=%s path=%s", logger.info("raise_problem: type=%s from=%s/%s qid=%s -> handler=%s/%s",
pipeline_id, problem_type, from_role, from_agentid, qid, path) problem_type, from_role, from_agentid, qid, hr, ha)
return qid 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): resume_task=True):
"""当前处理方解决了问题 → 停止冒泡(status=answered)，任务恢复 submitted。""" """当前处理方解决了问题 → 停止冒泡(status=answered)，任务恢复 submitted。"""
db, dbname = _get_db() 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} return {'id': question_id, 'status': S_ANSWERED, 'resumed': resumed}
async def escalate_problem(question_id, by_role='', by_agentid='', note=''): async def escalate_problem(question_id, next_handler_role, next_handler_agentid="",
"""当前处理方没解决 → 冒泡到路径下一个处理方（pos+1）。""" by_role="", by_agentid=""):
"""当前处理方没解决 → 显式转给下一个处理方(role+agentid)。
下一个处理方由调用方读 skill（冒泡路径）后决定，不读 DB 里的路径快照。
"""
db, dbname = _get_db() db, dbname = _get_db()
async with db.sqlorContext(dbname) as sor: async with db.sqlorContext(dbname) as sor:
recs = await sor.R('pipeline_agent_questions', {'id': question_id}) recs = await sor.R('pipeline_agent_questions', {'id': question_id})
if not recs: if not recs:
return None 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', { await sor.U('pipeline_agent_questions', {
'id': question_id, '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/%s",
logger.info("escalate_problem: qid=%s by=%s/%s -> %s (pos %d)", question_id, by_role, by_agentid,
question_id, by_role, by_agentid, handler, new_pos) next_handler_role, next_handler_agentid)
return {'id': question_id, 'escalation_pos': new_pos, 'current_handler': handler} return {'id': question_id, 'current_handler_role': next_handler_role,
'current_handler_agentid': next_handler_agentid}
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)
async def list_problems_for(role, agentid=None, tenant_id=None, task_id=None, limit=50) -> list: 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 指定时，返回「该 agent 处理」或「该角色任意 agent 处理」的问题；
的问题；不指定 agentid 时返回该角色所有待处理问题。 不指定 agentid 时返回该角色所有待处理问题。
替代旧版「按 status 无差别列全部 pending」——把归属不同的问题混给同一个处理者。
""" """
db, dbname = _get_db() db, dbname = _get_db()
async with db.sqlorContext(dbname) as sor: async with db.sqlorContext(dbname) as sor:
conditions = ["status='pending'"] conditions = ["status='pending'", "current_handler_role=${role}$"]
params = {} params = {"role": role}
if agentid:
conditions.append(
"(current_handler_agentid='' OR current_handler_agentid=${agentid}$)")
params["agentid"] = agentid
if tenant_id: if tenant_id:
conditions.append("tenant_id=${tenant_id}$") conditions.append("tenant_id=${tenant_id}$")
params["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) limit = int(limit)
except (TypeError, ValueError): except (TypeError, ValueError):
limit = 50 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) recs = await sor.sqlExe(sql, params)
# 释放 SELECT 元数据锁 # 释放 SELECT 元数据锁
await sor.sqlExe("COMMIT", {}) await sor.sqlExe("COMMIT", {})
result = [] result = []
for rec in (recs or []): for rec in (recs or []):
h = current_handler_of(rec) result.append(_rec_to_dict(rec))
if not _handler_matches(h, role, agentid):
continue
d = _rec_to_dict(rec)
d['current_handler'] = h
result.append(d)
return result return result
@ -285,13 +183,12 @@ def _rec_to_dict(rec):
return dict(rec) return dict(rec)
async def get_task_qa(task_id, include_pending_for=None, async def get_task_qa(task_id, role=None, agentid=None) -> dict:
include_pending_agentid=None) -> dict: """取任务的历史问答(answered) + 可选「待该处理方」的 pending 退回意见。
"""取任务的历史问答（已回答）+ 可选「待某处理方处理」的 pending 退回意见。
Args: Args:
include_pending_for: 角色名，额外返回当前该角色应响应的 pending 问题 role: 若给角色名，pending 只返回当前该角色应响应的（按 current_handler 过滤）
include_pending_agentid: 具体 agent 标识（多 agent 场景） agentid: 具体 agent 标识
Returns: Returns:
{"answered": [...], "pending": [...]} {"answered": [...], "pending": [...]}
@ -312,18 +209,6 @@ async def get_task_qa(task_id, include_pending_for=None,
}) })
pending = [] pending = []
if include_pending_for: if role:
precs = await sor.sqlExe( pending = await list_problems_for(role, agentid=agentid, task_id=task_id)
"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,
})
return {"answered": answered, "pending": pending} return {"answered": answered, "pending": pending}

5
pipeline_service/init.py
View File

@ -39,7 +39,7 @@ from .questions import (
) )
from .communication import ( from .communication import (
raise_problem, resolve_problem, escalate_problem, 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 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, handle_failed_task
@ -543,11 +543,10 @@ def load_pipeline_service():
env.question_list = list_questions env.question_list = list_questions
env.question_qna = get_task_qna env.question_qna = get_task_qna
# 团队沟通（通用问题冒泡引擎，多 agent 感知） # 团队沟通（通用问题冒泡引擎，多 agent 感知；规范在 team-communication skill）
env.raise_problem = raise_problem env.raise_problem = raise_problem
env.resolve_problem = resolve_problem env.resolve_problem = resolve_problem
env.escalate_problem = escalate_problem env.escalate_problem = escalate_problem
env.problem_forward = forward_to_customer
env.problem_list_for = list_problems_for env.problem_list_for = list_problems_for
env.problem_qa = get_task_qa env.problem_qa = get_task_qa

43
pipeline_service/questions.py
View File

@ -1,11 +1,9 @@
"""问题回路 — 向后兼容层，委托 communication.py 通用冒泡引擎。 """问题回路 — 向后兼容层，委托 communication.py 通用冒泡引擎。
历史状态 pending/forwarded/answered 收敛为通用模型： 规范（团队角色/问题类型/冒泡路径）在 team-communication skill 里；
- 问题类型(pipeline_problem_types)定义冒泡路径(escalation_path) 本模块只保留旧签名供 DSPY(env.question_*) 与既有调用方使用，内部走 communication.py。
- 问题沿路径冒泡：解决→answered(停)，没解决→pos+1(沿路径走)，尽头=人兜底
- 归属不再靠 status 无差别路由，而靠 current_handler(escalation_path+pos)
本文件保留旧签名供 DSPY(env.question_*) 与既有调用方使用，内部全部走 communication.py。 状态只有 pending/answered；归属不再靠 status，而靠 current_handler_role/agentid 两列。
""" """
import json import json
@ -15,7 +13,7 @@ from sqlor.dbpools import DBPools
from .communication import ( from .communication import (
raise_problem, raise_problem,
resolve_problem, resolve_problem,
forward_to_customer, escalate_problem,
list_problems_for, list_problems_for,
get_task_qa, 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, async def agent_ask(tenant_id: str, task_id: str, from_role: str, question: str,
context: dict = None) -> str: context: dict = None) -> str:
"""角色 agent / PM 提问。按 from_role 推断问题类型，走通用冒泡引擎。""" """旧签名：按 from_role 推断问题类型 + 首处理方，走通用冒泡引擎。
# 审核退回(PM) vs 故障报告(cockpit/pm) vs 缺信息(角色)：
ptype = "review_reject" if from_role == "pm" else ( 新代码请直接调 communication.raise_problem 显式指定首处理方。
"fault_report" if from_role == "cockpit" else "need_info") """
return await raise_problem( if from_role == "pm":
"", ptype, from_role, question, ptype, hr = "review_reject", "develop"
task_id=task_id, tenant_id=tenant_id, context=context) 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 = "", 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) resume_task=resume)
async def forward_question(question_id: str): async def forward_question(question_id: str, next_handler_role: str = "customer",
"""主 agent 答不了 → 沿路径冒泡给客户(人)。""" next_handler_agentid: str = ""):
return await forward_to_customer(question_id) """答不了 → 沿冒泡路径转给下一个处理方（默认客户）。"""
return await escalate_problem(question_id, next_handler_role, next_handler_agentid,
by_role="main_agent")
async def get_question(question_id: str): 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, async def list_questions(tenant_id: str = None, task_id: str = None,
status: str = None, limit: int = 50) -> list: status: str = None, limit: int = 50) -> list:
"""客户视角：列出当前该「客户(人)」处理的问题。 """客户视角：列出当前该「客户(人)」处理的问题。"""
旧签名兼容 status 过滤；无 status 时列客户待处理(pending)。
"""
rows = await list_problems_for("customer", tenant_id=tenant_id, rows = await list_problems_for("customer", tenant_id=tenant_id,
task_id=task_id, limit=limit) task_id=task_id, limit=limit)
if status: if status:

8
pipeline_service/sdlc_ability.py
View File

@ -449,7 +449,7 @@ async def _h_escalate_question(sor, p, ctx):
qid = p.get("question_id", "") qid = p.get("question_id", "")
if not qid: if not qid:
return "需要问题ID" return "需要问题ID"
from .communication import forward_to_customer from .communication import escalate_problem
# 模糊匹配短 id # 模糊匹配短 id
full_qid = qid full_qid = qid
if len(qid) < 32: if len(qid) < 32:
@ -458,11 +458,11 @@ async def _h_escalate_question(sor, p, ctx):
{"prefix": qid + "%"}) {"prefix": qid + "%"})
if recs: if recs:
full_qid = getattr(recs[0], "id", qid) 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: if r is None:
return f"问题不存在: {qid}" return f"问题不存在: {qid}"
h = r.get("current_handler", {}) or {} return f"OK: 已沿冒泡路径转给 {r.get('current_handler_role', 'customer')}"
return f"OK: 已沿冒泡路径转给 {h.get('role', '?')}"
async def _h_answer_question(sor, p, ctx): async def _h_answer_question(sor, p, ctx):