72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
"""Signale (Fehlerklassen) und ihre Erkennung."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from enum import StrEnum
|
|
|
|
|
|
class Signal(StrEnum):
|
|
NO_SKILL = "no_skill"
|
|
SKILL_CONFLICT = "skill_conflict"
|
|
SLOT_UNAVAILABLE = "slot_unavailable"
|
|
CONTEXT_EXCEEDED = "context_exceeded" # präventiv vom Router erkannt
|
|
COMPACT_PROMPT = "compact_prompt" # Ausweichen auf kompakte Skill-Variante
|
|
LARGER_SLOT = "larger_slot" # Ausweichen auf Slot mit mehr Kontext
|
|
NEEDS_SPLIT = "needs_split"
|
|
OUTPUT_TRUNCATED = "output_truncated"
|
|
DEGENERATE_OUTPUT = "degenerate_output"
|
|
INVALID_FORMAT = "invalid_format"
|
|
CONTEXT_OVERFLOW = "context_overflow" # Engine meldet Overflow
|
|
ENDPOINT_ERROR = "endpoint_error"
|
|
VERIFIER_FAILED = "verifier_failed"
|
|
CLASSIFIER_LOW_CONFIDENCE = "classifier_low_confidence"
|
|
PROGRAM_ERROR = "program_error"
|
|
MISROUTE_SUSPECTED = "misroute_suspected"
|
|
|
|
|
|
INFRA_SIGNALS = {Signal.ENDPOINT_ERROR, Signal.CONTEXT_OVERFLOW}
|
|
QUALITY_SIGNALS = {
|
|
Signal.VERIFIER_FAILED,
|
|
Signal.INVALID_FORMAT,
|
|
Signal.DEGENERATE_OUTPUT,
|
|
Signal.OUTPUT_TRUNCATED,
|
|
Signal.CLASSIFIER_LOW_CONFIDENCE,
|
|
Signal.PROGRAM_ERROR,
|
|
}
|
|
|
|
_OVERFLOW_PATTERNS = re.compile(
|
|
r"context length|maximum context|too many tokens|context window|prompt is too long", re.IGNORECASE
|
|
)
|
|
_WORD = re.compile(r"\w+", re.UNICODE)
|
|
|
|
|
|
def is_context_overflow_message(text: str) -> bool:
|
|
return bool(_OVERFLOW_PATTERNS.search(text or ""))
|
|
|
|
|
|
def repetition_ratio(text: str, n: int = 4) -> float:
|
|
"""Anteil der n-Gramme, die Wiederholungen sind (0 = keine)."""
|
|
words = _WORD.findall(text.lower())
|
|
if len(words) < n:
|
|
return 0.0
|
|
grams = [tuple(words[i:i + n]) for i in range(len(words) - n + 1)]
|
|
return 1.0 - len(set(grams)) / len(grams)
|
|
|
|
|
|
def is_degenerate(text: str, min_words: int = 40, max_ratio: float = 0.5) -> bool:
|
|
return len(_WORD.findall(text)) >= min_words and repetition_ratio(text) > max_ratio
|
|
|
|
|
|
def extract_json(text: str) -> object:
|
|
"""Parst JSON, toleriert ```json-Codeblöcke. Wirft ValueError bei ungültigem JSON."""
|
|
stripped = text.strip()
|
|
fence = re.match(r"^```(?:json)?\s*\n(.*?)\n```\s*$", stripped, re.DOTALL)
|
|
if fence:
|
|
stripped = fence.group(1)
|
|
try:
|
|
return json.loads(stripped)
|
|
except json.JSONDecodeError as exc:
|
|
raise ValueError(str(exc)) from exc
|