"""团队沟通引擎 — 通用问题冒泡（显式路由版）。 核心思想（规范在 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 import logging from sqlor.dbpools import DBPools from appPublic.uniqueID import getID from .audit import record_audit DBNAME = "pipeline" logger = logging.getLogger("pipeline.communication") S_PENDING = "pending" # 冒泡中（未解决） S_ANSWERED = "answered" # 已解决，停止冒泡 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 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: problem_type: 问题类型标识（skill 里定义的字符串，如 need_info/review_reject/fault_report） question: 问题内容 from_role: 谁提的（角色名） from_agentid: 提问的具体 agent 标识（多 agent 场景） first_handler_role: 首处理角色（不传默认 main_agent） first_handler_agentid: 首处理方具体 agent（空=该角色任意 agent） suspend_task: 是否把任务置 waiting（角色提问挂起等回答） Returns: question_id """ db, dbname = _get_db() # 归一化 question：QC 决策的 questions 可能是 list（多条意见数组），须 join 成字符串， # 否则 INSERT 报 OperationalError(1241, 'Operand should contain 1 column(s)')，改进意见丢失、 # develop 重做时读不到任何退回意见 → 反复产出空交付件。 if isinstance(question, (list, tuple)): question = "\n".join(str(x) for x in question) async with db.sqlorContext(dbname) as sor: hr = first_handler_role or "agent.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', { '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, 'current_handler_role': hr, 'current_handler_agentid': ha, '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}) await record_audit(tenant_id or '', 'pipeline_agent_questions', qid, 'raise', to_state=S_PENDING, who=from_role, agent_id=from_agentid, detail=problem_type, sor=sor) 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="agent.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 '' tenant_id = getattr(rec, 'tenant_id', '') or '' await sor.U('pipeline_agent_questions', { 'id': question_id, 'answer': answer or '', 'answered_by': answered_by or '', 'answer_source': answer_source or 'agent.main_agent', 'status': S_ANSWERED, }) await record_audit(tenant_id, 'pipeline_agent_questions', question_id, 'resolve', from_state=S_PENDING, to_state=S_ANSWERED, who=answered_by, sor=sor) 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, 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] tenant_id = getattr(rec, 'tenant_id', '') or '' old_role = getattr(rec, 'current_handler_role', '') or '' await sor.U('pipeline_agent_questions', { 'id': question_id, 'current_handler_role': next_handler_role or '', 'current_handler_agentid': next_handler_agentid or '', }) await record_audit(tenant_id, 'pipeline_agent_questions', question_id, 'escalate', from_state=old_role, to_state=next_handler_role, who=by_role, sor=sor) 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 问题（确定性 SQL 过滤）。 agentid 指定时，返回「该 agent 处理」或「该角色任意 agent 处理」的问题； 不指定 agentid 时返回该角色所有待处理问题。 """ db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: 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 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} " 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 []): result.append(_rec_to_dict(rec)) return result def _rec_to_dict(rec): """把 sqlor 记录对象转成 dict。 sqlor 返回 appPublic.dictObject.DictObject，列名不暴露为实例属性（dir() 只列 dict 方法）， 必须用 dict(rec) 取列，不能用 dir(rec) + getattr（会取到 None/方法）。 """ if isinstance(rec, dict): return dict(rec) try: return dict(rec) except (TypeError, ValueError): if hasattr(rec, 'to_dict'): try: return rec.to_dict() except Exception: pass return {} async def get_task_qa(task_id, role=None, agentid=None) -> dict: """取任务的历史问答(answered) + 可选「待该处理方」的 pending 退回意见。 Args: role: 若给角色名，pending 只返回当前该角色应响应的（按 current_handler 过滤） agentid: 具体 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 role: pending = await list_problems_for(role, agentid=agentid, task_id=task_id) return {"answered": answered, "pending": pending}