Resilient LLM runner: model fallback chain, provider-death blacklist, cooldown, ANSI-repair
- llm.py: single retry/fallback/cooldown authority used by analyze/verify/pipeline - provider [404] blacklists model + switches; rate limit escalates then cooldown - unparsable model output (bad JSON/ANSI noise) skips to next model - verify.py: DS_VERIFY_TIMEOUT, true rate-limit handling, honest FAILED reporting - pipeline_all.py: honors model, reports verify status accurately - aggregate.py: skip epNNNN_corrected.json strays (was duplicating ep 304) - gitignore data outputs; drop stray analysis/analyst/analysts dirs
This commit is contained in:
n3wk
+67
-85
@@ -1,14 +1,15 @@
|
||||
"""Analyze episodes via the local `opencode run` CLI.
|
||||
"""Analyze episodes via the local `opencode run` CLI, routed through the
|
||||
resilient runner in llm.py (model fallback chain + provider-death blacklist).
|
||||
|
||||
For each transcript:
|
||||
1. export readable timestamped txt (export_txt.py)
|
||||
2. call: opencode run -f <txt> "<prompt with rubric + schema>"
|
||||
3. parse/validate the returned JSON (retry up to 3x on bad output)
|
||||
3. parse/validate the returned JSON (model fallback chain handles retries)
|
||||
4. save data/analysis/epXXXX.json
|
||||
|
||||
Then aggregate all analyses into one master file for routine opencode access:
|
||||
data/dickshow_analysis.json (run aggregate.py, or it runs automatically
|
||||
after each new analysis in this script)
|
||||
after each new analysis in this script)
|
||||
|
||||
Resumable: existing per-episode analyses are skipped unless --force.
|
||||
"""
|
||||
@@ -16,11 +17,12 @@ import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from llm import run as llm_run
|
||||
from llm import RateLimitError, ProviderError # noqa: F401 (compat)
|
||||
from config import DATA_DIR, BASE_DIR
|
||||
from feed import load_episodes
|
||||
from download import slug as ep_slug, audio_path
|
||||
@@ -36,7 +38,7 @@ OPENCODE = os.environ.get(
|
||||
"OPENCODE_BIN", os.path.expanduser(r"~\.opencode\bin\opencode.exe"))
|
||||
# analyzer model override, e.g. opencode/x-preview-f-free
|
||||
OPENCODE_MODEL = os.environ.get("OPENCODE_ANALYZE_MODEL", "").strip()
|
||||
OPENCODE_TIMEOUT = int(os.environ.get("DS_ANALYZE_TIMEOUT", "900")) # 15 min
|
||||
OPENCODE_TIMEOUT = int(os.environ.get("DS_ANALYZE_TIMEOUT", "1800")) # 30 min
|
||||
|
||||
PROMPT_TEMPLATE = """You are the show archivist for The Dick Show podcast. Attached is the FULL transcript of one episode (timestamped lines like [HH:MM:SS]).
|
||||
|
||||
@@ -69,14 +71,26 @@ def build_prompt() -> str:
|
||||
schema=SCHEMA_EXAMPLE.strip())
|
||||
|
||||
|
||||
_ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")
|
||||
# "tool call" display lines opencode may echo before the model's reply:
|
||||
# `→ Read file ...` / spinner glyphs / "Press Enter..." prompts.
|
||||
_TOOLCALL_RE = re.compile(
|
||||
r"(?m)^.*\x1b?\[?→.*$|^\s*(Press Enter|•|→)\s*$")
|
||||
|
||||
|
||||
def _repair(text: str) -> str:
|
||||
"""Common LLM-JSON repairs: fences, trailing commas, stray text, missing commas."""
|
||||
text = FENCES_RE.sub("", text.strip()).strip()
|
||||
"""Common LLM-JSON repairs: ANSI codes, fences, trailing commas, missing commas."""
|
||||
text = _ANSI_RE.sub("", text)
|
||||
text = FENCES_RE.sub("", text).strip()
|
||||
# extract just the outermost {...} JSON object if the reply is noisy
|
||||
start = text.find("{")
|
||||
end = text.rfind("}")
|
||||
if start > 0 or end < len(text) - 1:
|
||||
if start != -1 and end > start:
|
||||
text = text[start:end + 1]
|
||||
if start >= 0 and end > start:
|
||||
text = text[start:end + 1]
|
||||
elif text.strip():
|
||||
return text.strip()
|
||||
else:
|
||||
return ""
|
||||
# trailing commas before } or ]
|
||||
text = re.sub(r",(\s*[}\]])", r"\1", text)
|
||||
# missing commas between object members: "key": "value" "key2": "value2" -> add comma
|
||||
@@ -141,101 +155,69 @@ def validate(obj: dict) -> None:
|
||||
|
||||
|
||||
def run_opencode(prompt: str, attach: str) -> str:
|
||||
# NOTE: deliberately NO `-c`/`-s` session reuse: `run -c` continues the
|
||||
# globally-last session, which under concurrent pipelines cross-pollinates
|
||||
# analyses between episodes. Every call is a fresh session.
|
||||
# --auto: headless runs must not stall on permission prompts (the model
|
||||
# sometimes Reads attached files for more context); scope is our own data.
|
||||
cmd = [OPENCODE, "run", "--auto", "-f", attach,
|
||||
"--dir", BASE_DIR]
|
||||
if OPENCODE_MODEL:
|
||||
cmd += ["-m", OPENCODE_MODEL]
|
||||
cmd.append(prompt)
|
||||
proc = subprocess.run(cmd, capture_output=True, text=True,
|
||||
encoding="utf-8", errors="replace",
|
||||
cwd=BASE_DIR, timeout=OPENCODE_TIMEOUT)
|
||||
out = proc.stdout or ""
|
||||
err = proc.stderr or ""
|
||||
combined = (out + err).lower()
|
||||
if "rate limit" in combined or "429" in combined:
|
||||
raise RateLimitError("provider rate limit exceeded")
|
||||
if not out.strip():
|
||||
raise RuntimeError(f"opencode produced no stdout: {err[-400:]}")
|
||||
return out
|
||||
# NOTE: deliberately NO `-c`/`-s` session reuse. The resilient runner
|
||||
# (llm.py) owns fallback/retry/cooldown; this is the compat shim used by
|
||||
# verify.py for single-file analysis (JSON parsed by the caller).
|
||||
return llm_run(prompt, [attach], timeout=OPENCODE_TIMEOUT,
|
||||
label="analyze")
|
||||
|
||||
|
||||
# legacy alias kept for imports
|
||||
class RateLimitError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
# When the provider quota dries up, stop *attempting* analyses pipeline-wide
|
||||
# for a while; transcription (local) keeps running. Probe again after cooldown.
|
||||
# for a while; transcription (local) keeps running. Cooldown state lives in
|
||||
# llm.py and is shared with verify + pipeline_all.
|
||||
_RATE_COOLDOWN = {"until": 0.0}
|
||||
COOLDOWN_SECONDS = int(os.environ.get("DS_RATE_COOLDOWN", "900"))
|
||||
|
||||
|
||||
def analyze_episode(ep: dict, force: bool = False, retries: int = 3) -> dict | None:
|
||||
def analyze_episode(ep: dict, force: bool = False, retries: int = 1) -> dict | None:
|
||||
"""Analyze one episode. Retries/fallback/cooldown handled by llm.run.
|
||||
|
||||
Returns the analysis dict, or None if the episode was already done or the
|
||||
provider is unavailable (callers catch up on a later pass).
|
||||
"""
|
||||
os.makedirs(ANALYSIS_DIR, exist_ok=True)
|
||||
apath = os.path.join(ANALYSIS_DIR, f"{ep_slug(ep)}.json")
|
||||
if os.path.exists(apath) and not force:
|
||||
return None
|
||||
if time.time() < _RATE_COOLDOWN["until"]:
|
||||
print(f" (analysis cooldown {int(_RATE_COOLDOWN['until'] - time.time())}s "
|
||||
f"- will catch up later)", flush=True)
|
||||
return None
|
||||
if not os.path.exists(tpath(ep)):
|
||||
print(f" no transcript for {ep['title']}; skipping", flush=True)
|
||||
return None
|
||||
|
||||
attach = export_txt(ep)
|
||||
prompt = build_prompt()
|
||||
last_err = None
|
||||
# escalating waits; rate limits need minutes, not seconds
|
||||
waits = [10, 60, 180, 420, 600, 900]
|
||||
attempt = 0
|
||||
while attempt < len(waits):
|
||||
attempt += 1
|
||||
t0 = time.time()
|
||||
try:
|
||||
raw = run_opencode(
|
||||
(REPAIR_PREAMBLE + prompt) if attempt > 1 else prompt, attach)
|
||||
obj = strip_to_json(raw)
|
||||
validate(obj)
|
||||
ensure_content_match(obj, attach)
|
||||
obj["ep_num"] = ep["ep_num"] if ep["ep_num"] else obj.get("ep_num")
|
||||
obj["title"] = obj.get("title") or ep["title"]
|
||||
obj["pub_date"] = ep["pub_date"]
|
||||
obj["_analyzed_at"] = datetime.now(timezone.utc).isoformat()
|
||||
obj["_source_transcript"] = os.path.basename(tpath(ep))
|
||||
with open(apath + ".tmp", "w", encoding="utf-8") as f:
|
||||
json.dump(obj, f, ensure_ascii=False, indent=1)
|
||||
os.replace(apath + ".tmp", apath)
|
||||
dt = time.time() - t0
|
||||
ti = obj["transgression"]["overall_ti"]
|
||||
n_tp = len(obj.get("talking_points", []))
|
||||
n_q = len(obj.get("notable_quotes", []))
|
||||
print(f" OK {os.path.basename(apath)} in {dt:.0f}s "
|
||||
f"(TI={ti}, {n_tp} points, {n_q} quotes)", flush=True)
|
||||
return obj
|
||||
except RateLimitError as e:
|
||||
last_err = e
|
||||
wait = waits[min(attempt, len(waits) - 1)]
|
||||
print(f" rate-limited (attempt {attempt}); "
|
||||
f"sleeping {wait}s ...", flush=True)
|
||||
time.sleep(wait)
|
||||
except Exception as e: # noqa: BLE001
|
||||
last_err = e
|
||||
wait = waits[min(attempt - 1, len(waits) - 1)]
|
||||
print(f" attempt {attempt} failed: {e}; retry in {wait}s",
|
||||
flush=True)
|
||||
time.sleep(wait)
|
||||
print(f" GIVING UP on {ep['title']} after {attempt} attempts: "
|
||||
f"{last_err}", flush=True)
|
||||
if isinstance(last_err, RateLimitError):
|
||||
_RATE_COOLDOWN["until"] = time.time() + COOLDOWN_SECONDS
|
||||
print(f" entering analysis cooldown for {COOLDOWN_SECONDS}s "
|
||||
f"(transcription continues; analyses catch up later)",
|
||||
flush=True)
|
||||
return None
|
||||
|
||||
def parse_obj(raw: str) -> dict:
|
||||
obj = strip_to_json(raw)
|
||||
validate(obj)
|
||||
ensure_content_match(obj, attach)
|
||||
return obj
|
||||
|
||||
try:
|
||||
obj = llm_run(prompt, [attach], timeout=OPENCODE_TIMEOUT,
|
||||
label="analyze", parse=parse_obj)
|
||||
except (RateLimitError, ProviderError) as e:
|
||||
print(f" ANALYSIS NOT RUN {ep['title']}: {e}", flush=True)
|
||||
return None
|
||||
|
||||
obj["ep_num"] = ep["ep_num"] if ep["ep_num"] else obj.get("ep_num")
|
||||
obj["title"] = obj.get("title") or ep["title"]
|
||||
obj["pub_date"] = ep["pub_date"]
|
||||
obj["_analyzed_at"] = datetime.now(timezone.utc).isoformat()
|
||||
obj["_source_transcript"] = os.path.basename(tpath(ep))
|
||||
with open(apath + ".tmp", "w", encoding="utf-8") as f:
|
||||
json.dump(obj, f, ensure_ascii=False, indent=1)
|
||||
os.replace(apath + ".tmp", apath)
|
||||
ti = obj["transgression"]["overall_ti"]
|
||||
n_tp = len(obj.get("talking_points", []))
|
||||
n_q = len(obj.get("notable_quotes", []))
|
||||
print(f" OK {os.path.basename(apath)} "
|
||||
f"(TI={ti}, {n_tp} points, {n_q} quotes)", flush=True)
|
||||
return obj
|
||||
|
||||
|
||||
def main() -> None:
|
||||
|
||||
Reference in New Issue
Block a user