136 lines
4.4 KiB
Python
136 lines
4.4 KiB
Python
"""工具结果失败分类的单一事实源。"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from typing import Any, Optional, Tuple
|
|
|
|
|
|
_RE_UUID = re.compile(
|
|
r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-"
|
|
r"[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]")
|
|
_BARE_EXIT_RE = re.compile(r"^exit \d+$")
|
|
|
|
|
|
def normalize_failure_signature(value: str) -> str:
|
|
value = _RE_UUID.sub("<id>", value)
|
|
value = _RE_HEX.sub("<hex>", value)
|
|
value = _RE_PATH.sub("<path>", value)
|
|
value = _RE_NUM.sub("N", value)
|
|
return _RE_WS.sub(" ", value).strip()[:120]
|
|
|
|
|
|
def classify_failure(content: str) -> Optional[Tuple[str, str]]:
|
|
"""返回 ``(kind, 原始签名行)``；正常工具结果返回 None。"""
|
|
head = content.lstrip()
|
|
if "command timed out" in content:
|
|
return "timeout", "command timed out"
|
|
if head.startswith("[Error"):
|
|
return "error", head.splitlines()[0]
|
|
match = _EXIT_TAIL.search(content)
|
|
if match and match.group(1) != "0":
|
|
lines = [
|
|
line.strip()
|
|
for line in content.splitlines()[:-1]
|
|
if line.strip() and line.strip() not in _STREAM_MARKS
|
|
]
|
|
return "exit", (lines[-1] if lines else f"exit {match.group(1)}")
|
|
return None
|
|
|
|
|
|
def is_bare_exit_signature(value: str) -> bool:
|
|
return _BARE_EXIT_RE.match(value) is not None
|
|
|
|
|
|
def _command_hint(command: str) -> str:
|
|
command = command.strip().lower()
|
|
if not command:
|
|
return ""
|
|
if re.search(r"\b(?:apt|dpkg|pip)\b.*\b(?:list|show)\b", command):
|
|
return "dependency probe"
|
|
if re.search(r"(?:^|[;&|]\s*)(?:which|whereis|command\s+-v)\b", command):
|
|
return "dependency probe"
|
|
if re.search(r"(?:^|[;&|]\s*)(?:grep|rg)\b", command):
|
|
return "search/no match"
|
|
match = re.search(r"(?:^|&&|;|\|)\s*([a-z0-9_.-]+)", command)
|
|
return match.group(1) if match else ""
|
|
|
|
|
|
def shell_command_hint(prior_payload: Any, tool_call_id: str) -> str:
|
|
"""从历史 assistant tool_call 中提取稳定的 shell 命令类别。"""
|
|
if not isinstance(prior_payload, dict):
|
|
return ""
|
|
calls = prior_payload.get("tool_calls") or []
|
|
candidates = []
|
|
for call in calls:
|
|
if not isinstance(call, dict):
|
|
continue
|
|
function = call.get("function") or {}
|
|
if function.get("name") != "shell":
|
|
continue
|
|
if tool_call_id and call.get("id") == tool_call_id:
|
|
candidates = [call]
|
|
break
|
|
candidates.append(call)
|
|
if len(candidates) != 1:
|
|
return ""
|
|
return shell_command_hint_from_arguments(
|
|
(candidates[0].get("function") or {}).get("arguments")
|
|
)
|
|
|
|
|
|
def shell_command_hint_from_arguments(arguments: Any) -> str:
|
|
try:
|
|
args = json.loads(arguments) if isinstance(arguments, str) else arguments
|
|
except (TypeError, ValueError):
|
|
return ""
|
|
if not isinstance(args, dict):
|
|
return ""
|
|
return _command_hint(str(args.get("command") or ""))
|
|
|
|
|
|
def failure_category(kind: str, signature_line: str, sample: str) -> str:
|
|
if kind != "exit":
|
|
return "failure"
|
|
if signature_line.startswith("[GATE FAIL]"):
|
|
return "quality_gate"
|
|
if (
|
|
("[篇幅核算]" in sample or "[字数核算]" in sample)
|
|
and re.search(r"\[WARN\]\s*\d+\s*项超出\s*/\s*\d+\s*项不足", sample)
|
|
):
|
|
return "quality_gate"
|
|
if "[质量检查]" in sample and "[WARN] 共发现" in sample:
|
|
return "quality_gate"
|
|
return "failure"
|
|
|
|
|
|
def structured_failure(
|
|
tool: str,
|
|
content: str,
|
|
*,
|
|
arguments: Any = None,
|
|
) -> Optional[dict[str, str]]:
|
|
"""把工具结果转成可直接持久化的稳定失败事件。"""
|
|
hit = classify_failure(content)
|
|
if hit is None:
|
|
return None
|
|
kind, signature_line = hit
|
|
if tool == "shell" and kind == "exit" and _BARE_EXIT_RE.match(signature_line):
|
|
hint = shell_command_hint_from_arguments(arguments)
|
|
if hint:
|
|
signature_line = f"{signature_line} ({hint})"
|
|
return {
|
|
"tool": tool or "?",
|
|
"failure_kind": kind,
|
|
"signature": normalize_failure_signature(signature_line),
|
|
"category": failure_category(kind, signature_line, content),
|
|
"sample": content[:300],
|
|
}
|