clipforge / src /humeo /hook_detector.py
moonlantern1's picture
Add OpenRouter backup key failover
5cd4a32 verified
"""Stage 2.25 - Hook detection.
The clip-selection LLM returns a ``hook_start_sec`` / ``hook_end_sec`` pair
per clip, but in practice it almost always echoes the ``[0.0, 3.0]``
placeholder from the prompt instead of localising the real hook sentence.
That placeholder is toxic to Stage 2.5 pruning -- the clamp refuses to
trim past ``hook_start_sec``, so every ``trim_start_sec > 0`` the pruner
returns gets zeroed out silently.
This module is a dedicated Stage 2.25 that runs between clip selection and
content pruning. For each clip it:
1. Prepares a clip-relative segment listing (same format as pruning uses).
2. Asks Gemini, in one batched JSON call, to localise the hook sentence of
every clip with `hook_start_sec`, `hook_end_sec`, `hook_text`, `reason`.
3. Validates the returned window against the clip's duration + the "real
hook" heuristics, then overwrites ``clip.hook_start_sec`` /
``clip.hook_end_sec`` on a copy of the clip.
The stage is:
- **Cached** (``hooks.json`` / ``hooks.meta.json`` in ``work_dir``) on
``transcript_sha256 + clips_sha256 + gemini_model``.
- **Never fatal.** Any failure (API error, malformed JSON, clip not
returned, window that still looks like the 0.0-3.0 placeholder) falls
back to the original clip with its original hook -- pruning will then
skip hook protection via the fingerprint guard in
:func:`humeo.content_pruning._looks_like_default_hook`.
The stage writes three artifacts to ``work_dir`` for audit:
- ``hooks.meta.json``: cache key (version, fingerprints, model).
- ``hooks.json``: structured per-clip hook windows actually applied.
- ``hooks_raw.json``: verbatim Gemini response text (for prompt tuning).
"""
from __future__ import annotations
import hashlib
import json
import logging
import time
from pathlib import Path
from typing import Any, Callable, TypeVar
from google import genai
from openai import OpenAI
from pydantic import BaseModel, Field, ValidationError
from humeo_core.schemas import Clip
from humeo.config import GEMINI_MODEL, PipelineConfig
from humeo.content_pruning import _looks_like_default_hook, _segments_within_clip
from humeo.env import (
OPENROUTER_BASE_URL,
current_llm_provider,
model_name_for_provider,
openrouter_default_headers,
resolve_gemini_api_key,
resolve_llm_provider,
resolve_openrouter_api_keys,
)
from humeo.gemini_generate import gemini_generate_config
from humeo.hook_library import (
format_hook_examples,
hook_library_fingerprint,
resolve_hook_library_path,
retrieve_hook_examples,
)
from humeo.prompt_loader import hook_detection_system_prompt
logger = logging.getLogger(__name__)
T = TypeVar("T")
HOOK_META_VERSION = 2
HOOK_META_FILENAME = "hooks.meta.json"
HOOK_ARTIFACT_FILENAME = "hooks.json"
HOOK_RAW_FILENAME = "hooks_raw.json"
LLM_MAX_ATTEMPTS = 3
LLM_RETRY_DELAY_SEC = 2.0
# Hook window validation thresholds. The prompt asks for 1.5-7.0s windows;
# we enforce 1.0-10.0s to be lenient on rounding while still rejecting
# obvious "LLM returned the whole paragraph" mistakes.
_MIN_HOOK_DURATION_SEC = 1.0
_MAX_HOOK_DURATION_SEC = 10.0
def _openai_message_text(content: object) -> str:
if isinstance(content, str):
return content
if isinstance(content, list):
parts: list[str] = []
for item in content:
if isinstance(item, dict) and item.get("type") == "text":
text = item.get("text")
if isinstance(text, str):
parts.append(text)
return "".join(parts)
return ""
class _HookDecision(BaseModel):
"""Per-clip hook window returned by Gemini (clip-relative seconds)."""
clip_id: str
hook_start_sec: float = Field(ge=0.0)
hook_end_sec: float = Field(ge=0.0)
hook_text: str = ""
reason: str = ""
class _HookResponse(BaseModel):
hooks: list[_HookDecision] = Field(default_factory=list)
def _retry_llm(name: str, fn: Callable[[], T], attempts: int = LLM_MAX_ATTEMPTS) -> T:
last: Exception | None = None
for i in range(attempts):
try:
return fn()
except Exception as e: # noqa: BLE001 - rethrown below
last = e
if i < attempts - 1:
logger.warning("%s attempt %d/%d failed: %s", name, i + 1, attempts, e)
time.sleep(LLM_RETRY_DELAY_SEC * (i + 1))
assert last is not None
raise last
# ---------------------------------------------------------------------------
# Prompt construction
# ---------------------------------------------------------------------------
def _build_user_message(clips: list[Clip], transcript: dict) -> str:
"""Render clip-relative segments + selector-guessed hook text for each clip."""
blocks: list[str] = []
for clip in clips:
segs = _segments_within_clip(transcript, clip)
header_lines = [
f"clip_id: {clip.clip_id}",
f"duration_sec: {clip.duration_sec:.2f}",
f"topic: {clip.topic}",
]
if clip.viral_hook:
header_lines.append(f"viral_hook_text: {clip.viral_hook}")
if clip.hook_start_sec is not None and clip.hook_end_sec is not None:
header_lines.append(
f"selector_hook_window_sec: [{clip.hook_start_sec:.2f}, "
f"{clip.hook_end_sec:.2f}] (may be a placeholder; verify)"
)
header = "\n".join(header_lines)
body = "\n".join(
f"[{seg['start']:.2f}s - {seg['end']:.2f}s] {seg['text']}" for seg in segs
)
if not body:
body = "(no segments overlap this clip window)"
blocks.append(f"{header}\n---\n{body}")
return "\n\n===\n\n".join(blocks)
# ---------------------------------------------------------------------------
# Validation
# ---------------------------------------------------------------------------
def _validate_hook_window(
clip: Clip, hook_start: float, hook_end: float
) -> tuple[float, float] | None:
"""Return a valid (hook_start, hook_end) or None if rejected.
Rules:
- ``0 <= hook_start < hook_end <= duration_sec``
- hook duration between ``_MIN_HOOK_DURATION_SEC`` and ``_MAX_HOOK_DURATION_SEC``
- NOT the ``(0.0, 3.0)`` placeholder fingerprint (we'd rather keep the
selector's value untouched than re-apply the same fake hook).
"""
if hook_start < 0.0 or hook_end <= hook_start:
return None
if hook_end > clip.duration_sec + 1e-3:
# Clamp trailing rounding to duration; reject anything beyond.
if hook_end - clip.duration_sec > 0.5:
return None
hook_end = clip.duration_sec
dur = hook_end - hook_start
if dur < _MIN_HOOK_DURATION_SEC or dur > _MAX_HOOK_DURATION_SEC:
return None
if _looks_like_default_hook(hook_start, hook_end):
return None
return float(hook_start), float(hook_end)
# ---------------------------------------------------------------------------
# Apply decisions -> new clips
# ---------------------------------------------------------------------------
def apply_hook_decisions(
clips: list[Clip],
decisions: list[_HookDecision],
) -> list[Clip]:
"""Return new clips whose hook fields reflect validated decisions.
Clips without a matching valid decision are returned unchanged (their
original hook metadata, placeholder or not, is preserved).
"""
by_id = {d.clip_id: d for d in decisions}
out: list[Clip] = []
changed = 0
rejected = 0
for clip in clips:
d = by_id.get(clip.clip_id)
if d is None:
out.append(clip)
continue
validated = _validate_hook_window(clip, d.hook_start_sec, d.hook_end_sec)
if validated is None:
logger.info(
"Clip %s: rejected hook window [%.2f, %.2f] (failed validation); "
"keeping selector hook.",
clip.clip_id,
d.hook_start_sec,
d.hook_end_sec,
)
rejected += 1
out.append(clip)
continue
hs, he = validated
if (
clip.hook_start_sec is not None
and clip.hook_end_sec is not None
and abs(clip.hook_start_sec - hs) < 1e-3
and abs(clip.hook_end_sec - he) < 1e-3
):
out.append(clip)
continue
changed += 1
logger.info(
"Clip %s: hook set to [%.2f, %.2f] (was [%s, %s]) -- %s",
clip.clip_id,
hs,
he,
f"{clip.hook_start_sec:.2f}" if clip.hook_start_sec is not None else "None",
f"{clip.hook_end_sec:.2f}" if clip.hook_end_sec is not None else "None",
d.reason[:120] if d.reason else "(no reason)",
)
out.append(
clip.model_copy(update={"hook_start_sec": hs, "hook_end_sec": he})
)
logger.info(
"Hook detection: updated %d / %d clips (%d rejected, %d kept as-is).",
changed,
len(clips),
rejected,
len(clips) - changed - rejected,
)
return out
# ---------------------------------------------------------------------------
# Cache
# ---------------------------------------------------------------------------
def _clips_fingerprint(clips: list[Clip]) -> str:
payload = json.dumps(
[
{"id": c.clip_id, "s": round(c.start_time_sec, 3), "e": round(c.end_time_sec, 3)}
for c in clips
],
sort_keys=True,
ensure_ascii=False,
)
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
def _resolved_gemini_model(config: PipelineConfig) -> str:
return (config.gemini_model or GEMINI_MODEL).strip()
def _hook_meta(
*,
transcript_fp: str,
clips_fp: str,
config: PipelineConfig,
) -> dict[str, Any]:
return {
"version": HOOK_META_VERSION,
"transcript_sha256": transcript_fp,
"clips_sha256": clips_fp,
"gemini_model": _resolved_gemini_model(config),
"llm_backend": current_llm_provider() or "google",
"hook_library_sha256": hook_library_fingerprint(resolve_hook_library_path(config)),
}
def _hook_cache_valid(
work_dir: Path,
*,
transcript_fp: str,
clips_fp: str,
config: PipelineConfig,
) -> bool:
meta_path = work_dir / HOOK_META_FILENAME
if not meta_path.is_file():
return False
try:
with open(meta_path, encoding="utf-8") as f:
meta = json.load(f)
except Exception:
return False
if meta.get("version") != HOOK_META_VERSION:
return False
if meta.get("transcript_sha256") != transcript_fp:
return False
if meta.get("clips_sha256") != clips_fp:
return False
current_provider = current_llm_provider()
meta_provider = meta.get("llm_backend")
if current_provider == "openrouter":
if meta_provider != "openrouter":
return False
elif current_provider == "google":
if meta_provider not in (None, "google"):
return False
if meta.get("gemini_model") != _resolved_gemini_model(config):
return False
if meta.get("hook_library_sha256", "") != hook_library_fingerprint(resolve_hook_library_path(config)):
return False
return True
def _load_cached_hooks(
work_dir: Path, clips: list[Clip]
) -> list[Clip] | None:
artifact = work_dir / HOOK_ARTIFACT_FILENAME
if not artifact.is_file():
return None
try:
with open(artifact, "r", encoding="utf-8") as f:
data = json.load(f)
cached = {item["clip_id"]: item for item in data.get("hooks", [])}
except Exception as e: # noqa: BLE001 - surfaced as warning below
logger.warning("Hook cache artifact unreadable (%s); re-running.", e)
return None
out: list[Clip] = []
for clip in clips:
c = cached.get(clip.clip_id)
if c is None:
out.append(clip)
continue
hs = c.get("hook_start_sec")
he = c.get("hook_end_sec")
if hs is None or he is None:
out.append(clip)
continue
out.append(
clip.model_copy(
update={"hook_start_sec": float(hs), "hook_end_sec": float(he)}
)
)
return out
def _write_cache(
work_dir: Path,
*,
clips_with_hooks: list[Clip],
decisions: list[_HookDecision],
meta: dict[str, Any],
raw_response: str,
) -> None:
work_dir.mkdir(parents=True, exist_ok=True)
reasons = {d.clip_id: d for d in decisions}
payload = {
"hooks": [
{
"clip_id": c.clip_id,
"hook_start_sec": c.hook_start_sec,
"hook_end_sec": c.hook_end_sec,
"hook_text": (reasons.get(c.clip_id).hook_text if reasons.get(c.clip_id) else ""),
"reason": (reasons.get(c.clip_id).reason if reasons.get(c.clip_id) else ""),
}
for c in clips_with_hooks
]
}
(work_dir / HOOK_ARTIFACT_FILENAME).write_text(
json.dumps(payload, indent=2), encoding="utf-8"
)
(work_dir / HOOK_RAW_FILENAME).write_text(raw_response, encoding="utf-8")
with open(work_dir / HOOK_META_FILENAME, "w", encoding="utf-8") as f:
json.dump(meta, f, indent=2)
f.write("\n")
logger.info(
"Wrote %s, %s and %s",
HOOK_META_FILENAME,
HOOK_ARTIFACT_FILENAME,
HOOK_RAW_FILENAME,
)
# ---------------------------------------------------------------------------
# Gemini call
# ---------------------------------------------------------------------------
def _parse_decisions(raw_json: str) -> list[_HookDecision]:
data = json.loads(raw_json)
if isinstance(data, dict) and "hooks" in data:
try:
return _HookResponse.model_validate(data).hooks
except ValidationError as e:
logger.warning("Hook response failed validation: %s", e)
return []
if isinstance(data, list):
out: list[_HookDecision] = []
for item in data:
try:
out.append(_HookDecision.model_validate(item))
except ValidationError:
continue
return out
return []
def request_hook_decisions(
clips: list[Clip],
transcript: dict,
*,
gemini_model: str | None = None,
hook_library_path: Path | None = None,
) -> tuple[list[_HookDecision], str]:
"""Ask Gemini to localise the hook sentence for each clip.
Returns ``(decisions, raw_response)``. ``raw_response`` is the literal
JSON text from Gemini (cached to ``hooks_raw.json`` for audit). On
transport/parse failure this raises; callers should catch and treat as
no-op.
"""
if not clips:
return [], '{"hooks": []}'
example_query = " ".join(
filter(None, [*(clip.topic for clip in clips[:4]), *(clip.viral_hook for clip in clips[:4])])
)
hook_examples = format_hook_examples(
retrieve_hook_examples(example_query, path=hook_library_path, limit=8)
)
system = hook_detection_system_prompt(hook_examples=hook_examples)
user_text = _build_user_message(clips, transcript)
provider = resolve_llm_provider()
model_name = model_name_for_provider((gemini_model or GEMINI_MODEL).strip(), provider)
def _call() -> str:
logger.info(
"%s hook detection (model=%s, clips=%d)...", provider, model_name, len(clips)
)
if provider == "google":
client = genai.Client(api_key=resolve_gemini_api_key())
response = client.models.generate_content(
model=model_name,
contents=user_text,
config=gemini_generate_config(
system_instruction=system,
temperature=0.2,
response_mime_type="application/json",
),
)
if not response.text:
raise RuntimeError("Gemini returned empty response text for hook detection")
return response.text
keys = resolve_openrouter_api_keys()
last_error: Exception | None = None
for key_idx, api_key in enumerate(keys, start=1):
try:
client = OpenAI(
api_key=api_key,
base_url=OPENROUTER_BASE_URL,
default_headers=openrouter_default_headers(),
)
response = client.chat.completions.create(
model=model_name,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": user_text},
],
temperature=0.2,
response_format={"type": "json_object"},
)
text = _openai_message_text(response.choices[0].message.content)
if not text:
raise RuntimeError("OpenRouter returned empty response text for hook detection")
if key_idx > 1:
logger.info("OpenRouter hook detection succeeded with fallback key %d/%d", key_idx, len(keys))
return text
except Exception as exc:
last_error = exc
if key_idx < len(keys):
logger.warning(
"OpenRouter hook detection failed with key %d/%d: %s; trying fallback",
key_idx,
len(keys),
exc,
)
assert last_error is not None
raise last_error
raw = _retry_llm("Gemini hook detection", _call)
decisions = _parse_decisions(raw)
return decisions, raw
# ---------------------------------------------------------------------------
# Public stage entrypoint
# ---------------------------------------------------------------------------
def run_hook_detection_stage(
work_dir: Path,
clips: list[Clip],
transcript: dict,
*,
transcript_fp: str,
config: PipelineConfig,
) -> list[Clip]:
"""Run Stage 2.25 hook detection and return clips with localised hooks.
- Disabled (``config.detect_hooks is False``): return clips unchanged.
- Cache hit: read ``hooks.json`` and apply cached windows.
- LLM failure: log a warning and return clips unchanged. The downstream
content pruner's fingerprint guard will treat any remaining placeholder
hooks as "no hook" so pruning still runs.
"""
if not config.detect_hooks:
logger.info("Hook detection disabled (detect_hooks=False); skipping Stage 2.25.")
return clips
if not clips:
return clips
clips_fp = _clips_fingerprint(clips)
if not config.force_hook_detection and _hook_cache_valid(
work_dir,
transcript_fp=transcript_fp,
clips_fp=clips_fp,
config=config,
):
cached = _load_cached_hooks(work_dir, clips)
if cached is not None:
logger.info(
"Hook detection cache hit (%d clips); skipping LLM.", len(clips)
)
return cached
try:
decisions, raw = request_hook_decisions(
clips,
transcript,
gemini_model=config.gemini_model,
hook_library_path=resolve_hook_library_path(config),
)
except Exception as e: # noqa: BLE001 - pipeline must not die here
logger.warning(
"Hook detection call failed (%s); continuing with selector hooks. "
"Content pruning will treat any [0.0, 3.0] placeholder as 'no hook'.",
e,
)
return clips
updated = apply_hook_decisions(clips, decisions)
meta = _hook_meta(
transcript_fp=transcript_fp, clips_fp=clips_fp, config=config
)
try:
_write_cache(
work_dir,
clips_with_hooks=updated,
decisions=decisions,
meta=meta,
raw_response=raw,
)
except Exception as e: # noqa: BLE001 - cache failure is not fatal
logger.warning("Failed to write hook cache (%s); continuing.", e)
return updated