diff --git a/models/pipeline_agent_questions.json b/models/pipeline_agent_questions.json index 5e1f55f..a849524 100644 --- a/models/pipeline_agent_questions.json +++ b/models/pipeline_agent_questions.json @@ -12,8 +12,12 @@ {"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": "answer", "title": "回答内容", "type": "text"}, {"name": "answered_by", "title": "回答人", "type": "str", "length": 64, "nullable": "yes"}, {"name": "answer_source", "title": "回答来源", "type": "str", "length": 16, "nullable": "yes"}, diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index d06cec7..2b0fa7d 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -439,22 +439,20 @@ async def _get_workspace_dir(sor, project_id): return _resolve_workspace(ws) -async def _build_qna_section(sor, task_id): - from .questions import get_task_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}) +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) + qna = qa.get('answered', []) or [] if isinstance(qa, dict) else [] + # pending = 当前该「本角色/本 agent」处理的退回意见，重新认领时必须先逐条响应再产出。 + pending = qa.get('pending', []) or [] if isinstance(qa, dict) else [] if not qna and not pending: return "" lines = [] if pending: lines.append("⚠️ 待处理问题（审核/PM 退回意见，你必须先逐条响应这些再产出交付件）：") for p in pending: - lines.append(f"- [{getattr(p, 'from_role', '') or '审核'}] {getattr(p, 'question', '')}") + lines.append(f"- [{p.get('from_role', '') or '审核'}] {p.get('question', '')}") if qna: lines.append("历史问答：") for item in qna: @@ -727,7 +725,7 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): logger.warning(f"role_agent_run setup_repos failed: {e}") role_specific = _role_specific - qna_section = await _build_qna_section(sor, task_id) + qna_section = await _build_qna_section(sor, task_id, role, agent_id) tools_text = json.dumps(AGENT_TOOLS, ensure_ascii=False) system = (AGENT_SYSTEM_PROMPT @@ -811,9 +809,10 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): break if ask_question: - from .questions import agent_ask - qid = await agent_ask(project_id, task_id, role, ask_question, - context={"title": title}) + 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}) return {"status": "need_info", "task_id": task_id, "question_id": qid, "question": ask_question} if not deliverable: @@ -989,12 +988,15 @@ 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，避免问题永久堆积 + # 审核通过：该任务 review_reject 类的 pending 退回意见已由角色 agent 响应 → 解决，避免永久堆积。 + # 只答结 review_reject（归属角色 agent 的），不碰 need_info（归属主 agent）等其它 pending。 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}) + "answer_source='role_agent', answered_by=${role}$ " + "WHERE task_id=${tid}$ AND status='pending' " + "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}) next_role = await _get_next_role(task_role, project_id) if next_role: @@ -1004,9 +1006,15 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"} elif status == 'rejected': - from .questions import agent_ask + from .communication import raise_problem rejection_q = decision.get("questions") or comment or "交付件不满足要求" - await agent_ask(project_id, task_id, "pm", rejection_q, context={"pm_comment": comment, "deliverable_type": deliverable_type}) + # 审核退回：路由到被退角色（该角色任意 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, + 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}) return {"status": "rejected", "task_id": task_id, "comment": comment, "question": rejection_q} @@ -1088,11 +1096,12 @@ async def handle_failed_task(task_id: str, project_id: str) -> dict: return {"status": "retry", "task_id": task_id, "attempt": retry_count + 1} # 报故障：建问题通知用户，任务置 waiting 等待人工介入 - from .questions import agent_ask + from .communication import raise_problem reporter = "cockpit" if role == "pm" else "pm" - qid = await agent_ask( - project_id, task_id, reporter, + qid = await raise_problem( + project_id, "fault_report", reporter, f"任务「{title}」已失败 {retry_count} 次，自动重试无法解决，请人工介入。失败原因：{last_error or '未知'}", + task_id=task_id, tenant_id=project_id, 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 new file mode 100644 index 0000000..3c15e69 --- /dev/null +++ b/pipeline_service/communication.py @@ -0,0 +1,329 @@ +"""团队沟通引擎 — 通用问题冒泡（多 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 +""" + +import json +import logging +from sqlor.dbpools import DBPools +from appPublic.uniqueID import getID + +DBNAME = "pipeline" +logger = logging.getLogger("pipeline.communication") + +S_PENDING = "pending" # 冒泡中（未解决） +S_ANSWERED = "answered" # 已解决，停止冒泡 + +DEFAULT_PIPELINE_ID = "sdlc_general" + + +def _get_db(): + db = DBPools() + if not db.databases: + from appPublic.jsonConfig import getConfig + config = getConfig() + if config.databases: + db.databases = config.databases + 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: + """提出问题。按问题类型的冒泡路径路由到第一个处理方。 + + Args: + pipeline_id: 产线 id（为空则从 tenant_id/project 解析） + problem_type: 问题类型标识（pipeline_problem_types.name） + from_role: 谁提的（角色名） + from_agentid: 提问的具体 agent 标识（多 agent 场景必传） + first_handler: 可选，指定第一个处理角色（如审核退回指定被退角色） + first_handler_agentid: 可选，第一个处理方具体 agent + suspend_task: 是否把任务置 waiting（角色 agent 提问时挂起任务等回答） + + 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 ""}] + + qid = getID() + ctx_json = json.dumps(context, ensure_ascii=False, default=str) if context else None + await sor.C('pipeline_agent_questions', { + 'id': qid, + 'tenant_id': tenant_id or '', + 'task_id': task_id or '', + 'from_role': from_role or '', + 'from_agentid': from_agentid or '', + 'problem_type': problem_type or '', + 'question': question or '', + 'context': ctx_json, + 'escalation_path': json.dumps(path, ensure_ascii=False), + 'escalation_pos': 0, + 'status': S_PENDING, + }) + if suspend_task and task_id: + # 挂起任务并清 claimed_by（否则 poller 要求 claimed_by IS NULL 才重新认领，任务僵尸） + 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) + return qid + + +async def resolve_problem(question_id, answer, answered_by='', answer_source='main_agent', + resume_task=True): + """当前处理方解决了问题 → 停止冒泡(status=answered)，任务恢复 submitted。""" + 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] + task_id = getattr(rec, 'task_id', '') or '' + + await sor.U('pipeline_agent_questions', { + 'id': question_id, + 'answer': answer or '', + 'answered_by': answered_by or '', + 'answer_source': answer_source or 'main_agent', + 'status': S_ANSWERED, + }) + + resumed = False + if resume_task and task_id: + trecs = await sor.R('pipeline_tasks', {'id': task_id}) + if trecs: + tstate = getattr(trecs[0], 'state', '') + if tstate == 'waiting': + await sor.sqlExe( + "UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL " + "WHERE id=${tid}$", {"tid": task_id}) + resumed = True + logger.info("resolve_problem: qid=%s task=%s by=%s resumed=%s", + question_id, task_id, answered_by, resumed) + return {'id': question_id, 'status': S_ANSWERED, 'resumed': resumed} + + +async def escalate_problem(question_id, by_role='', by_agentid='', note=''): + """当前处理方没解决 → 冒泡到路径下一个处理方（pos+1）。""" + 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, + }) + 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) + + +async def list_problems_for(role, agentid=None, tenant_id=None, task_id=None, limit=50) -> list: + """列出「当前该 (role, agentid) 处理」的 pending 问题。 + + 多 agent 感知：agentid 指定时，只返回"该 agent 处理"或"该角色任意 agent 处理" + 的问题；不指定 agentid 时返回该角色所有待处理问题。 + + 替代旧版「按 status 无差别列全部 pending」——把归属不同的问题混给同一个处理者。 + """ + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + conditions = ["status='pending'"] + params = {} + if tenant_id: + conditions.append("tenant_id=${tenant_id}$") + params["tenant_id"] = tenant_id + if task_id: + conditions.append("task_id=${task_id}$") + params["task_id"] = task_id + where = " AND ".join(conditions) + try: + limit = int(limit) + except (TypeError, ValueError): + limit = 50 + sql = f"SELECT * FROM pipeline_agent_questions WHERE {where} 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) + return result + + +def _rec_to_dict(rec): + if hasattr(rec, '__dict__'): + return {k: getattr(rec, k) for k in dir(rec) + if not k.startswith('_') and not callable(getattr(rec, k))} + return dict(rec) + + +async def get_task_qa(task_id, include_pending_for=None, + include_pending_agentid=None) -> dict: + """取任务的历史问答（已回答）+ 可选「待某处理方处理」的 pending 退回意见。 + + Args: + include_pending_for: 角色名，额外返回当前该角色应响应的 pending 问题 + include_pending_agentid: 具体 agent 标识（多 agent 场景） + + Returns: + {"answered": [...], "pending": [...]} + """ + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + recs = await sor.sqlExe( + "SELECT * FROM pipeline_agent_questions " + "WHERE task_id=${tid}$ AND status='answered' ORDER BY created_at ASC", + {"tid": task_id}) + answered = [] + for rec in (recs or []): + answered.append({ + "question": getattr(rec, 'question', '') or '', + "answer": getattr(rec, 'answer', '') or '', + "answer_source": getattr(rec, 'answer_source', '') or '', + "from_role": getattr(rec, 'from_role', '') or '', + }) + + 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, + }) + return {"answered": answered, "pending": pending} diff --git a/pipeline_service/init.py b/pipeline_service/init.py index 1d8964e..a273e85 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -37,6 +37,10 @@ from .questions import ( agent_ask, answer_question, forward_question, get_question, list_questions, get_task_qna, ) +from .communication import ( + raise_problem, resolve_problem, escalate_problem, + forward_to_customer, 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 @@ -539,6 +543,14 @@ def load_pipeline_service(): env.question_list = list_questions env.question_qna = get_task_qna + # 团队沟通（通用问题冒泡引擎，多 agent 感知） + 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 + # DevOps: git/shell operations (v3.2.1) env.shell_exec = shell_exec env.skill_import_git = skill_import_git diff --git a/pipeline_service/questions.py b/pipeline_service/questions.py index 1739351..67b4628 100644 --- a/pipeline_service/questions.py +++ b/pipeline_service/questions.py @@ -1,28 +1,28 @@ -"""问题回路 — 角色agent缺信息时提问，主agent路由回答。 +"""问题回路 — 向后兼容层，委托 communication.py 通用冒泡引擎。 -流程： -1. 角色agent执行任务时发现缺关键信息 → agent_ask() 写入 pipeline_agent_questions - (status=pending)，任务置 waiting，退出本轮执行。 -2. 主agent轮询 pending 问题： - - 能直接回答 → answer_question(answer_source='main_agent')，任务回 submitted， - 角色agent下轮重新认领并带着问答历史继续。 - - 答不了 → forward_question() 转客户 (status=forwarded)； - 客户在会话中回答后，主agent调 answer_question(answer_source='customer')。 +历史状态 pending/forwarded/answered 收敛为通用模型： +- 问题类型(pipeline_problem_types)定义冒泡路径(escalation_path) +- 问题沿路径冒泡：解决→answered(停)，没解决→pos+1(沿路径走)，尽头=人兜底 +- 归属不再靠 status 无差别路由，而靠 current_handler(escalation_path+pos) + +本文件保留旧签名供 DSPY(env.question_*) 与既有调用方使用，内部全部走 communication.py。 """ import json import logging from sqlor.dbpools import DBPools -from appPublic.uniqueID import getID + +from .communication import ( + raise_problem, + resolve_problem, + forward_to_customer, + list_problems_for, + get_task_qa, +) DBNAME = "pipeline" logger = logging.getLogger("pipeline.questions") -# 问题状态 -Q_PENDING = "pending" # 待主agent回答 -Q_FORWARDED = "forwarded" # 主agent已转客户，等客户回复 -Q_ANSWERED = "answered" # 已回答 - def _get_db(): db = DBPools() @@ -34,86 +34,27 @@ def _get_db(): return db, DBNAME -async def agent_ask(tenant_id: str, task_id: str, from_role: str, question: str, context: dict = None) -> str: - """角色agent提问。任务置 waiting，等待主agent/客户回答。返回 question_id。""" - db, dbname = _get_db() - async with db.sqlorContext(dbname) as sor: - qid = getID() - ctx_json = json.dumps(context, ensure_ascii=False, default=str) if context else None - await sor.C('pipeline_agent_questions', { - "id": qid, - "tenant_id": tenant_id or "", - "task_id": task_id or "", - "from_role": from_role or "", - "question": question or "", - "context": ctx_json, - "status": Q_PENDING, - }) - if task_id: - # 置 waiting 同时清 claimed_by，否则任务僵尸（poll器要求 claimed_by IS NULL 才会重新认领） - await sor.sqlExe( - "UPDATE pipeline_tasks SET state='waiting', claimed_by=NULL WHERE id=${tid}$", - {"tid": task_id}) - logger.info(f"agent_ask: task={task_id} role={from_role} qid={qid}") - return qid +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) async def answer_question(question_id: str, answer: str, answered_by: str = "", answer_source: str = "main_agent", resume: bool = True): - """回答问题。任务从 waiting 回 submitted，让角色agent重新认领继续执行。 - - Args: - answer_source: 'main_agent' 或 'customer' - resume: 是否把任务恢复为 submitted - - Returns: - 更新后的问题记录 dict；问题不存在返回 None。 - """ - 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] - task_id = rec.task_id if hasattr(rec, 'task_id') else rec['task_id'] - - await sor.U('pipeline_agent_questions', { - "id": question_id, - "answer": answer or "", - "answered_by": answered_by or "", - "answer_source": answer_source or "main_agent", - "status": Q_ANSWERED, - }) - - resumed = False - if resume and task_id: - trecs = await sor.R('pipeline_tasks', {'id': task_id}) - if trecs: - t = trecs[0] - tstate = t.state if hasattr(t, 'state') else t['state'] - if tstate == "waiting": - # 恢复 submitted 同时清 claimed_by，否则 poll 器永不重新认领该任务 - await sor.sqlExe( - "UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL WHERE id=${tid}$", - {"tid": task_id}) - resumed = True - - logger.info(f"answer_question: qid={question_id} task={task_id} source={answer_source} resumed={resumed}") - return { - "id": question_id, - "task_id": task_id, - "status": Q_ANSWERED, - "answer_source": answer_source, - "resumed": resumed, - } + """回答问题（解决 → 停止冒泡，任务恢复 submitted）。""" + return await resolve_problem(question_id, answer, answered_by, answer_source, + resume_task=resume) async def forward_question(question_id: str): - """主agent答不了 → 转给客户。status: pending → forwarded。""" - db, dbname = _get_db() - async with db.sqlorContext(dbname) as sor: - await sor.U('pipeline_agent_questions', {"id": question_id, "status": Q_FORWARDED}) - logger.info(f"forward_question: qid={question_id}") + """主 agent 答不了 → 沿路径冒泡给客户(人)。""" + return await forward_to_customer(question_id) async def get_question(question_id: str): @@ -139,58 +80,18 @@ 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: - """问题列表，支持租户/任务/状态过滤。""" - db, dbname = _get_db() - async with db.sqlorContext(dbname) as sor: - conditions = [] - params = {} - if tenant_id: - conditions.append("tenant_id=${tenant_id}$") - params["tenant_id"] = tenant_id - if task_id: - conditions.append("task_id=${task_id}$") - params["task_id"] = task_id - if status: - conditions.append("status=${status}$") - params["status"] = status - where = " AND ".join(conditions) if conditions else "1=1" - try: - limit = int(limit) - except (TypeError, ValueError): - limit = 50 - sql = f"SELECT * FROM pipeline_agent_questions WHERE {where} ORDER BY created_at DESC LIMIT {limit}" - recs = await sor.sqlExe(sql, params) - result = [] - for rec in (recs or []): - if hasattr(rec, '__dict__'): - d = {k: getattr(rec, k) for k in dir(rec) - if not k.startswith('_') and not callable(getattr(rec, k))} - else: - d = dict(rec) - ctx = d.get('context') - if ctx and isinstance(ctx, str): - try: - d['context'] = json.loads(ctx) - except (json.JSONDecodeError, TypeError): - pass - result.append(d) - return result + """客户视角：列出当前该「客户(人)」处理的问题。 + + 旧签名兼容 status 过滤；无 status 时列客户待处理(pending)。 + """ + rows = await list_problems_for("customer", tenant_id=tenant_id, + task_id=task_id, limit=limit) + if status: + rows = [r for r in rows if (r.get('status') or '') == status] + return rows async def get_task_qna(task_id: str) -> list: - """取任务的全部已回答问答（按时间序），用于角色agent继续执行时注入prompt。""" - db, dbname = _get_db() - async with db.sqlorContext(dbname) as sor: - recs = await sor.sqlExe( - "SELECT question, answer, answer_source, from_role FROM pipeline_agent_questions " - "WHERE task_id=${tid}$ AND status='answered' ORDER BY created_at ASC", - {"tid": task_id}) - result = [] - for rec in (recs or []): - result.append({ - "question": getattr(rec, 'question', '') or '', - "answer": getattr(rec, 'answer', '') or '', - "answer_source": getattr(rec, 'answer_source', '') or '', - "from_role": getattr(rec, 'from_role', '') or '', - }) - return result + """取任务的全部已回答问答（旧返回结构：list），供角色 agent 注入 prompt。""" + qa = await get_task_qa(task_id) + return qa.get('answered', []) if isinstance(qa, dict) else [] diff --git a/pipeline_service/sdlc_ability.py b/pipeline_service/sdlc_ability.py index bbebe01..7e6228f 100644 --- a/pipeline_service/sdlc_ability.py +++ b/pipeline_service/sdlc_ability.py @@ -75,16 +75,22 @@ SDL_TOOLS = [ ), ToolDefinition( name="list_questions", - description="查看待回答的问题", + description="查看待我(主agent)处理的问题", parameters={}, category="agent", ), ToolDefinition( name="answer_question", - description="回答agent提出的问题", + description="回答agent提出的问题（解决→停止冒泡）", parameters={"question_id": "问题ID", "answer": "回答内容"}, category="agent", ), + ToolDefinition( + name="escalate_question", + description="答不了的问题沿冒泡路径转给下一个处理方(通常转客户)", + parameters={"question_id": "问题ID"}, + category="agent", + ), ToolDefinition( name="add_repo", description="添加项目关联的Git仓库", @@ -129,7 +135,7 @@ SDL_PROMPT = """你是「开发产线」的驾驶舱 agent，负责软件项目 - 用户报告异常/故障 → 先用 diagnose_project 定位根因，再用 task_detail 查详情。 - 仓库管理 → add_repo / list_repos / clone_repo。 - 发现卡点（任务卡死、审核超时、失败、僵尸 claimed_by）时，主动定位根因并推动修复。 -- 发现「待回答问题」> 0 或角色提问时，加载 question-escalation 技能按问题升级链处理（列问题 → 能答则答 → 不能答抛给客户）。""" +- 发现「待回答问题」> 0 或角色提问时，加载 team-communication 技能按冒泡链处理（list_questions 列问题 → 能答则 answer_question → 不能答 escalate_question 沿冒泡路径转下一个处理方）。""" # ── 产线角色集（可插拔，替代硬编码 ROLE_SPECIFICS/ROLE_ALIASES/ROLE_CHAIN） ── @@ -337,10 +343,10 @@ async def _h_diagnose_project(sor, p, ctx): qc = 0 try: - questions = await sor.sqlExe( - "SELECT COUNT(*) as c FROM pipeline_agent_questions WHERE tenant_id=${pid}$ AND status='pending'", - {"pid": pid}) - qc = getattr(questions[0], "c", 0) if questions else 0 + from .communication import list_problems_for + # 待回答问题 = 当前该「主 agent」处理的问题（不混入 PM 退回给角色 agent 的） + mine = await list_problems_for("main_agent", tenant_id=pid, limit=200) + qc = len(mine) except Exception: pass @@ -424,27 +430,41 @@ async def _h_list_questions(sor, p, ctx): pid = ctx.get("project_id", "") if not pid: return "请先切换到项目" - try: - recs = await sor.sqlExe( - "SELECT id, question, answer_source FROM pipeline_agent_questions " - "WHERE tenant_id=${pid}$ AND status='pending' ORDER BY created_at DESC LIMIT 10", - {"pid": pid}) - except Exception: - try: - recs = await sor.sqlExe( - "SELECT id, question, answer_source FROM pipeline_agent_questions " - "WHERE tenant_id=${pid}$ ORDER BY created_at DESC LIMIT 10", - {"pid": pid}) - except Exception: - return "暂无问题数据" + from .communication import list_problems_for + # 只列「当前该主 agent 处理」的问题（不混入 PM 退回给角色 agent 的退回意见） + recs = await list_problems_for("main_agent", tenant_id=pid, limit=10) if not recs: return "没有待回答问题" lines = [] for r in recs: - lines.append(f"- [{getattr(r, 'id', '')}] {getattr(r, 'question', '')[:200]}") + qid = r.get('id', '') or '' + q = (r.get('question', '') or '')[:200] + ptype = r.get('problem_type', '') or '' + tag = f"[{ptype}] " if ptype else "" + lines.append(f"- [{qid}] {tag}{q}") return "\n".join(lines) +async def _h_escalate_question(sor, p, ctx): + qid = p.get("question_id", "") + if not qid: + return "需要问题ID" + from .communication import forward_to_customer + # 模糊匹配短 id + full_qid = qid + if len(qid) < 32: + recs = await sor.sqlExe( + "SELECT id FROM pipeline_agent_questions WHERE id LIKE ${prefix}$ LIMIT 1", + {"prefix": qid + "%"}) + if recs: + full_qid = getattr(recs[0], "id", qid) + r = await forward_to_customer(full_qid, by_role="main_agent") + if r is None: + return f"问题不存在: {qid}" + h = r.get("current_handler", {}) or {} + return f"OK: 已沿冒泡路径转给 {h.get('role', '?')}" + + async def _h_answer_question(sor, p, ctx): qid = p.get("question_id", "") answer = p.get("answer", "") @@ -594,6 +614,7 @@ SDL_HANDLERS = { "view_deliverable": _h_view_deliverable, "list_questions": _h_list_questions, "answer_question": _h_answer_question, + "escalate_question": _h_escalate_question, "add_repo": _h_add_repo, "list_repos": _h_list_repos, "clone_repo": _h_clone_repo,