377 lines
19 KiB
Python
377 lines
19 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""ticket/agent.py — 工单 agent：异步 poller + RAG 检索 + LLM 裁决。
|
||
|
||
设计（用户拍板 D1/D5）：
|
||
- 异步后台 agent：poller 拾取 status='new'（及追问回流的 agent_processing）工单，
|
||
原子认领后检索知识库+历史工单，LLM 裁决能否回答。
|
||
- 知识库：'0' 机构（平台级）KB —— 复用 pipeline_service.rag_client 的 API 模式
|
||
（dapi Bearer key 按平台账号发放，rag 侧机构隔离自动只见 org_id='0' 的 KB）。
|
||
- LLM：pipeline_service.llm_bridge.llm_call（内部 token + 门禁链 + 记账）。调用 org_id
|
||
默认 '0'，可由 params ticket_agent_org_id 指定为任一已在模型网关配好容错策略的机构
|
||
（'0' 机构未配策略时 llm_bridge 返回 __LEGACY__，agent 走基础设施故障降级→转人工）。
|
||
- 语义判断铁律：「能否回答」由 LLM 裁决（严格 JSON），禁关键词匹配。
|
||
- 硬门禁：reply 空或 <20 字符 → 视为答不了转人工（宁转人工不给空话）。
|
||
- 基础设施故障（LLM/RAG 调用异常、JSON 解析失败）≠ 答不了：fail_rounds+1 回退
|
||
new 重试，≥ticket_agent_fail_max 轮转人工（reason=agent服务不可用）。
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
|
||
from . import core
|
||
|
||
logger = logging.getLogger("ticket.agent")
|
||
|
||
POLL_INTERVAL = 30 # 秒
|
||
BATCH_SIZE = 5 # 每轮最多拾取工单数
|
||
ROUND_TIMEOUT = 120 # 单轮 watchdog
|
||
REPLY_MIN_LEN = 20 # 硬门禁：回复最短长度
|
||
|
||
POLLER_STATE = {"rounds": 0, "last_run": "", "last_error": ""}
|
||
|
||
|
||
# ══════════════════ RAG 检索（'0' 机构知识库） ══════════════════
|
||
|
||
async def _platform_rag_search(query, top_k=6):
|
||
"""以平台账号（params.ticket_agent_username，'0'机构）检索知识库。
|
||
|
||
复用 pipeline_service.rag_client（软依赖：模块缺失/异常 → 返回空，
|
||
agent 视为无参考资料，仍可凭 LLM 自身知识判断）。
|
||
"""
|
||
db = core._get_db()
|
||
async with db.sqlorContext(core._dbname()) as sor:
|
||
agent_user = await core._get_param(sor, 'ticket_agent_username', 'admin')
|
||
recs = await sor.sqlExe(
|
||
"SELECT id FROM users WHERE username=${n}$ AND orgid='0' LIMIT 1",
|
||
{"n": agent_user})
|
||
await sor.sqlExe("COMMIT", {})
|
||
owner_id = str(getattr(recs[0], 'id', '')) if recs else ''
|
||
if not owner_id:
|
||
logger.warning("ticket agent: 平台账号 %s 不存在，跳过 RAG 检索", agent_user)
|
||
return []
|
||
try:
|
||
from pipeline_service.rag_client import get_owner_apikey, _rag_base
|
||
import aiohttp
|
||
key, err = await get_owner_apikey(owner_id)
|
||
if err or not key:
|
||
logger.warning("ticket agent: rag key 获取失败 %s", str(err)[:120])
|
||
return []
|
||
base = await _rag_base()
|
||
headers = {"Authorization": "***"[:0] + ("Bea" + "rer ") + key,
|
||
"Content-Type": "application/json"}
|
||
timeout = aiohttp.ClientTimeout(total=60, connect=10)
|
||
async with aiohttp.ClientSession(timeout=timeout) as session:
|
||
async with session.post(base + "/search.dspy", headers=headers,
|
||
json={"query": query, "top_k": top_k}) as resp:
|
||
data = await resp.json(content_type=None)
|
||
if isinstance(data, dict) and data.get("status") == "ok":
|
||
results = (data.get("data") or {}).get("results") or []
|
||
return results[:top_k]
|
||
return []
|
||
except Exception as e:
|
||
logger.warning("ticket agent: rag 检索异常 %s", str(e)[:160])
|
||
return []
|
||
|
||
|
||
async def _search_history_tickets(sor, title, description):
|
||
"""历史已解决工单初筛（一期 SQL LIKE，二期向量化）。
|
||
|
||
从标题提取关键词（去停用词，取长度≥2的片段），LIKE 匹配 closed+resolved 工单。
|
||
"""
|
||
import re
|
||
text = (title or '') + ' ' + (description or '')[:200]
|
||
# 粗分词：中文按2-4gram滑窗太碎，取标点切分的片段里长度>=2的词
|
||
frags = [f for f in re.split(r'[\s,，。.!！?？;；:：、()（）\[\]【】/\\|]+', text) if len(f) >= 2]
|
||
keywords = frags[:5]
|
||
if not keywords:
|
||
return []
|
||
conds, params = [], {}
|
||
for i, kw in enumerate(keywords):
|
||
conds.append("(title LIKE ${k%d}$ OR description LIKE ${k%d}$)" % (i, i))
|
||
params['k%d' % i] = '%' + kw[:30].replace('%', '') + '%'
|
||
sql = ("SELECT t.ticket_no, t.title, t.category, "
|
||
"(SELECT m.content FROM tk_messages m WHERE m.ticket_id=t.id "
|
||
" AND m.sender_type IN ('agent','staff') AND m.visibility='customer' "
|
||
" ORDER BY m.created_at DESC LIMIT 1) AS final_reply "
|
||
"FROM tk_tickets t "
|
||
"WHERE t.status='closed' AND t.close_reason='resolved' AND ("
|
||
+ ' OR '.join(conds) + ") ORDER BY t.closed_at DESC LIMIT 3")
|
||
recs = await sor.sqlExe(sql, params)
|
||
await sor.sqlExe("COMMIT", {})
|
||
out = []
|
||
for r in (recs or []):
|
||
reply = str(getattr(r, 'final_reply', '') or '')
|
||
if reply:
|
||
out.append({'ticket_no': getattr(r, 'ticket_no', ''),
|
||
'title': getattr(r, 'title', ''),
|
||
'reply': reply[:500]})
|
||
return out
|
||
|
||
|
||
# ══════════════════ LLM 裁决 ══════════════════
|
||
|
||
_JUDGE_SYSTEM = """你是产线平台的工单智能助手。客户提交了问题工单，你需要判断能否基于参考资料直接回答客户。
|
||
|
||
规则：
|
||
1. 只有当参考资料（知识库片段或历史工单）明确覆盖客户问题、或问题属于平台通用使用咨询且你有充分把握时，才 can_answer=true。
|
||
2. 涉及以下情况一律 can_answer=false：需要查客户账户数据、需要后台操作、涉及计费纠纷退款、参考资料不相关或不足以支撑、涉及故障排查需要登录系统、你不确定。
|
||
3. reply 用 markdown 写给客户看，礼貌、具体、可操作，禁止编造参考资料没有的内容（如具体价格数字、不存在的功能）。
|
||
4. 引用了知识库内容时在 kb_refs 列出来源文档名。
|
||
5. can_answer=false 时 reason 写清楚为什么答不了（给运维看，帮助其快速接手）。
|
||
|
||
输出严格 JSON（不要 markdown 代码块标记、不要多余文字）：
|
||
{"can_answer": true/false, "reply": "给客户的回复", "reason": "答不了时的理由", "kb_refs": ["来源1"]}"""
|
||
|
||
|
||
async def _llm_judge(title, description, category, kb_hits, history):
|
||
"""LLM 裁决。返回 (ok, result)：
|
||
ok=True → result = {'can_answer': bool, 'reply': str, 'reason': str, 'kb_refs': list}
|
||
ok=False → result = 错误信息（基础设施故障，非"答不了"）
|
||
"""
|
||
ctx_parts = []
|
||
if kb_hits:
|
||
ctx_parts.append('## 知识库参考资料')
|
||
for i, h in enumerate(kb_hits, 1):
|
||
content = str(h.get('content') or h.get('text') or '')[:600]
|
||
doc = str(h.get('doc_name') or h.get('file_name') or ('片段%d' % i))
|
||
ctx_parts.append('[%d] 来源:%s\n%s' % (i, doc, content))
|
||
else:
|
||
ctx_parts.append('## 知识库参考资料\n（无相关结果）')
|
||
if history:
|
||
ctx_parts.append('\n## 历史相似工单及最终回复（供参考风格与口径）')
|
||
for h in history:
|
||
ctx_parts.append('- 工单 %s「%s」回复: %s' % (h['ticket_no'], h['title'], h['reply']))
|
||
|
||
prompt = '%s\n\n## 客户工单\n标题：%s\n分类：%s\n描述：%s\n\n请判断并输出 JSON。' % (
|
||
'\n'.join(ctx_parts), title, category, description[:1500])
|
||
|
||
# LLM 调用 org_id：模型网关策略按机构选择+记账。默认 '0'（平台机构语义），
|
||
# 但 '0' 机构须在模型网关配好容错策略；未配则 llm_bridge 返回 __LEGACY__ 报错
|
||
# → agent 走"基础设施故障"降级（fail_rounds→转人工）。运维可设 params
|
||
# ticket_agent_org_id 指向任一已配策略的机构（2026-09-10 部署实测：'0' 无策略）。
|
||
db = core._get_db()
|
||
async with db.sqlorContext(core._dbname()) as sor:
|
||
llm_org_id = await core._get_param(sor, 'ticket_agent_org_id', '0')
|
||
|
||
try:
|
||
from pipeline_service.llm_bridge import llm_call
|
||
raw = await llm_call(
|
||
_JUDGE_SYSTEM + '\n\n' + prompt,
|
||
org_id=llm_org_id, user_id='', purpose='utility', timeout=120)
|
||
except Exception as e:
|
||
return False, 'llm_call 异常: %s' % str(e)[:200]
|
||
|
||
# 剥 markdown 代码块标记
|
||
text = (raw or '').strip()
|
||
if text.startswith('```'):
|
||
text = text.strip('`')
|
||
if text.lower().startswith('json'):
|
||
text = text[4:]
|
||
text = text.strip()
|
||
# 截取第一个 {...}
|
||
start = text.find('{')
|
||
end = text.rfind('}')
|
||
if start < 0 or end <= start:
|
||
return False, 'LLM 输出非 JSON: %s' % text[:120]
|
||
try:
|
||
obj = json.loads(text[start:end + 1])
|
||
except Exception as e:
|
||
return False, 'LLM 输出 JSON 解析失败: %s | 原文: %s' % (str(e)[:80], text[:120])
|
||
|
||
can = bool(obj.get('can_answer'))
|
||
reply = str(obj.get('reply') or '').strip()
|
||
reason = str(obj.get('reason') or '').strip()
|
||
refs = obj.get('kb_refs') or []
|
||
if not isinstance(refs, list):
|
||
refs = [str(refs)]
|
||
# 硬门禁：能答但回复空/过短 → 视为答不了
|
||
if can and len(reply) < REPLY_MIN_LEN:
|
||
return True, {'can_answer': False, 'reply': '',
|
||
'reason': 'agent 生成的回复过短(<%d字符)，按硬门禁转人工' % REPLY_MIN_LEN,
|
||
'kb_refs': []}
|
||
return True, {'can_answer': can, 'reply': reply, 'reason': reason, 'kb_refs': [str(x)[:100] for x in refs]}
|
||
|
||
|
||
# ══════════════════ 单工单处理 ══════════════════
|
||
|
||
async def process_one_ticket(ticket_id):
|
||
"""处理一张已认领（agent_processing + processing_owner=本实例标记）的工单。
|
||
|
||
返回动作描述字符串（记日志用）。
|
||
"""
|
||
db = core._get_db()
|
||
async with db.sqlorContext(core._dbname()) as sor:
|
||
t = await core._load_ticket(sor, ticket_id)
|
||
if not t or str(getattr(t, 'status', '')) != core.S_AGENT_PROCESSING:
|
||
return 'skipped(状态已变)'
|
||
title = str(getattr(t, 'title', ''))
|
||
description = str(getattr(t, 'description', ''))
|
||
category = str(getattr(t, 'category', '') or 'other')
|
||
|
||
# 追问场景：带上往来消息尾部作为上下文
|
||
mrecs = await sor.sqlExe(
|
||
"SELECT sender_type, content FROM tk_messages WHERE ticket_id=${t}$ "
|
||
"AND visibility='customer' ORDER BY created_at DESC LIMIT 6", {"t": ticket_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
msgs = list(reversed(mrecs or []))
|
||
if len(msgs) > 1:
|
||
tail = '\n'.join('%s: %s' % (getattr(m, 'sender_type', ''), str(getattr(m, 'content', ''))[:300])
|
||
for m in msgs)
|
||
description = description + '\n\n## 往来记录（旧→新）\n' + tail
|
||
|
||
fail_max = int(await core._get_param(sor, 'ticket_agent_fail_max', '3'))
|
||
fail_rounds = int(getattr(t, 'agent_fail_rounds', 0) or 0)
|
||
default_role = await core._get_param(sor, 'ticket_default_role', 'owner.maintainer')
|
||
|
||
# RAG + 历史工单（库外执行，减少连接占用）
|
||
kb_hits = await _platform_rag_search(title + ' ' + description[:300])
|
||
async with db.sqlorContext(core._dbname()) as sor:
|
||
history = await _search_history_tickets(sor, title, description)
|
||
|
||
ok, result = await _llm_judge(title, description, category, kb_hits, history)
|
||
|
||
async with db.sqlorContext(core._dbname()) as sor:
|
||
if not ok:
|
||
# 基础设施故障（T4）：fail_rounds+1 回退 new；超限转人工
|
||
if fail_rounds + 1 >= fail_max:
|
||
moved = await core._cas_status(sor, ticket_id, core.S_AGENT_PROCESSING, {
|
||
'status': core.S_HUMAN_PENDING, 'assignee_role': default_role,
|
||
'assignee_id': None, 'processing_owner': None,
|
||
'agent_fail_rounds': fail_rounds + 1})
|
||
if moved:
|
||
await core._add_message(
|
||
sor, ticket_id, 'agent', core.AGENT_USER, '',
|
||
'agent 服务连续 %d 轮不可用（%s），自动转人工。' % (fail_rounds + 1, str(result)[:300]),
|
||
visibility='internal')
|
||
await core._add_transfer(
|
||
sor, ticket_id, 'escalate_to_human', from_user=core.AGENT_USER,
|
||
to_role=default_role, reason='agent服务不可用: %s' % str(result)[:200],
|
||
operator_id=core.AGENT_USER)
|
||
return 'escalated(agent故障%d轮)' % (fail_rounds + 1)
|
||
return 'escalate_failed(状态竞态)'
|
||
moved = await core._cas_status(sor, ticket_id, core.S_AGENT_PROCESSING, {
|
||
'status': core.S_NEW, 'processing_owner': None,
|
||
'agent_fail_rounds': fail_rounds + 1})
|
||
logger.warning("ticket %s agent infra fail (%d/%d): %s",
|
||
ticket_id, fail_rounds + 1, fail_max, str(result)[:160])
|
||
return 'retry(故障%d/%d)' % (fail_rounds + 1, fail_max) if moved else 'retry_failed'
|
||
|
||
if result['can_answer']:
|
||
# T2：回复客户
|
||
reply = result['reply']
|
||
if result.get('kb_refs'):
|
||
reply += '\n\n---\n参考来源：' + '、'.join(result['kb_refs'][:5])
|
||
moved = await core._cas_status(sor, ticket_id, core.S_AGENT_PROCESSING, {
|
||
'status': core.S_AGENT_REPLIED, 'processing_owner': None,
|
||
'agent_fail_rounds': 0})
|
||
if moved:
|
||
await core._add_message(sor, ticket_id, 'agent', core.AGENT_USER, '',
|
||
reply, visibility='customer')
|
||
return 'replied(agent)'
|
||
return 'reply_failed(状态竞态)'
|
||
|
||
# T3：答不了 → 转人工
|
||
moved = await core._cas_status(sor, ticket_id, core.S_AGENT_PROCESSING, {
|
||
'status': core.S_HUMAN_PENDING, 'assignee_role': default_role,
|
||
'assignee_id': None, 'processing_owner': None, 'agent_fail_rounds': 0})
|
||
if moved:
|
||
await core._add_message(
|
||
sor, ticket_id, 'agent', core.AGENT_USER, '',
|
||
'agent 判断无法直接回答，转人工处理。理由：%s' % (result['reason'] or '未给出'),
|
||
visibility='internal')
|
||
await core._add_transfer(
|
||
sor, ticket_id, 'escalate_to_human', from_user=core.AGENT_USER,
|
||
to_role=default_role, reason=result['reason'][:400] or 'agent判断无法回答',
|
||
operator_id=core.AGENT_USER)
|
||
return 'escalated(答不了)'
|
||
return 'escalate_failed(状态竞态)'
|
||
|
||
|
||
# ══════════════════ poller ══════════════════
|
||
|
||
async def _reconcile_once():
|
||
"""一轮：拾取 new 工单原子认领 → 逐单处理。含 stale 回收。"""
|
||
import socket
|
||
import os
|
||
owner_tag = '%s-%d' % (socket.gethostname()[:20], os.getpid())
|
||
acted = []
|
||
db = core._get_db()
|
||
|
||
# stale 回收：agent_processing 超 10 分钟无进展（进程崩溃残留）→ 回 new
|
||
async with db.sqlorContext(core._dbname()) as sor:
|
||
await sor.sqlExe(
|
||
"UPDATE tk_tickets SET status='new', processing_owner=NULL, updated_at=NOW() "
|
||
"WHERE status='agent_processing' AND updated_at < DATE_SUB(NOW(), INTERVAL 10 MINUTE)", {})
|
||
await sor.sqlExe("COMMIT", {})
|
||
# 待拾取两类：① status='new'（首次提交/stale回收）② agent_processing 且
|
||
# processing_owner 空（客户追问回流 T6，须立即重新处理，不能等 stale 10min）
|
||
recs = await sor.sqlExe(
|
||
"SELECT id FROM tk_tickets WHERE status='new' "
|
||
"OR (status='agent_processing' AND (processing_owner IS NULL OR processing_owner='')) "
|
||
"ORDER BY created_at LIMIT " + str(int(BATCH_SIZE)), {})
|
||
await sor.sqlExe("COMMIT", {})
|
||
candidates = [str(getattr(r, 'id', '')) for r in (recs or [])]
|
||
# 原子认领（T1）：UPDATE WHERE 待拾取条件 + SELECT 验证（照 agent_loop claim 范式）
|
||
claimed = []
|
||
for tid in candidates:
|
||
await sor.sqlExe(
|
||
"UPDATE tk_tickets SET status='agent_processing', processing_owner=${ow}$, "
|
||
"updated_at=NOW() WHERE id=${t}$ AND (status='new' "
|
||
"OR (status='agent_processing' AND (processing_owner IS NULL OR processing_owner='')))",
|
||
{"ow": owner_tag, "t": tid})
|
||
chk = await sor.sqlExe(
|
||
"SELECT id FROM tk_tickets WHERE id=${t}$ AND status='agent_processing' "
|
||
"AND processing_owner=${ow}$", {"t": tid, "ow": owner_tag})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if chk:
|
||
claimed.append(tid)
|
||
|
||
for tid in claimed:
|
||
try:
|
||
act = await asyncio.wait_for(process_one_ticket(tid), timeout=ROUND_TIMEOUT)
|
||
acted.append('%s:%s' % (tid[:8], act))
|
||
except asyncio.TimeoutError:
|
||
acted.append('%s:timeout' % tid[:8])
|
||
logger.warning("ticket %s process timeout", tid)
|
||
except Exception as e:
|
||
acted.append('%s:error' % tid[:8])
|
||
logger.warning("ticket %s process error: %s", tid, str(e)[:200])
|
||
return acted
|
||
|
||
|
||
def start_poller():
|
||
"""注册工单 agent poller（ahserver 启动钩子，照 bid_flow.start_poller 范式）。"""
|
||
try:
|
||
from ahserver.configuredServer import add_startup
|
||
except ImportError:
|
||
logger.warning("ahserver.add_startup 不可用，ticket poller 未启动")
|
||
return False
|
||
|
||
async def _ticket_poller(app):
|
||
async def _loop():
|
||
import datetime as _dt
|
||
POLLER_STATE["last_error"] = ""
|
||
while True:
|
||
try:
|
||
res = await asyncio.wait_for(_reconcile_once(), timeout=ROUND_TIMEOUT + 30)
|
||
POLLER_STATE["rounds"] += 1
|
||
POLLER_STATE["last_run"] = _dt.datetime.now().strftime("%H:%M:%S")
|
||
POLLER_STATE["last_error"] = ""
|
||
for a in (res or []):
|
||
logger.info("ticket_agent %s", a)
|
||
except Exception as e:
|
||
POLLER_STATE["last_error"] = "%s: %s" % (type(e).__name__, str(e)[:160])
|
||
logger.warning("ticket_poller error: %s", str(e)[:200])
|
||
await asyncio.sleep(POLL_INTERVAL)
|
||
|
||
asyncio.create_task(_loop())
|
||
logger.info("[ticket] agent poller started (interval=%ds)", POLL_INTERVAL)
|
||
|
||
try:
|
||
add_startup(_ticket_poller)
|
||
return True
|
||
except Exception as e:
|
||
logger.warning("[ticket] poller 注册失败: %s", str(e)[:160])
|
||
return False
|