Spaces:

infoclassifier
/

InfoClassifier

Sleeping

UTKARSH MISHRA
Initial commit: InfoClassifier project
e5e36db 7 months ago
24.7 kB
"""
inference.py
============
Baseline ReAct-style agent for InformationIntegrityEnv.
Two run modes
-------------
benchmark   Run all difficulties over all scenarios (default).
            Prints per-episode [START]/[STEP]/[END] lines, then a final
            summary table and tier-separation check.
single      Run one episode against a specific scenario or a live server.
Environment variables
---------------------
  API_BASE_URL      URL of the LLM provider (OpenAI-compatible)
  MODEL_NAME        Model to use (e.g. gpt-4o, claude-3-5-sonnet, etc.)
  HF_TOKEN          Bearer token / API key for the provider
  ENV_SERVER_URL    URL of the env server (default http://localhost:8000)
  MAX_STEPS         Per-episode step limit (default 20)
  BENCHMARK_SEED    Random seed for reproducible scenario sampling (default 42)
Logging protocol (required by OpenEnv)
---------------------------------------
  [START]  — once at episode start
  [STEP]   — at every step
  [END]    — with final score
Benchmark summary format (required)
-------------------------------------
  difficulty=easy   avg_score=X.XXXX  avg_reward=X.XXXX  n=N
  difficulty=medium avg_score=X.XXXX  avg_reward=X.XXXX  n=N
  difficulty=hard   avg_score=X.XXXX  avg_reward=X.XXXX  n=N
Usage
-----
  # Full benchmark (local, all scenarios)
  python inference.py --benchmark
  # Single episode from dataset
  python inference.py --local --difficulty medium --scenario-id 0
  # Single episode from JSON file
  python inference.py --local --scenario scenario.json
  # Against running env server
  python inference.py --server --difficulty easy --scenario-id 1
  # Write benchmark results to JSON
  python inference.py --benchmark --out results.json
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import random
import sys
import time
from typing import Any, Dict, List, Optional

try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    pass  # rely on shell environment

import requests as http_requests
from openai import OpenAI

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger("inference")

# ---------------------------------------------------------------------------
# Config from environment
# ---------------------------------------------------------------------------

API_BASE_URL    = os.getenv("API_BASE_URL",    "https://router.huggingface.co/v1")
MODEL_NAME      = os.getenv("MODEL_NAME",      "Qwen/Qwen2.5-72B-Instruct")
HF_TOKEN        = os.getenv("HF_TOKEN")
ENV_SERVER_URL  = os.getenv("ENV_SERVER_URL",  "http://localhost:8000")
MAX_STEPS       = int(os.getenv("MAX_STEPS",   10))
BENCHMARK_SEED  = int(os.getenv("BENCHMARK_SEED", 42))
DIFFICULTY      = os.getenv("DIFFICULTY",      "easy")
SCENARIO_ID     = os.getenv("SCENARIO_ID",     None)   # int index or string scenario id

DIFFICULTIES    = ["easy", "medium", "hard"]


# ---------------------------------------------------------------------------
# LLM client
# ---------------------------------------------------------------------------

def _build_client() -> OpenAI:
    if not HF_TOKEN:
        raise ValueError(
            "No API key found. Set HF_TOKEN environment variable."
        )
    return OpenAI(api_key=HF_TOKEN, base_url=API_BASE_URL)


# ---------------------------------------------------------------------------
# Environment interface — local (in-process)
# ---------------------------------------------------------------------------

class LocalEnvClient:
    """Runs InformationIntegrityEnv in-process against the static dataset."""

    def __init__(
        self,
        difficulty: str,
        scenario_id: Optional[str | int] = None,
        scenario_dict: Optional[dict] = None,
    ) -> None:
        from environment import InformationIntegrityEnv

        if scenario_dict is not None:
            self._env = InformationIntegrityEnv(scenario=scenario_dict)
        else:
            self._env = InformationIntegrityEnv.from_dataset(
                difficulty=difficulty,
                scenario_id=scenario_id,
            )
        self._done = False

    def reset(self, difficulty: Optional[str] = None, scenario_id: Optional[str | int] = None) -> dict:
        obs = self._env.reset(difficulty=difficulty, scenario_id=scenario_id)
        self._done = False
        return obs.model_dump()

    def step(self, action: dict) -> dict:
        from models import Action
        act = Action(**action)
        obs, rew, done, info = self._env.step(act)
        self._done = done
        return {
            "observation": obs.model_dump(),
            "reward":      rew.model_dump(),
            "done":        done,
            "info":        info,
        }

    @property
    def done(self) -> bool:
        return self._done

    def episode_result(self) -> dict:
        return self._env.episode_result().model_dump()


# ---------------------------------------------------------------------------
# Environment interface — remote server
# ---------------------------------------------------------------------------

class ServerEnvClient:
    """Communicates with a running FastAPI env server."""

    def __init__(self, base_url: str = ENV_SERVER_URL) -> None:
        self._base = base_url.rstrip("/")
        self._session_id: Optional[str] = None

    def reset(
        self,
        difficulty: Optional[str] = None,
        scenario_id: Optional[str | int] = None,
        scenario_dict: Optional[dict] = None,
    ) -> dict:
        payload: dict = {"max_steps": MAX_STEPS}
        if scenario_dict:
            payload["scenario"] = scenario_dict
        elif difficulty:
            payload["difficulty"] = difficulty
            if scenario_id is not None:
                payload["scenario_id"] = scenario_id
        r = http_requests.post(f"{self._base}/reset", json=payload, timeout=30)
        r.raise_for_status()
        data = r.json()
        self._session_id = data["session_id"]
        return data["observation"]

    def step(self, action: dict) -> dict:
        payload = {"session_id": self._session_id, "action": action}
        r = http_requests.post(f"{self._base}/step", json=payload, timeout=30)
        r.raise_for_status()
        return r.json()

    def episode_result(self) -> dict:
        # Server does not expose a dedicated episode_result endpoint;
        # grader_score is included in the final step's info dict.
        return {}


# ---------------------------------------------------------------------------
# System prompt
# ---------------------------------------------------------------------------

SYSTEM_PROMPT = SYSTEM_PROMPT = """You are a high-precision content moderation agent.
Your goal is to make the MOST ACCURATE moderation decision based on evidence.
You MUST follow a structured reasoning process:
1. First verify claims.
2. Then evaluate sources carefully:
   - Prefer HIGH reliability sources
   - Consider stance (support/refute/neutral)
3. Avoid redundant actions.
4. Only decide when evidence is sufficient.
Decision guidelines:
- REMOVE → if content is clearly false and harmful
- FLAG   → if uncertain, conflicting evidence, or partially misleading
- ALLOW  → if claims are well-supported by reliable sources
Action rules:
- You MUST verify at least one claim
- You MUST check at least two sources
- NEVER repeat the same action
- Be efficient (avoid unnecessary steps)
Output STRICT JSON ONLY:
{"type": "...", ...}
"""


def _observation_prompt(obs_dict: dict) -> str:
    content = obs_dict.get("content", "")
    claims  = obs_dict.get("claims", [])
    sources = obs_dict.get("available_sources", [])
    history = obs_dict.get("interaction_history", [])
    flags   = obs_dict.get("status_flags", {})
    step    = obs_dict.get("step_count", 0)

    # Track what has already been done from full history
    checked_source_ids  = {
        h["action_payload"].get("source_id")
        for h in history
        if h.get("action_type") == "check_source" and h.get("action_payload", {}).get("source_id")
    }
    verified_claim_ids  = {
        h["action_payload"].get("claim_id")
        for h in history
        if h.get("action_type") == "verify_claim" and h.get("action_payload", {}).get("claim_id")
    }

    claims_text = "\n".join(
        f"  [{c['id']}] {c['text']} — status: {c.get('verification_status','unverified')}"
        + (" [ALREADY VERIFIED — do not repeat]" if c['id'] in verified_claim_ids else "")
        for c in claims
    )
    sources_text = "\n".join(
        f"  [{s['id']}] {s['title']} | stance={s['stance']} | reliability={s['reliability']:.2f}"
        + (" [ALREADY CHECKED — do not repeat]" if s['id'] in checked_source_ids else "")
        for s in sources
    )

    last_result = ""
    if history:
        last = history[-1]
        last_result = f"\nLast step result: {json.dumps(last.get('result', {}), ensure_ascii=False)[:300]}"

    unchecked_sources  = [s['id'] for s in sources  if s['id']  not in checked_source_ids]
    unverified_claims  = [c['id'] for c in claims   if c['id']  not in verified_claim_ids]
    sufficient         = flags.get('sufficient_evidence', False)

    next_step_hint = ""
    if not sufficient:
        if unverified_claims:
            next_step_hint = f"\nHINT: You still need to verify claims: {unverified_claims}. Use verify_claim next."
        elif unchecked_sources:
            next_step_hint = f"\nHINT: You still need to check sources: {unchecked_sources}. Use check_source next."
    else:
        next_step_hint = "\nHINT: You have sufficient evidence. Use the 'decide' action now (allow/flag/remove)."

    return f"""STEP {step}
CONTENT UNDER REVIEW:
{content}
CLAIMS:
{claims_text}
AVAILABLE SOURCES:
{sources_text}
STATUS FLAGS:
  decision_made={flags.get('decision_made', False)}
  sufficient_evidence={sufficient}
  redundant_actions={flags.get('redundant_actions', 0)}
{last_result}{next_step_hint}
Choose your next action (JSON only):"""


# ---------------------------------------------------------------------------
# LLM action call
# ---------------------------------------------------------------------------

def _llm_action(llm: OpenAI, messages: list, step: int, retries: int = 3) -> dict:
    """Call LLM and parse JSON action. Retries on parse failure."""
    last_exc = None
    for attempt in range(retries):
        try:
            resp = llm.chat.completions.create(
                model=MODEL_NAME,
                messages=messages,
                temperature=0.0,
                max_tokens=256,
                timeout=30,
            )
            raw = resp.choices[0].message.content.strip()
            if raw.startswith("```"):
                raw = raw.split("\n", 1)[-1].rsplit("```", 1)[0].strip()
            return json.loads(raw)
        except json.JSONDecodeError as exc:
            logger.warning("Step %d attempt %d: JSON parse error — %s", step, attempt + 1, exc)
            last_exc = exc
        except Exception as exc:
            logger.error("LLM call failed at step %d: %s", step, exc)
            last_exc = exc
            if attempt < retries - 1:
                time.sleep(2 ** attempt)

    # Raise so the episode fails loudly rather than silently looping
    raise RuntimeError(f"LLM call failed after {retries} attempts at step {step}: {last_exc}")


# ---------------------------------------------------------------------------
# Explanation builder
# ---------------------------------------------------------------------------

def _build_explanation(obs_dict: dict, decision: Optional[str]) -> str:
    try:
        from explainability import ExplainabilityEngine, build_reasoning_from_episode
        from models import Observation
        obs = Observation(**obs_dict)
        reasoning = build_reasoning_from_episode(obs=obs, decision=decision or "none")
        return ExplainabilityEngine().generate_explanation(reasoning)
    except Exception as exc:
        return f"(explanation unavailable: {exc})"


# ---------------------------------------------------------------------------
# OpenEnv-compliant logging helpers
# ---------------------------------------------------------------------------

def log_start(task: str, env: str, model: str) -> None:
    print(f"[START] task={task} env={env} model={model}", flush=True)


def log_step(step: int, action: str, reward: float, done: bool, error: str = "null") -> None:
    print(
        f"[STEP] step={step} action={action} reward={reward:.2f} "
        f"done={str(done).lower()} error={error}",
        flush=True,
    )


def log_end(success: bool, steps: int, rewards: List[float], score: float = 0.0) -> None:
    rewards_str = ",".join(f"{r:.2f}" for r in rewards)
    print(
        f"[END] success={str(success).lower()} steps={steps} score={score:.4f} rewards={rewards_str}",
        flush=True,
    )


# ---------------------------------------------------------------------------
# OpenEnv single-episode runner (primary entry point)
# ---------------------------------------------------------------------------

def run_single_episode(llm: OpenAI) -> dict:
    """
    Run one episode controlled entirely by DIFFICULTY and SCENARIO_ID env vars.
    Prints strict OpenEnv-format logs: [START], [STEP]×n, [END].
    """
    from tasks.registry import TASKS
    from environment import InformationIntegrityEnv

    # Resolve scenario_id: prefer int index, fall back to string id
    sid: Optional[str | int] = SCENARIO_ID
    if sid is not None:
        try:
            sid = int(sid)
        except (ValueError, TypeError):
            pass  # keep as string scenario id

    task         = TASKS[DIFFICULTY]
    scenario_dict = task.get_scenario(sid if sid is not None else 0)
    task_name = scenario_dict.get('id', f'{DIFFICULTY}_unknown')

    env = InformationIntegrityEnv(scenario=scenario_dict)
    obs = env.reset(difficulty=DIFFICULTY, scenario_id=sid)

    log_start(task=task_name, env="information-integrity-env", model=MODEL_NAME)

    messages: List[Dict[str, str]] = [{"role": "system", "content": SYSTEM_PROMPT}]
    rewards: List[float] = []
    steps_taken = 0
    score       = 0.0

    for step in range(1, MAX_STEPS + 1):
        obs_dict = obs.model_dump()
        messages.append({"role": "user", "content": _observation_prompt(obs_dict)})

        try:
            action_dict = _llm_action(llm, messages, step)
        except RuntimeError as exc:
            logger.error("Aborting episode: %s", exc)
            break
        messages.append({"role": "assistant", "content": json.dumps(action_dict)})

        error_str   = "null"
        step_reward = 0.0
        done        = False

        try:
            from models import Action
            obs, reward_obj, done, info = env.step(Action(**action_dict))
            step_reward = reward_obj.score
        except Exception as exc:
            error_str = str(exc)[:80].replace("\n", " ")

        rewards.append(step_reward)
        steps_taken = step

        log_step(
            step=step,
            action=action_dict.get("type", "unknown"),
            reward=step_reward,
            done=done,
            error=error_str,
        )

        if done:
            try:
                ep_result = env.episode_result()
                score = float(ep_result.grader_score or 0.0)
            except Exception:
                score = 0.0
            score = min(max(score, 0.005), 0.995)
            break

    # Compute grader score even if episode didn't complete (LLM failed / max steps)
    if score == 0.0:
        try:
            ep_result = env.episode_result()
            score = float(ep_result.grader_score or 0.0)
        except Exception:
            score = 0.0
    score = min(max(score, 0.005), 0.995)  # strictly in (0, 1) as required by validator

    success = score >= 0.5
    log_end(success=success, steps=steps_taken, rewards=rewards, score=score)

    return {"score": score, "steps": steps_taken, "success": success}


# ---------------------------------------------------------------------------
# Single episode runner (legacy — used by benchmark loop and CLI)
# ---------------------------------------------------------------------------

def run_episode(
    env: LocalEnvClient | ServerEnvClient,
    llm: OpenAI,
    difficulty: Optional[str] = None,
    scenario_id: Optional[str | int] = None,
) -> dict:
    """Run one full episode using compliant log_start/log_step/log_end."""
    messages: List[Dict[str, str]] = [{"role": "system", "content": SYSTEM_PROMPT}]

    obs_dict = env.reset(difficulty=difficulty, scenario_id=scenario_id)

    task_name = f"{difficulty}_{scenario_id}" if difficulty and scenario_id is not None \
        else obs_dict.get("task_difficulty", "unknown")
    log_start(task=task_name, env="information-integrity-env", model=MODEL_NAME)

    rewards: List[float] = []
    step        = 0
    grader_score: Optional[float] = None

    for step in range(1, MAX_STEPS + 1):
        messages.append({"role": "user", "content": _observation_prompt(obs_dict)})

        try:
            action_dict = _llm_action(llm, messages, step)
        except RuntimeError as exc:
            break
        messages.append({"role": "assistant", "content": json.dumps(action_dict)})

        error_str   = "null"
        step_reward = 0.0
        done        = False

        try:
            result      = env.step(action_dict)
            obs_dict    = result["observation"]
            step_reward = result["reward"].get("score", 0.0)
            done        = result["done"]
        except Exception as exc:
            error_str = str(exc)[:80].replace("\n", " ")

        rewards.append(step_reward)
        log_step(step=step, action=action_dict.get("type", "unknown"),
                 reward=step_reward, done=done, error=error_str)

        if done:
            break

    if isinstance(env, LocalEnvClient):
        try:
            grader_score = env.episode_result().get("grader_score")
        except Exception:
            pass

    score   = max(float(grader_score or 0.005), 0.005)  # strictly > 0 as required by validator
    success = score >= 0.5
    log_end(success=success, steps=step, rewards=rewards, score=score)

    return {
        "difficulty":    difficulty,
        "scenario_id":   scenario_id,
        "steps":         step,
        "grader_score":  score,      # always a float >= 0.005
        "total_reward": sum(rewards),
    }



# ---------------------------------------------------------------------------
# Benchmark loop
# ---------------------------------------------------------------------------

def run_benchmark(llm: OpenAI, out_path: Optional[str] = None) -> Dict[str, Any]:
    """
    Loop over all difficulties × all scenarios in each dataset.
    Prints per-episode logs, then the required summary table.
    Returns
    -------
    dict with per-difficulty aggregates and all raw episode results.
    """
    from tasks.registry import TASKS

    all_results: Dict[str, List[dict]] = {d: [] for d in DIFFICULTIES}
    rng = random.Random(BENCHMARK_SEED)

    for difficulty in DIFFICULTIES:
        task       = TASKS[difficulty]
        scenarios  = task.get_all_scenarios()
        n          = len(scenarios)
        logger.info("=== Benchmarking difficulty=%s  (%d scenarios) ===", difficulty, n)

        for idx, sc_dict in enumerate(scenarios):
            # Build a fresh local env for each scenario
            env = LocalEnvClient(
                difficulty=difficulty,
                scenario_id=idx,
            )
            result = run_episode(
                env=env,
                llm=llm,
                difficulty=difficulty,
                scenario_id=idx,
            )
            result["scenario_index"] = idx
            result["scenario_id"]    = sc_dict.get("id", str(idx))
            all_results[difficulty].append(result)

    # ---- Summary ----
    summary: Dict[str, dict] = {}
    print()  # blank line before summary
    for difficulty in DIFFICULTIES:
        episodes = all_results[difficulty]
        n = len(episodes)

        scores  = [e["grader_score"] for e in episodes]  # always float >= 0.005
        rewards = [e["total_reward"]  for e in episodes]

        avg_score  = sum(scores) / len(scores) if scores else 0.005
        avg_reward = sum(rewards) / len(rewards) if rewards else 0.0

        summary[difficulty] = {
            "avg_score":  round(avg_score,  4),
            "avg_reward": round(avg_reward, 4),
            "n":          n,
        }
        print(
            f"difficulty={difficulty:<6} "
            f"avg_score={avg_score:.4f}  "
            f"avg_reward={avg_reward:.4f}  "
            f"n={n}"
        )

    # ---- Difficulty tier separation check ----
    easy_s   = summary["easy"]["avg_score"]
    medium_s = summary["medium"]["avg_score"]
    hard_s   = summary["hard"]["avg_score"]

    if easy_s > medium_s > hard_s:
        print("Difficulty tiers well separated: easy > medium > hard ✓")
    else:
        print(
            "WARNING: Difficulty tiers not well separated "
            f"(easy={easy_s:.4f}, medium={medium_s:.4f}, hard={hard_s:.4f})"
        )

    benchmark_output = {
        "summary":  summary,
        "episodes": all_results,
    }

    if out_path:
        with open(out_path, "w", encoding="utf-8") as fh:
            json.dump(benchmark_output, fh, indent=2, ensure_ascii=False)
        logger.info("Benchmark results written to %s", out_path)

    return benchmark_output


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="InformationIntegrity benchmark agent",
        formatter_class=argparse.RawTextHelpFormatter,
    )

    mode = parser.add_mutually_exclusive_group()
    mode.add_argument(
        "--benchmark", action="store_true",
        help="Run full benchmark over all difficulties and all scenarios (default mode)",
    )
    mode.add_argument(
        "--local", action="store_true",
        help="Run a single episode in-process (requires --difficulty or --scenario)",
    )
    mode.add_argument(
        "--server", action="store_true",
        help="Run a single episode against a running env server",
    )

    parser.add_argument("--difficulty", default="medium",
                        choices=DIFFICULTIES,
                        help="Difficulty tier (used with --local or --server)")
    parser.add_argument("--scenario-id", default=None,
                        help="Scenario index (int) or id (str); None = random")
    parser.add_argument("--scenario", default=None,
                        help="Path to a pre-collected scenario JSON file (overrides --difficulty)")
    parser.add_argument("--out", default=None,
                        help="Write episode/benchmark result JSON to this path")

    return parser.parse_args()


def main() -> None:
    args   = _parse_args()
    llm    = _build_client()

    # ---------- Default: OpenEnv single-episode mode (env-var driven) ----------
    if not args.benchmark and not args.local and not args.server:
        run_single_episode(llm)
        return

    # ---------- Benchmark ----------
    if args.benchmark:
        run_benchmark(llm, out_path=args.out)
        return

    # ---------- Single episode ----------
    scenario_id: Optional[str | int] = args.scenario_id
    if scenario_id is not None:
        try:
            scenario_id = int(scenario_id)
        except (ValueError, TypeError):
            pass  # keep as string id

    # Load scenario from file if specified
    scenario_dict: Optional[dict] = None
    if args.scenario:
        with open(args.scenario, "r", encoding="utf-8") as fh:
            scenario_dict = json.load(fh)

    if args.local:
        env = LocalEnvClient(
            difficulty=args.difficulty,
            scenario_id=scenario_id,
            scenario_dict=scenario_dict,
        )
    else:
        env = ServerEnvClient(base_url=ENV_SERVER_URL)

    result = run_episode(
        env=env,
        llm=llm,
        difficulty=args.difficulty,
        scenario_id=scenario_id,
    )

    if args.out:
        with open(args.out, "w", encoding="utf-8") as fh:
            json.dump(result, fh, indent=2, ensure_ascii=False)
        logger.info("Episode result written to %s", args.out)


if __name__ == "__main__":
    main()