297 lines
14 KiB
Python
297 lines
14 KiB
Python
"""团队沟通引擎 — 通用问题冒泡（显式路由版）。
|
||
|
||
核心思想（规范在 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, form_schema=None) -> 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（角色提问挂起等回答）
|
||
form_schema: 动态表单声明（可选）。{"fields":[{"name","type":"file|text|textarea",
|
||
"label","target","required","accept"}]}——待办详情按它渲染输入界面，
|
||
人类在待办里直接上传文件/填写信息完成任务（文件落项目工作空间，
|
||
路径随回答回流给 agent）。存进 context.form_schema。
|
||
|
||
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:
|
||
# 已删除项目门禁（2026-09-15）：项目行已不存在的冒泡问题没有处理载体
|
||
# （待办详情/文件链接/任务恢复全悬空），且孤儿任务会反复触发同款问题
|
||
# 灌给 owner.superuser（实测 38 条堆积）——直接拒发并告知调用方。
|
||
# 全局问题（tenant_id 空或 '0'）不受影响。暂停项目不在此拦：failed 任务
|
||
# 报故障的流程是「先 pause 再 raise」，拦 paused 会把故障通知吞掉。
|
||
_tid = (tenant_id or '').strip()
|
||
if _tid and _tid != '0':
|
||
_prec = await sor.sqlExe(
|
||
"SELECT id FROM sd_projects WHERE id=${pid}$", {"pid": _tid})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not _prec:
|
||
logger.info("raise_problem skipped: project %s deleted (type=%s task=%s)",
|
||
_tid, problem_type, task_id)
|
||
return ""
|
||
hr = first_handler_role or "agent.main_agent"
|
||
ha = first_handler_agentid or ""
|
||
# 重复冒泡去重（2026-09-15 用户指令「避免重复」）：同任务+同问题类型已有
|
||
# pending 问题 → 不新建，把正文/context 更新到既有行并返回既有 qid。
|
||
# 实测历史数据：同 task_id 的 qc_reject/review_reject/fault_report 各堆 2~3 条
|
||
# pending（每轮退回都新发一条，旧的不答结），角标虚高、用户重复看到同一事项。
|
||
_tk = (task_id or '').strip()
|
||
if _tk and problem_type:
|
||
_dup = await sor.sqlExe(
|
||
"SELECT id FROM pipeline_agent_questions WHERE task_id=${tk}$ "
|
||
"AND problem_type=${pt}$ AND status='pending' "
|
||
"ORDER BY created_at DESC LIMIT 1",
|
||
{"tk": _tk, "pt": problem_type})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if _dup:
|
||
old_qid = getattr(_dup[0], 'id', '')
|
||
_ctx2 = dict(context or {})
|
||
if form_schema:
|
||
_ctx2['form_schema'] = form_schema
|
||
await sor.sqlExe(
|
||
"UPDATE pipeline_agent_questions SET question=${q}$, context=${c}$, "
|
||
"updated_at=NOW() WHERE id=${i}$",
|
||
{"q": question or '', "c": json.dumps(_ctx2, ensure_ascii=False, default=str)
|
||
if _ctx2 else None, "i": old_qid})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if suspend_task:
|
||
await sor.sqlExe(
|
||
"UPDATE pipeline_tasks SET state='waiting', claimed_by=NULL WHERE id=${tid}$",
|
||
{"tid": _tk})
|
||
await sor.sqlExe("COMMIT", {})
|
||
logger.info("raise_problem dedup: task=%s type=%s reuse qid=%s",
|
||
_tk, problem_type, old_qid)
|
||
return old_qid
|
||
qid = getID()
|
||
# form_schema 并入 context 存储（待办详情按它渲染动态表单）
|
||
if form_schema:
|
||
context = dict(context or {})
|
||
context['form_schema'] = form_schema
|
||
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}
|