"""Plug-and-play LLM adapter for TalkingHeadBench ingestion bundles.""" from __future__ import annotations import ipaddress import json import os from typing import Any, Literal from urllib.parse import urlparse from openai import APIConnectionError, APIStatusError, OpenAI Provider = Literal["openai", "anthropic", "huggingface", "local"] TaskTier = Literal[ "image_audit", "clip_audit", "weight_audit", "easy", "medium", "hard", ] _DRIFT_WARNING_THRESHOLD = 0.22 _DRIFT_HIGH_RISK_THRESHOLD = 0.35 _BLUR_WARNING_THRESHOLD = 0.30 _BLUR_HIGH_RISK_THRESHOLD = 0.20 _LIPSYNC_WARNING_THRESHOLD = 0.40 _LIPSYNC_HIGH_RISK_THRESHOLD = 0.30 _SYSTEM_PROMPT = ( "You are a TalkingHeadBench diagnostic assistant specializing in talking-head LoRA pipelines.\n" "You receive only pre-extracted, deterministic signals and never raw generated outputs.\n" "Rules:\n" "1. Every claim must cite a specific digest field name and measured value.\n" "2. Parameter recommendations must be directional only (increase/decrease/enable/disable/reconsider) and never include target numbers.\n" "3. If fallback extractor data is present, treat fallback clip lip-sync and phoneme signals as unreliable.\n" "4. Do not invent any signal that is not present in the digest.\n" "5. Follow the required section structure from the user prompt exactly." ) class LLMAdapterError(Exception): """Structured error surfaced by the one-shot analysis adapter.""" def __init__( self, *, code: str, message: str, status_code: int, retryable: bool = False, ) -> None: super().__init__(message) self.code = code self.message = message self.status_code = status_code self.retryable = retryable def analyze_ingested_bundle( bundle: dict[str, Any], *, model_id: str | None, api_key: str | None, provider: str | None, base_url: str | None, max_tokens: int, temperature: float, timeout_s: float, task_tier: TaskTier = "weight_audit", ) -> dict[str, Any]: """Generate a one-shot natural-language report from an ingested signal bundle.""" if not isinstance(bundle, dict): raise LLMAdapterError( code="invalid_ingestion_bundle", message="The ingested bundle payload is invalid.", status_code=400, retryable=False, ) resolved_provider = _resolve_provider( provider=provider, model_id=model_id, api_key=api_key, base_url=base_url, ) resolved_model = _resolve_model_id(resolved_provider, model_id) resolved_base_url = _resolve_custom_base_url( provider=resolved_provider, base_url=base_url, ) signal_digest = _build_signal_digest(bundle) prompt = _build_prompt(signal_digest, task_tier=task_tier) if resolved_provider == "openai": report = _call_openai( model_id=resolved_model, api_key=api_key, prompt=prompt, max_tokens=max_tokens, temperature=temperature, timeout_s=timeout_s, base_url=resolved_base_url, ) elif resolved_provider == "anthropic": report = _call_anthropic( model_id=resolved_model, api_key=api_key, prompt=prompt, max_tokens=max_tokens, temperature=temperature, timeout_s=timeout_s, base_url=resolved_base_url, ) elif resolved_provider == "huggingface": report = _call_huggingface( model_id=resolved_model, api_key=api_key, prompt=prompt, max_tokens=max_tokens, temperature=temperature, timeout_s=timeout_s, base_url=resolved_base_url, ) else: report = _call_local( model_id=resolved_model, api_key=api_key, prompt=prompt, max_tokens=max_tokens, temperature=temperature, timeout_s=timeout_s, base_url=resolved_base_url, ) text = (report or "").strip() if not text: raise LLMAdapterError( code="empty_model_response", message="The model provider returned an empty analysis response.", status_code=502, retryable=True, ) return { "provider": resolved_provider, "model_id": resolved_model, "report": text, "signal_digest": signal_digest, } def _resolve_provider( *, provider: str | None, model_id: str | None, api_key: str | None, base_url: str | None, ) -> Provider: requested = (provider or "auto").strip().lower() if requested == "hf": requested = "huggingface" if requested not in {"auto", "openai", "anthropic", "huggingface", "local"}: raise LLMAdapterError( code="unsupported_provider", message=( "provider must be one of: auto, openai, anthropic, huggingface, local" ), status_code=400, retryable=False, ) if requested != "auto": return requested # type: ignore[return-value] key = (api_key or "").strip() model = (model_id or os.getenv("MODEL_NAME") or "").strip().lower() resolved_base = (base_url or os.getenv("API_BASE_URL") or "").strip().lower() if key.startswith("sk-ant-") or model.startswith("claude"): return "anthropic" if key.startswith("hf_"): return "huggingface" if key.startswith("sk-") or model.startswith(("gpt-", "o1", "o3", "o4")): return "openai" if "/" in model: return "huggingface" if "router.huggingface.co" in resolved_base or "api-inference.huggingface.co" in resolved_base: return "huggingface" if "api.openai.com" in resolved_base: return "openai" if "api.anthropic.com" in resolved_base: return "anthropic" if resolved_base and not any( marker in resolved_base for marker in ["localhost", "127.0.0.1", "0.0.0.0", "11434"] ): # Any configured remote OpenAI-compatible base should avoid local fallback. return "openai" return "local" def _resolve_model_id(provider: Provider, model_id: str | None) -> str: cleaned = (model_id or "").strip() if cleaned: return cleaned global_model = os.getenv("MODEL_NAME", "").strip() if global_model: return global_model if provider == "openai": return os.getenv("THB_OPENAI_MODEL", "gpt-4o-mini") if provider == "anthropic": return os.getenv("THB_ANTHROPIC_MODEL", "claude-3-5-haiku-latest") if provider == "huggingface": env_model = os.getenv("THB_HF_MODEL", "").strip() if env_model: return env_model # Default to a well-supported chat model via the HF router. return "Qwen/Qwen2.5-72B-Instruct" return os.getenv("THB_LOCAL_MODEL", "llama3.1:8b-instruct-q4_K_M") def _env_truthy(name: str, *, default: bool) -> bool: raw = os.getenv(name) if raw is None: return default return raw.strip().lower() in {"1", "true", "yes", "on"} def _allowed_base_url_prefixes() -> list[str]: raw = os.getenv("THB_ALLOWED_BASE_URL_PREFIXES", "") prefixes = [item.strip().rstrip("/") for item in raw.split(",") if item.strip()] return prefixes def _resolve_custom_base_url(*, provider: Provider, base_url: str | None) -> str | None: custom = (base_url or "").strip() if not custom: return None configured_api_base = (os.getenv("API_BASE_URL") or "").strip() if configured_api_base and _normalize_base_url(custom) == _normalize_base_url(configured_api_base): return custom if not _env_truthy("THB_ALLOW_CUSTOM_BASE_URLS", default=False): raise LLMAdapterError( code="custom_base_url_disabled", message="Custom base_url is disabled for this deployment.", status_code=400, retryable=False, ) _validate_custom_base_url(custom, provider=provider) allowed_prefixes = _allowed_base_url_prefixes() if allowed_prefixes: normalized = custom.rstrip("/") if not any(normalized.startswith(prefix) for prefix in allowed_prefixes): raise LLMAdapterError( code="base_url_not_allowed", message="base_url is not in THB_ALLOWED_BASE_URL_PREFIXES.", status_code=400, retryable=False, ) return custom def _normalize_base_url(url: str) -> str: return url.strip().rstrip("/") def _validate_custom_base_url(base_url: str, *, provider: Provider) -> None: parsed = urlparse(base_url) if parsed.scheme not in {"http", "https"} or not parsed.netloc: raise LLMAdapterError( code="invalid_base_url", message="base_url must be a valid http(s) URL.", status_code=400, retryable=False, ) if parsed.username or parsed.password: raise LLMAdapterError( code="invalid_base_url", message="base_url must not contain embedded credentials.", status_code=400, retryable=False, ) host = (parsed.hostname or "").strip().lower() if not host: raise LLMAdapterError( code="invalid_base_url", message="base_url must include a hostname.", status_code=400, retryable=False, ) if host in {"localhost"} or host.endswith(".local"): raise LLMAdapterError( code="unsafe_base_url", message=( f"base_url host is not allowed for public deployment (provider={provider})." ), status_code=400, retryable=False, ) try: ip = ipaddress.ip_address(host) except ValueError: return if ( ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved ): raise LLMAdapterError( code="unsafe_base_url", message=( f"base_url host is not allowed for public deployment (provider={provider})." ), status_code=400, retryable=False, ) def _require_api_key(provider: str, api_key: str | None) -> str: key = (api_key or "").strip() if key: return key raise LLMAdapterError( code="missing_api_key", message=f"Provider '{provider}' requires api_key.", status_code=400, retryable=False, ) def _build_signal_digest(bundle: dict[str, Any]) -> dict[str, Any]: image_obs = _as_dict(bundle.get("image_observation")) clip_obs = _as_list_of_dicts(bundle.get("clip_signal_observations")) weight_obs = _as_dict(bundle.get("weight_observation")) metadata = _as_dict(bundle.get("ingestion_metadata")) extractor_metadata = _as_dict(metadata.get("extractor_metadata")) fallback_clip_ids = { str(item).strip() for item in _as_list(extractor_metadata.get("fallback_clip_ids")) if str(item).strip() } drifts: list[float] = [] blurs: list[float] = [] lips: list[float] = [] reliable_lips: list[float] = [] fallback_clip_count = 0 high_risk_clip_ids: list[str] = [] high_risk_clips: list[dict[str, Any]] = [] quality_distribution: dict[str, int] = { "good": 0, "acceptable": 0, "marginal": 0, "poor": 0, } per_clip_breakdown: list[dict[str, Any]] = [] per_clip_signals: list[dict[str, Any]] = [] max_per_clip_signals = _max_clip_signals_limit() for index, item in enumerate(clip_obs): clip_id = str(item.get("clip_id", "unknown")) drift = _as_float(item.get("identity_cosine_drift"), default=0.0) blur = _as_float(item.get("blur_score"), default=1.0) lip_sync = _as_float(item.get("lip_sync_confidence"), default=1.0) fallback_used = _as_bool(item.get("fallback_extractor_used")) or clip_id in fallback_clip_ids lip_sync_reliable = not fallback_used if fallback_used: fallback_clip_count += 1 drifts.append(drift) blurs.append(blur) lips.append(lip_sync) if lip_sync_reliable: reliable_lips.append(lip_sync) high_risk_reasons, warning_reasons = _clip_risk_reasons( drift=drift, blur=blur, lip_sync=lip_sync, lip_sync_reliable=lip_sync_reliable, fallback_used=fallback_used, ) high_risk_reason_count = len(high_risk_reasons) is_high_risk = high_risk_reason_count >= 2 has_warning = bool(warning_reasons) or high_risk_reason_count == 1 risk_level = "high_risk" if is_high_risk else ("warning" if has_warning else "clean") quality_tier = _clip_quality_tier( drift=drift, blur=blur, lip_sync=lip_sync, lip_sync_reliable=lip_sync_reliable, fallback_used=fallback_used, high_risk_reason_count=high_risk_reason_count, ) quality_distribution[quality_tier] += 1 if is_high_risk: high_risk_clip_ids.append(clip_id) high_risk_clips.append( { "clip_id": clip_id, "reasons": high_risk_reasons, "risk_level": risk_level, "quality_tier": quality_tier, "lip_sync_reliable": lip_sync_reliable, } ) if index < max_per_clip_signals: per_clip_signals.append( { "clip_id": clip_id, "identity_cosine_drift": drift, "drift_severity": _drift_severity(drift), "blur_score": blur, "lip_sync_confidence": lip_sync, "phoneme_coverage_new": _as_int(item.get("phoneme_coverage_new")), "blink_count": _as_int(item.get("blink_count")), "occlusion_frames": _as_int(item.get("occlusion_frames")), "fallback_extractor_used": fallback_used, "provisional_due_to_fallback": fallback_used, "lip_sync_reliable": lip_sync_reliable, "is_high_risk": is_high_risk, "risk_level": risk_level, "high_risk_reasons": high_risk_reasons, "warning_reasons": warning_reasons, "quality_tier": quality_tier, } ) per_clip_breakdown.append( { "clip_id": clip_id, "identity_cosine_drift": round(drift, 4), "blur_score": round(blur, 4), "lip_sync_confidence": round(lip_sync, 4), "landmark_stability_score": _as_float(item.get("landmark_stability_score")), "phoneme_coverage_new": _as_float(item.get("phoneme_coverage_new")), "exposure_score": _as_float(item.get("exposure_score")), "occlusion_frames": _as_int(item.get("occlusion_frames")), "fallback_extractor_used": fallback_used, "risk_flags": ( (["high_drift"] if drift >= _DRIFT_WARNING_THRESHOLD else []) + (["low_blur"] if blur <= _BLUR_WARNING_THRESHOLD else []) + (["poor_lip_sync"] if lip_sync <= _LIPSYNC_WARNING_THRESHOLD else []) ), } ) layer_entropy = _as_dict(weight_obs.get("canonical_entropy_per_layer")) rank_util = _as_dict(weight_obs.get("layer_rank_utilization")) layer_sparsity = _as_dict(weight_obs.get("layer_sparsity")) gradient_noise = _as_float(weight_obs.get("gradient_noise_estimate")) high_entropy_positions = _as_list(weight_obs.get("high_entropy_token_positions")) token_map = weight_obs.get("token_position_to_phoneme") configured_fallback_count = _as_int(extractor_metadata.get("clip_extractor_fallback_count")) if configured_fallback_count > fallback_clip_count: fallback_clip_count = configured_fallback_count rank_values = [_as_float(value) for value in rank_util.values()] mean_rank_utilization = ( round(sum(rank_values) / len(rank_values), 4) if rank_values else None ) min_rank_layer = ( min(rank_util, key=lambda key: _as_float(rank_util[key])) if rank_util else None ) min_rank_value = _dict_min(rank_util) if rank_util else None worst_sparsity_layer = ( max(layer_sparsity, key=lambda key: _as_float(layer_sparsity[key])) if layer_sparsity else None ) worst_sparsity_value = ( _as_float(layer_sparsity.get(worst_sparsity_layer)) if worst_sparsity_layer is not None else None ) overfit_sig = _as_float(weight_obs.get("overfitting_signature")) mean_rank_for_assessment = ( sum(_as_float(v) for v in rank_util.values()) / max(len(rank_util), 1) if rank_util else 0.5 ) if overfit_sig >= 0.6: training_quality_assessment = "overfit" elif gradient_noise >= 0.5: training_quality_assessment = "unstable" elif mean_rank_for_assessment <= 0.3: training_quality_assessment = "underfit" else: training_quality_assessment = "healthy" digest: dict[str, Any] = { "case_id": bundle.get("case_id"), "prompt": str(bundle.get("prompt", "")), "param_config": _as_dict(bundle.get("param_config")), "ingestion_metadata": { "created_at_unix": metadata.get("created_at_unix"), "clip_extractor_fallback_count": fallback_clip_count, "fallback_clip_ids": sorted(fallback_clip_ids), }, } if image_obs: digest["image_summary"] = { "face_occupancy_ratio": _as_float(image_obs.get("face_occupancy_ratio")), "estimated_sharpness": _as_float(image_obs.get("estimated_sharpness")), "estimated_yaw_degrees": _as_float(image_obs.get("estimated_yaw_degrees")), "estimated_pitch_degrees": _as_float(image_obs.get("estimated_pitch_degrees")), "lighting_uniformity_score": _as_float(image_obs.get("lighting_uniformity_score")), "background_complexity_score": _as_float( image_obs.get("background_complexity_score") ), "occlusion_detected": _as_bool(image_obs.get("occlusion_detected")), "prompt_token_count": _as_int(image_obs.get("prompt_token_count")), "prompt_semantic_density": _as_float(image_obs.get("prompt_semantic_density")), "conflicting_descriptors": _as_list(image_obs.get("conflicting_descriptors")), "identity_anchoring_strength": _as_float( image_obs.get("identity_anchoring_strength") ), "face_detection_measured": _as_bool( extractor_metadata.get("face_detection_measured"), default=True, ), } if clip_obs: fallback_clips_note = ( "Clips extracted via fallback extractor use synthetic lip_sync_confidence=0.35 " "and empty phoneme_sequence; treat lip-sync and phoneme evidence as unreliable." if fallback_clip_count > 0 else None ) digest["clip_summary"] = { "clip_count": len(clip_obs), "fallback_clip_count": fallback_clip_count, "non_fallback_clip_count": max(0, len(clip_obs) - fallback_clip_count), "all_clips_fallback": fallback_clip_count == len(clip_obs), "parameter_tuning_reliability": ( "low" if fallback_clip_count == len(clip_obs) else "normal" ), "high_risk_clip_ids": high_risk_clip_ids, "high_risk_clips": high_risk_clips, "mean_identity_drift": _safe_mean(drifts), "mean_blur_score": _safe_mean(blurs), "mean_lip_sync_confidence": _safe_mean(lips), "mean_lip_sync_confidence_reliable_only": _safe_mean(reliable_lips), "quality_distribution": quality_distribution, "per_clip_signals": per_clip_signals, "per_clip_breakdown": per_clip_breakdown, "per_clip_signals_truncated": len(clip_obs) > max_per_clip_signals, "fallback_clips_note": fallback_clips_note, } if weight_obs: digest["weight_summary"] = { "available": True, "weight_file_id": str(weight_obs.get("weight_file_id", "")), "lora_rank": _as_int(weight_obs.get("lora_rank")), "target_module_count": len(_as_list(weight_obs.get("target_modules"))), "max_canonical_entropy": _dict_max(layer_entropy), "min_rank_utilization": _dict_min(rank_util), "high_entropy_token_positions": high_entropy_positions[:24], "suspected_anomalous_phonemes": _phonemes_from_positions( high_entropy_positions, token_map, ), "overfitting_signature": overfit_sig, "layer_rank_utilization_summary": { "mean": mean_rank_utilization, "min_layer": min_rank_layer, "min_value": min_rank_value, }, "worst_sparsity_layer": worst_sparsity_layer, "worst_sparsity_value": worst_sparsity_value, "gradient_noise_estimate": gradient_noise, "training_quality_assessment": training_quality_assessment, } return digest def _normalize_task_tier(task_tier: TaskTier) -> Literal["image_audit", "clip_audit", "weight_audit"]: mapped = { "easy": "image_audit", "medium": "clip_audit", "hard": "weight_audit", } return mapped.get(task_tier, task_tier) # type: ignore[return-value] def _build_prompt( signal_digest: dict[str, Any], *, task_tier: TaskTier = "weight_audit", ) -> str: _ = _normalize_task_tier(task_tier) fallback_note = ( _as_dict(signal_digest.get("clip_summary")).get("fallback_clips_note") if isinstance(signal_digest.get("clip_summary"), dict) else None ) fallback_warning = ( f"\n\nWARNING - FALLBACK EXTRACTOR ACTIVE: {fallback_note}" if isinstance(fallback_note, str) and fallback_note.strip() else "" ) return ( "Analyze the TalkingHeadBench signal digest below and produce a structured report.\n\n" "## Signal Thresholds (use these to classify findings)\n" "Identity drift severity: <0.05=none, <0.12=minor, <0.22=moderate, >=0.22=severe\n" "Lip sync quality: >=0.75=good, >=0.50=acceptable, >=0.20=poor, <0.20=absent\n" "Blur score: >=0.60=sharp, >=0.30=acceptable, <0.30=blurry\n" "Overfitting signature: <0.30=low, 0.30-0.59=medium, 0.60-0.79=high, >=0.80=critical\n" "Rank utilization mean: >=0.65=efficient, 0.30-0.64=wasteful, <0.30=collapsed\n" "Face occupancy ratio: >=0.20=good, 0.10-0.19=marginal, <0.10=face too small\n\n" "## Required Report Sections\n" "Write exactly these sections in order. Each section must cite specific digest values.\n\n" "### Overall Readiness\n" "1-2 sentences. State readiness level (ready / marginal / not ready) and the single most critical blocker with its measured value.\n\n" "### Critical Risks\n" "Bullet list. For each risk: name the affected signal, cite exact value, and classify severity with thresholds above. Minimum 1 bullet, maximum 5.\n\n" "### Parameter Fixes\n" "Directional only - no specific numbers. For each anomalous param in param_config, state direction to adjust and why, citing the signal that drives it. If param_config is reasonable, state that explicitly.\n\n" "### Data and Weight Concerns\n" "Per-clip breakdown: for each entry in per_clip_breakdown, state clip_id and risk_flags (or clean if none). Then summarize weight health using overfitting_signature, layer_rank_utilization_summary.mean, and worst_sparsity_layer.\n\n" "### Weight Status\n" "If weight_summary.available is false or absent, state 'Not uploaded - no data'. Otherwise state training quality assessment (overfit/healthy/underfit/unstable) based on training_quality_assessment and supporting fields.\n\n" "### Top 3 Next Actions\n" "Numbered list. Each action must be actionable, specific, and evidence-cited.\n\n" "### Confidence\n" "State confidence level (high/medium/low) and what limits it (fallback extractor, missing weights, or insufficient clips)." f"{fallback_warning}" "\n\n## Signal Digest\n" f"{json.dumps(signal_digest, indent=2, sort_keys=True)}" ) def _call_openai( *, model_id: str, api_key: str | None, prompt: str, max_tokens: int, temperature: float, timeout_s: float, base_url: str | None, ) -> str: key = _require_api_key("openai", api_key) base = _resolve_openai_base_url(base_url) return _call_chat_completions( provider="openai", base_url=base, model_id=model_id, api_key=key, prompt=prompt, max_tokens=max_tokens, temperature=temperature, timeout_s=timeout_s, ) def _call_anthropic( *, model_id: str, api_key: str | None, prompt: str, max_tokens: int, temperature: float, timeout_s: float, base_url: str | None, ) -> str: key = _require_api_key("anthropic", api_key) base = _resolve_anthropic_base_url(base_url) return _call_chat_completions( provider="anthropic", base_url=base, model_id=model_id, api_key=key, prompt=prompt, max_tokens=max_tokens, temperature=temperature, timeout_s=timeout_s, ) def _call_huggingface( *, model_id: str, api_key: str | None, prompt: str, max_tokens: int, temperature: float, timeout_s: float, base_url: str | None, ) -> str: key = _require_api_key("huggingface", api_key) routed_model = model_id if not base_url and ":" not in model_id.split("/")[-1]: routed_model = f"{model_id}:fastest" base = _resolve_huggingface_base_url(base_url) return _call_chat_completions( provider="huggingface", base_url=base, model_id=routed_model, api_key=key, prompt=prompt, max_tokens=max_tokens, temperature=temperature, timeout_s=timeout_s, ) def _call_local( *, model_id: str, api_key: str | None, prompt: str, max_tokens: int, temperature: float, timeout_s: float, base_url: str | None, ) -> str: key = (api_key or os.getenv("HF_TOKEN") or "local").strip() base = _resolve_local_base_url(base_url) return _call_chat_completions( provider="local", base_url=base, model_id=model_id, api_key=key, prompt=prompt, max_tokens=max_tokens, temperature=temperature, timeout_s=timeout_s, ) def _call_chat_completions( *, provider: str, base_url: str, model_id: str, api_key: str, prompt: str, max_tokens: int, temperature: float, timeout_s: float, ) -> str: client = OpenAI(api_key=api_key, base_url=base_url, timeout=timeout_s) try: response = client.chat.completions.create( model=model_id, messages=[ {"role": "system", "content": _SYSTEM_PROMPT}, {"role": "user", "content": prompt}, ], temperature=temperature, max_tokens=max_tokens, ) except APIStatusError as exc: status_code = int(exc.status_code or 502) message = _format_http_error(provider, status_code) raise LLMAdapterError( code="provider_http_error", message=message, status_code=502, retryable=status_code >= 500 or status_code == 429, ) from exc except APIConnectionError as exc: raise LLMAdapterError( code="provider_connection_error", message=f"Unable to reach {provider} provider endpoint.", status_code=502, retryable=True, ) from exc except Exception as exc: # noqa: BLE001 raise LLMAdapterError( code="provider_invalid_response", message=f"{provider} provider returned an unexpected response.", status_code=502, retryable=True, ) from exc choices = getattr(response, "choices", None) or [] if not choices: raise LLMAdapterError( code="invalid_provider_response", message="Provider response did not include choices.", status_code=502, retryable=True, ) content = choices[0].message.content text = _extract_text_content(content) if not text: raise LLMAdapterError( code="invalid_provider_response", message="Provider response did not include message content.", status_code=502, retryable=True, ) return text def _resolve_openai_base_url(base_url: str | None) -> str: url = ( base_url or os.getenv("THB_OPENAI_BASE_URL") or os.getenv("API_BASE_URL") or "https://api.openai.com/v1" ).rstrip("/") return _strip_chat_completions_suffix(url) def _resolve_anthropic_base_url(base_url: str | None) -> str: url = ( base_url or os.getenv("THB_ANTHROPIC_BASE_URL") or "https://api.anthropic.com/v1" ).rstrip("/") return _strip_chat_completions_suffix(url) def _resolve_huggingface_base_url(base_url: str | None) -> str: url = ( base_url or os.getenv("THB_HF_BASE_URL") or os.getenv("API_BASE_URL") or "https://router.huggingface.co/v1" ).rstrip("/") return _strip_chat_completions_suffix(url) def _resolve_local_base_url(base_url: str | None) -> str: raw = ( base_url or os.getenv("THB_LOCAL_LLM_URL") or os.getenv("API_BASE_URL") or "http://localhost:11434/v1" ).rstrip("/") url = _strip_chat_completions_suffix(raw) if url.endswith("/api/generate"): url = f"{url[:-len('/api/generate')]}/v1" if not url.endswith("/v1"): url = f"{url}/v1" return url def _strip_chat_completions_suffix(url: str) -> str: if url.endswith("/chat/completions"): return url[: -len("/chat/completions")] return url def _format_http_error(provider: str, status_code: int) -> str: if status_code in {401, 403}: return f"Authentication failed for {provider} provider." if status_code == 404: return f"Requested model or endpoint was not found for {provider}." if status_code == 429: return f"Rate limit exceeded for {provider} provider." if status_code >= 500: return f"{provider} provider is temporarily unavailable." return f"{provider} provider request failed." def _extract_text_content(content: Any) -> str: if isinstance(content, str): return content if isinstance(content, list): parts: list[str] = [] for item in content: if isinstance(item, str): parts.append(item) continue if isinstance(item, dict): text = item.get("text") or item.get("content") if isinstance(text, str): parts.append(text) return "\n".join(part for part in parts if part).strip() if isinstance(content, dict): text = content.get("text") or content.get("content") if isinstance(text, str): return text return "" def _max_clip_signals_limit() -> int: raw = os.getenv("THB_MAX_CLIP_SIGNALS_IN_DIGEST", "24").strip() try: value = int(raw) except ValueError: return 24 return max(1, value) def _drift_severity(drift: float) -> str: if drift < 0.05: return "none" if drift < 0.12: return "minor" if drift < 0.22: return "moderate" return "severe" def _clip_risk_reasons( *, drift: float, blur: float, lip_sync: float, lip_sync_reliable: bool, fallback_used: bool, ) -> tuple[list[str], list[str]]: high_risk_reasons: list[str] = [] warning_reasons: list[str] = [] if fallback_used: warning_reasons.append("fallback_extractor_used = true (proxy-only clip signals)") if drift >= _DRIFT_WARNING_THRESHOLD: warning_reasons.append("identity_cosine_drift proxy >= 0.22") if blur <= _BLUR_WARNING_THRESHOLD: warning_reasons.append("blur_score proxy <= 0.30") warning_reasons.append("lip_sync_confidence unavailable (fallback_extractor_used)") return high_risk_reasons, warning_reasons if drift >= _DRIFT_HIGH_RISK_THRESHOLD: high_risk_reasons.append("identity_cosine_drift >= 0.35") elif drift >= _DRIFT_WARNING_THRESHOLD: warning_reasons.append("identity_cosine_drift >= 0.22") if blur <= _BLUR_HIGH_RISK_THRESHOLD: high_risk_reasons.append("blur_score <= 0.20") elif blur <= _BLUR_WARNING_THRESHOLD: warning_reasons.append("blur_score <= 0.30") if lip_sync_reliable: if lip_sync <= _LIPSYNC_HIGH_RISK_THRESHOLD: high_risk_reasons.append("lip_sync_confidence <= 0.30") elif lip_sync <= _LIPSYNC_WARNING_THRESHOLD: warning_reasons.append("lip_sync_confidence <= 0.40") else: warning_reasons.append("lip_sync_confidence unavailable (fallback_extractor_used)") return high_risk_reasons, warning_reasons def _clip_quality_tier( *, drift: float, blur: float, lip_sync: float, lip_sync_reliable: bool, fallback_used: bool, high_risk_reason_count: int, ) -> str: if fallback_used: if drift < _DRIFT_WARNING_THRESHOLD and blur >= 0.12: return "acceptable" if drift < _DRIFT_HIGH_RISK_THRESHOLD and blur >= 0.08: return "marginal" return "poor" if high_risk_reason_count >= 2: return "poor" if high_risk_reason_count == 1: return "marginal" is_good = drift < 0.12 and blur > 0.40 and (lip_sync > 0.50 or not lip_sync_reliable) if is_good: return "good" if lip_sync_reliable else "acceptable" is_acceptable = drift < _DRIFT_WARNING_THRESHOLD and blur > 0.25 if is_acceptable and (lip_sync > _LIPSYNC_WARNING_THRESHOLD or not lip_sync_reliable): return "acceptable" is_marginal = drift < _DRIFT_HIGH_RISK_THRESHOLD and blur > _BLUR_HIGH_RISK_THRESHOLD if is_marginal and (lip_sync > _LIPSYNC_HIGH_RISK_THRESHOLD or not lip_sync_reliable): return "marginal" return "poor" def _as_dict(value: Any) -> dict[str, Any]: return value if isinstance(value, dict) else {} def _as_list(value: Any) -> list[Any]: return value if isinstance(value, list) else [] def _as_list_of_dicts(value: Any) -> list[dict[str, Any]]: if not isinstance(value, list): return [] return [item for item in value if isinstance(item, dict)] def _as_float(value: Any, *, default: float = 0.0) -> float: if isinstance(value, (int, float)): return float(value) return default def _as_int(value: Any, *, default: int = 0) -> int: if isinstance(value, bool): return default if isinstance(value, int): return value if isinstance(value, float): return int(value) return default def _as_bool(value: Any, *, default: bool = False) -> bool: if isinstance(value, bool): return value if isinstance(value, str): cleaned = value.strip().lower() if cleaned in {"1", "true", "yes", "on"}: return True if cleaned in {"0", "false", "no", "off"}: return False return default def _safe_mean(values: list[float]) -> float: clean = [value for value in values if isinstance(value, (int, float))] if not clean: return 0.0 return round(float(sum(clean) / len(clean)), 4) def _dict_max(values: dict[str, Any]) -> float: if not values: return 0.0 return max(_as_float(v) for v in values.values()) def _dict_min(values: dict[str, Any]) -> float: if not values: return 0.0 return min(_as_float(v) for v in values.values()) def _phonemes_from_positions( positions: list[Any], token_position_to_phoneme: Any, ) -> list[str]: if not isinstance(token_position_to_phoneme, dict): return [] mapped: list[str] = [] for position in positions: idx = _as_int(position, default=-1) if idx < 0: continue key = str(idx) value = token_position_to_phoneme.get(key) if value is None and idx in token_position_to_phoneme: value = token_position_to_phoneme.get(idx) if isinstance(value, str) and value not in mapped: mapped.append(value) if len(mapped) >= 16: break return mapped