288 lines
12 KiB
Python
288 lines
12 KiB
Python
"""工具失败聚集巡检:同签名的工具报错反复出现 → 主动冒头。
|
|
|
|
背景(2026-07,task 9dcae061 终案的结构性教训):mermaid 渲染在生产挂了 90 天
|
|
0 成功(67 次超时 + 26 次 launch fail、烧掉数十万 token),没有任何机制发现,
|
|
靠人工扫 DB 才挖出来。本模块把「失败聚集」变成信号:扫 messages 里 role=tool
|
|
的错误结果,按 (工具名 + 归一化错误签名) 聚合,超阈值即算聚集。
|
|
|
|
纯只读查询、无新表无状态;告警通道由调用方决定(web/app.py 的巡检 loop 发
|
|
开发者邮箱,admin API 直接返给前端表格)。
|
|
|
|
第二数据源(0.58.19):被丢弃的畸形 tool_call 参数(kind=malformed)——这类失败
|
|
整轮不入 messages(防投毒级联),loop 落 usage_events(kind=tool_malformed),
|
|
在此并入同一聚合口径(signature=归一化 JSON 报错,sample=损坏参数首尾片段)。
|
|
|
|
第三数据源(0.58.21):run 级终态错误(kind=run)—— LLM 请求层/构建期直接抛异常
|
|
(RateLimitError 余额不足、认证失败等),整轮无 tool 消息,_run_agent_bg 落
|
|
usage_events(kind=run_error),在此按归一化错误签名聚合(tool 名固定 "(run)")。
|
|
其中 provider 级致命错误(余额/认证)另走 `alert_provider_critical` 即时邮件,
|
|
不等日巡检 —— 这类错误会让该 provider 上所有用户的所有 run 全挂。
|
|
|
|
失败判定(tool content 的三类标记,形态见 executor_docker/_host):
|
|
- `[Error` 开头 —— 执行器/工具层报错([Error]、[Error executing ...])
|
|
- `command timed out` —— shell/run_python 超时
|
|
- 尾部 `[exit N]` 且 N != 0 —— shell 非零退出
|
|
`[exit 0]` 但语义失败(如 "No mermaid charts found")不判 —— 无通用判据,不猜。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import re
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from sqlalchemy import text
|
|
|
|
from core.storage import session_scope
|
|
|
|
# 签名归一:同一类错误在不同 task/参数下的差异(路径/数字/uuid/十六进制)抹平,
|
|
# 让 "figures/a.png doesn't exist" 和 "figures/b.png doesn't exist" 聚成一条。
|
|
_RE_UUID = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}")
|
|
_RE_HEX = re.compile(r"0x[0-9a-fA-F]+")
|
|
_RE_PATH = re.compile(r"(?:[A-Za-z]:)?(?:[/\\][\w.\-一-鿿*]+){2,}")
|
|
_RE_NUM = re.compile(r"\d+")
|
|
_RE_WS = re.compile(r"\s+")
|
|
|
|
_EXIT_TAIL = re.compile(r"\[exit (\d+)\]\s*$")
|
|
_STREAM_MARKS = ("[stdout]", "[stderr]")
|
|
|
|
|
|
def _normalize(s: str) -> str:
|
|
s = _RE_UUID.sub("<id>", s)
|
|
s = _RE_HEX.sub("<hex>", s)
|
|
s = _RE_PATH.sub("<path>", s)
|
|
s = _RE_NUM.sub("N", s)
|
|
s = _RE_WS.sub(" ", s).strip()
|
|
return s[:120]
|
|
|
|
|
|
def _classify(content: str) -> Optional[Tuple[str, str]]:
|
|
"""返回 (kind, 原始签名行) 或 None(不算失败)。"""
|
|
head = content.lstrip()
|
|
# 超时判定在前:超时结果形如 "[Error] command timed out after 30s",
|
|
# 让它归 timeout 而不是被 [Error 前缀截走(kind 对排查方向有指示意义)
|
|
if "command timed out" in content:
|
|
return "timeout", "command timed out"
|
|
if head.startswith("[Error"):
|
|
return "error", head.splitlines()[0]
|
|
m = _EXIT_TAIL.search(content)
|
|
if m and m.group(1) != "0":
|
|
# 签名取 [exit N] 前最后一行有实际内容的输出(通常是真正的报错行)
|
|
lines = [
|
|
ln.strip() for ln in content.splitlines()[:-1]
|
|
if ln.strip() and ln.strip() not in _STREAM_MARKS
|
|
]
|
|
return "exit", (lines[-1] if lines else f"exit {m.group(1)}")
|
|
return None
|
|
|
|
|
|
def scan_tool_failures(
|
|
days: float = 7,
|
|
min_count: int = 5,
|
|
min_tasks: int = 2,
|
|
) -> List[Dict[str, Any]]:
|
|
"""扫近 `days` 天的 tool 错误消息,返回超阈值的聚集。
|
|
|
|
阈值语义:同签名 >= min_count 次 且 跨 >= min_tasks 个 task —— 单 task 内
|
|
模型试错几次就自愈的正常噪音不触发;跨 task 复现的才是平台性问题。
|
|
|
|
时间分布:每个聚集带 `daily`(从 now 往回按 24h 分桶的次数,旧→新,
|
|
非日历日)和 `count_24h`(= daily 尾桶)—— 修复部署后看尾桶是否归零,
|
|
区分「还在发生」和「窗口内的存量记录」。排序:近 24h 活跃的在前
|
|
(count_24h 降序),其后按 count 降序。
|
|
同步阻塞(DB 查询),asyncio 调用方放 to_thread/executor。
|
|
"""
|
|
now = datetime.now(timezone.utc)
|
|
cutoff = now - timedelta(days=days)
|
|
n_buckets = max(1, math.ceil(days))
|
|
with session_scope() as s:
|
|
rows = s.execute(
|
|
text(
|
|
"select m.task_id, t.user_id, m.created_at, "
|
|
" m.payload->>'name' as tool_name, "
|
|
" m.payload->>'content' as content "
|
|
"from messages m join tasks t on t.task_id = m.task_id "
|
|
"where m.created_at >= :cutoff "
|
|
" and m.payload->>'role' = 'tool' "
|
|
" and (m.payload->>'content' like '[Error%' "
|
|
" or m.payload->>'content' like '%command timed out%' "
|
|
" or m.payload->>'content' like '%[exit %')"
|
|
),
|
|
{"cutoff": cutoff},
|
|
).fetchall()
|
|
# 第二段:被丢弃的畸形 tool_call 参数(kind=tool_malformed)。这类失败整轮
|
|
# 不入 messages(防投毒),loop._log_malformed_args 落在 usage_events,
|
|
# 是它们进面板/巡检邮件的唯一路径。
|
|
mrows = s.execute(
|
|
text(
|
|
"select task_id, user_id, created_at, "
|
|
" units->>'tool' as tool_name, "
|
|
" units->>'err' as err, "
|
|
" units->>'head' as head, "
|
|
" units->>'tail' as tail "
|
|
"from usage_events "
|
|
"where kind = 'tool_malformed' and created_at >= :cutoff"
|
|
),
|
|
{"cutoff": cutoff},
|
|
).fetchall()
|
|
# 第三段:run 级终态错误(kind=run)。LLM 层抛异常时整轮无 tool 消息,
|
|
# tasks.run_error 只留最后一次,usage_events(kind=run_error)才是完整留痕。
|
|
rrows = s.execute(
|
|
text(
|
|
"select task_id, user_id, created_at, "
|
|
" units->>'err' as err "
|
|
"from usage_events "
|
|
"where kind = 'run_error' and created_at >= :cutoff"
|
|
),
|
|
{"cutoff": cutoff},
|
|
).fetchall()
|
|
|
|
agg: Dict[Tuple[str, str], Dict[str, Any]] = {}
|
|
|
|
def _add(
|
|
tool_name: str, kind: str, sig_line: str, sample: str,
|
|
task_id: Any, user_id: Any, created_at: datetime,
|
|
) -> None:
|
|
# DB 列若是 naive timestamp(存 UTC),补 tzinfo 才能和 now 做减法
|
|
ts = created_at if created_at.tzinfo else created_at.replace(tzinfo=timezone.utc)
|
|
key = (tool_name or "?", _normalize(sig_line))
|
|
c = agg.get(key)
|
|
if c is None:
|
|
c = agg[key] = {
|
|
"tool": key[0],
|
|
"signature": key[1],
|
|
"kind": kind,
|
|
"count": 0,
|
|
"tasks": set(),
|
|
"users": set(),
|
|
"first_at": ts,
|
|
"last_at": ts,
|
|
"sample": sample[:300],
|
|
"daily": [0] * n_buckets,
|
|
}
|
|
c["count"] += 1
|
|
c["tasks"].add(task_id)
|
|
c["users"].add(user_id)
|
|
# 分桶:距 now 每满 24h 退一桶,尾桶 = 近 24h(时钟漂移/边界值 clamp 进首尾桶)
|
|
age_days = int((now - ts).total_seconds() // 86400)
|
|
c["daily"][n_buckets - 1 - min(n_buckets - 1, max(0, age_days))] += 1
|
|
if ts < c["first_at"]:
|
|
c["first_at"] = ts
|
|
if ts > c["last_at"]:
|
|
c["last_at"] = ts
|
|
c["sample"] = sample[:300]
|
|
|
|
for task_id, user_id, created_at, tool_name, content in rows:
|
|
if not content:
|
|
continue
|
|
hit = _classify(content)
|
|
if hit is None:
|
|
continue
|
|
kind, sig_line = hit
|
|
_add(tool_name, kind, sig_line, content, task_id, user_id, created_at)
|
|
|
|
for task_id, user_id, created_at, tool_name, err, head, tail in mrows:
|
|
_add(
|
|
tool_name, "malformed", err or "?",
|
|
f"{head or ''} … {tail or ''}",
|
|
task_id, user_id, created_at,
|
|
)
|
|
|
|
for task_id, user_id, created_at, err in rrows:
|
|
_add("(run)", "run", err or "?", err or "", task_id, user_id, created_at)
|
|
|
|
out = []
|
|
for c in agg.values():
|
|
if c["count"] < min_count or len(c["tasks"]) < min_tasks:
|
|
continue
|
|
out.append({
|
|
"tool": c["tool"],
|
|
"signature": c["signature"],
|
|
"kind": c["kind"],
|
|
"count": c["count"],
|
|
"count_24h": c["daily"][-1],
|
|
"daily": c["daily"],
|
|
"task_count": len(c["tasks"]),
|
|
"user_count": len(c["users"]),
|
|
"first_at": c["first_at"].isoformat(),
|
|
"last_at": c["last_at"].isoformat(),
|
|
"sample": c["sample"],
|
|
})
|
|
# 活跃的(近 24h 还在发生)排前面,已安静的沉底 —— 面板/邮件都先看还在烧的
|
|
out.sort(key=lambda x: (x["count_24h"], x["count"]), reverse=True)
|
|
return out
|
|
|
|
|
|
# ── provider 级致命错误即时告警 ──
|
|
# 命中判据:错误文案含余额/配额/认证类关键词 —— 这类错误不是单任务偶发,而是该
|
|
# provider 上所有 run 全挂(如 Zai 余额不足),等日巡检的 5 次/2 task 阈值太慢。
|
|
# 冷却:同归一化签名 6h 内只发一封(进程内存态,重启清零 —— 重启后再发一封可接受,
|
|
# 比引入持久化状态表划算)。
|
|
_CRITICAL_RE = re.compile(
|
|
r"余额不足|无可用资源包|请充值|欠费"
|
|
r"|insufficient[_ ](?:quota|balance|funds)"
|
|
r"|exceeded your current quota"
|
|
r"|AuthenticationError|invalid[_ ]api[_ ]?key|api key.{0,20}(?:invalid|expired)",
|
|
re.IGNORECASE,
|
|
)
|
|
_ALERT_COOLDOWN_S = 6 * 3600
|
|
_alerted_at: Dict[str, datetime] = {}
|
|
|
|
|
|
def alert_provider_critical(err: str, *, task_id: Any = None, model_profile: str = "") -> bool:
|
|
"""run 错误若属 provider 级致命(余额/认证)→ 立即邮件开发者;返回是否已发。
|
|
|
|
同步阻塞(SMTP),只应在 worker 线程调用(_run_agent_bg 的 except 路径本就在
|
|
to_thread 里)。所有失败静默 —— 告警绝不能反过来在错误路径上再抛异常。
|
|
"""
|
|
try:
|
|
if not err or not _CRITICAL_RE.search(err):
|
|
return False
|
|
sig = _normalize(err)
|
|
now = datetime.now(timezone.utc)
|
|
last = _alerted_at.get(sig)
|
|
if last and (now - last).total_seconds() < _ALERT_COOLDOWN_S:
|
|
return False
|
|
_alerted_at[sig] = now
|
|
import os
|
|
|
|
from tools.send_email import send_email_smtp, smtp_configured
|
|
print(f"[runerror] provider-critical task={task_id} mp={model_profile} "
|
|
f"err={err[:200]}", flush=True)
|
|
dev_email = os.getenv("ZCBOT_DEVELOPER_EMAIL", "").strip()
|
|
if not (dev_email and smtp_configured()):
|
|
print("[runerror] ZCBOT_DEVELOPER_EMAIL/SMTP 未配,仅日志", flush=True)
|
|
return False
|
|
body = "\n".join([
|
|
"检出 provider 级致命错误(余额/配额/认证),该 provider 上的 run 可能全部失败:",
|
|
"",
|
|
f"错误: {err[:500]}",
|
|
f"模型档: {model_profile or '?'}",
|
|
f"任务: {task_id or '?'}",
|
|
"",
|
|
f"同类签名 {_ALERT_COOLDOWN_S // 3600}h 内不再重复告警;"
|
|
"完整聚合见 admin 工具失败面板(kind=run)。",
|
|
])
|
|
send_email_smtp(dev_email, f"[zcbot] provider 级错误:{sig[:60]}", body)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def format_alert(clusters: List[Dict[str, Any]], days: float) -> str:
|
|
"""聚集列表 → 告警邮件正文(纯文本)。"""
|
|
lines = [f"近 {days:g} 天内检出 {len(clusters)} 类工具失败聚集(近 24h 活跃的在前):", ""]
|
|
for c in clusters:
|
|
n24 = c.get("count_24h", 0)
|
|
lines.append(
|
|
f"- [{c['tool']}/{c['kind']}] x{c['count']}"
|
|
f"(近24h {n24} 次{',已安静' if not n24 else ''},"
|
|
f"task {c['task_count']} 个 / 用户 {c['user_count']} 人,"
|
|
f"最近 {c['last_at']})"
|
|
)
|
|
lines.append(f" 签名: {c['signature']}")
|
|
lines.append(f" 样例: {c['sample'][:200]}")
|
|
lines.append("")
|
|
lines.append("排查入口:RUN.md 故障兜底表;历史案例:mermaid loopback DROP(0.58.10)。")
|
|
return "\n".join(lines)
|