57 lines
1.9 KiB
Python
57 lines
1.9 KiB
Python
"""对某 task:列出每条 run_python 报错的 tool-result,并回溯它配对的 assistant
|
|
tool_call 的 arguments(按 tool_call_id),判断报错那一刻 DB 里存的 args 是
|
|
真实 code / 空{} / 还是 _compacted 占位。"""
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
env = Path(__file__).resolve().parent.parent / ".env"
|
|
for line in env.read_text(encoding="utf-8").splitlines():
|
|
if line.strip().startswith("ZCBOT_DB_URL="):
|
|
os.environ["ZCBOT_DB_URL"] = line.split("=", 1)[1].strip()
|
|
|
|
from sqlalchemy import create_engine, text # noqa: E402
|
|
|
|
engine = create_engine(os.environ["ZCBOT_DB_URL"])
|
|
prefix = sys.argv[1] if len(sys.argv) > 1 else "9956b139"
|
|
ERR = "code or script_path must be provided"
|
|
|
|
with engine.connect() as conn:
|
|
tid = conn.execute(
|
|
text("select task_id from tasks where task_id::text like :p"),
|
|
{"p": prefix + "%"},
|
|
).fetchone()[0]
|
|
msgs = conn.execute(
|
|
text("select idx, payload from messages where task_id=:t order by idx"),
|
|
{"t": tid},
|
|
).fetchall()
|
|
|
|
# id -> (assist_idx, name, raw_args)
|
|
by_id = {}
|
|
for idx, payload in msgs:
|
|
if payload.get("role") == "assistant":
|
|
for tc in payload.get("tool_calls") or []:
|
|
fn = tc.get("function") or {}
|
|
by_id[tc.get("id")] = (idx, fn.get("name"), fn.get("arguments"))
|
|
|
|
print(f"task {tid}\n")
|
|
n = 0
|
|
for idx, payload in msgs:
|
|
if payload.get("role") != "tool":
|
|
continue
|
|
content = payload.get("content") or ""
|
|
if isinstance(content, list):
|
|
content = json.dumps(content, ensure_ascii=False)
|
|
if ERR not in content:
|
|
continue
|
|
n += 1
|
|
tcid = payload.get("tool_call_id")
|
|
src = by_id.get(tcid)
|
|
if src is None:
|
|
print(f"[err #{idx}] tcid={tcid} -> 找不到配对的 assistant 调用!")
|
|
continue
|
|
a_idx, name, raw = src
|
|
print(f"[err #{idx}] <- assist #{a_idx} {name} : {repr(raw)[:110]}")
|
|
print(f"\n共 {n} 条报错")
|