Spaces:
Paused
Paused
File size: 4,372 Bytes
4e75170 cf54850 4e75170 1e24aab 604836a 4e75170 1e24aab 604836a 1e24aab cf54850 1e24aab 604836a 4e75170 cf54850 4e75170 604836a 4e75170 cf54850 4e75170 604836a 4e75170 604836a 4e75170 604836a 4e75170 604836a cf54850 4e75170 cf54850 4e75170 604836a cf54850 4e75170 604836a cf54850 604836a 4e75170 604836a cf54850 604836a cf54850 4e75170 604836a 4e75170 cf54850 4e75170 604836a 4e75170 cf54850 4e75170 604836a 4e75170 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | from __future__ import annotations
import logging
import os
from typing import Any
from src.types import DetectionResponse, EngineResult
logger = logging.getLogger(__name__)
SYSTEM_INSTRUCTION = (
"You are a deepfake forensics analyst writing reports for security professionals. "
"Given detection engine outputs, write exactly 2-3 sentences in plain English "
"explaining why the content is real or fake. "
"Be specific and name the strongest signals. "
"Use direct declarative sentences. "
"Output only the explanation text."
)
DEFAULT_MODEL_CANDIDATES = (
"gemini-3.1-flash-lite-preview",
"gemini-2.5-flash",
)
_configured_candidates = [
value.strip()
for value in os.environ.get("GEMINI_MODEL_CANDIDATES", "").split(",")
if value.strip()
]
MODEL_CANDIDATES = (
tuple(_configured_candidates)
if _configured_candidates
else DEFAULT_MODEL_CANDIDATES
)
REQUEST_TIMEOUT_S = float(os.environ.get("GEMINI_REQUEST_TIMEOUT_S", "20"))
MAX_MODEL_ATTEMPTS = max(1, int(os.environ.get("GEMINI_MAX_MODEL_ATTEMPTS", "3")))
TEMPERATURE = float(os.environ.get("GEMINI_EXPLAIN_TEMPERATURE", "0.3"))
TOP_P = float(os.environ.get("GEMINI_EXPLAIN_TOP_P", "0.95"))
MAX_TOKENS = int(os.environ.get("GEMINI_EXPLAIN_MAX_TOKENS", "300"))
_client: Any | None = None
def _get_api_key() -> str:
return os.environ.get("GEMINI_API_KEY", "").strip()
def _get_client():
global _client
if _client is not None:
return _client
api_key = _get_api_key()
if not api_key:
raise RuntimeError("GEMINI_API_KEY is not configured")
try:
from google import genai
except Exception as exc:
raise RuntimeError("google-genai package is not installed") from exc
_client = genai.Client(api_key=api_key)
return _client
def _generate(prompt: str) -> str:
client = _get_client()
last_error: Exception | None = None
try:
from google.genai import types
except Exception as exc:
raise RuntimeError("google-genai types module is unavailable") from exc
for model_name in MODEL_CANDIDATES[:MAX_MODEL_ATTEMPTS]:
try:
response = client.models.generate_content(
model=model_name,
contents=prompt,
config=types.GenerateContentConfig(
system_instruction=SYSTEM_INSTRUCTION,
temperature=TEMPERATURE,
top_p=TOP_P,
max_output_tokens=MAX_TOKENS,
),
)
content = getattr(response, "text", None)
if content and content.strip():
logger.info("Gemini explain model selected: %s", model_name)
return content.strip()
except Exception as exc:
last_error = exc
logger.debug("Gemini explain model %s failed: %s", model_name, exc)
if last_error is not None:
raise last_error
raise RuntimeError("No Gemini model candidates succeeded")
def explain(
verdict: str,
confidence: float,
engine_results: list[EngineResult],
generator: str,
) -> str:
breakdown = "\n".join(
f"- {result.engine}: {result.verdict} ({result.confidence:.0%}) - {result.explanation}"
for result in engine_results
)
prompt = (
f"Verdict: {verdict} ({confidence:.0%} confidence)\n"
f"Attributed generator: {generator}\n"
f"Engine breakdown:\n{breakdown}\n\n"
"Write the forensics explanation."
)
try:
return _generate(prompt)
except Exception as exc:
logger.error("Gemini explain failed: %s", exc)
top = engine_results[0] if engine_results else None
primary = f"Primary signal came from the {top.engine} engine." if top else ""
return (
f"Content classified as {verdict} with {confidence:.0%} confidence. "
f"Attributed generator: {generator}. "
f"{primary}"
).strip()
class Explainer:
"""Compatibility wrapper for legacy callers expecting an object API."""
def explain(self, response: DetectionResponse) -> str:
return explain(
verdict=response.verdict,
confidence=response.confidence,
engine_results=response.engine_breakdown,
generator=response.attributed_generator,
)
|