b4e5e20ccf
sqlor 返回 DictObject，列名不暴露为实例属性(dir() 只列 dict 方法)， dir(rec)+getattr 会取到 None。改 dict(rec)，questions.get_question 同步修复。
101 lines
3.6 KiB
Python
101 lines
3.6 KiB
Python
"""问题回路 — 向后兼容层，委托 communication.py 通用冒泡引擎。
|
||
|
||
规范（团队角色/问题类型/冒泡路径）在 team-communication skill 里；
|
||
本模块只保留旧签名供 DSPY(env.question_*) 与既有调用方使用，内部走 communication.py。
|
||
|
||
状态只有 pending/answered；归属不再靠 status，而靠 current_handler_role/agentid 两列。
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
from sqlor.dbpools import DBPools
|
||
|
||
from .communication import (
|
||
raise_problem,
|
||
resolve_problem,
|
||
escalate_problem,
|
||
list_problems_for,
|
||
get_task_qa,
|
||
)
|
||
|
||
DBNAME = "pipeline"
|
||
logger = logging.getLogger("pipeline.questions")
|
||
|
||
|
||
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 agent_ask(tenant_id: str, task_id: str, from_role: str, question: str,
|
||
context: dict = None) -> str:
|
||
"""旧签名：按 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 = "",
|
||
answer_source: str = "main_agent", resume: bool = True):
|
||
"""回答问题（解决 → 停止冒泡，任务恢复 submitted）。"""
|
||
return await resolve_problem(question_id, answer, answered_by, answer_source,
|
||
resume_task=resume)
|
||
|
||
|
||
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):
|
||
"""取单条问题记录（含 context 解析）。"""
|
||
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]
|
||
try:
|
||
d = dict(rec)
|
||
except Exception:
|
||
d = {}
|
||
ctx = d.get('context')
|
||
if ctx and isinstance(ctx, str):
|
||
try:
|
||
d['context'] = json.loads(ctx)
|
||
except (json.JSONDecodeError, TypeError):
|
||
pass
|
||
return d
|
||
|
||
|
||
async def list_questions(tenant_id: str = None, task_id: str = None,
|
||
status: str = None, limit: int = 50) -> list:
|
||
"""客户视角：列出当前该「客户(人)」处理的问题。"""
|
||
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:
|
||
"""取任务的全部已回答问答（旧返回结构：list），供角色 agent 注入 prompt。"""
|
||
qa = await get_task_qa(task_id)
|
||
return qa.get('answered', []) if isinstance(qa, dict) else []
|