| """ |
| Simulate judge evaluation against the live HF Space with LLM agent. |
| Connects to the deployed Space and runs all tasks using OpenAI API. |
| """ |
| import asyncio |
| import json |
| import os |
| import sys |
| import time |
| from typing import List, Optional |
|
|
| from dotenv import load_dotenv |
| from openai import OpenAI |
|
|
| from client import InvoiceGuardEnv |
| from models import ( |
| ActionType, DecisionType, ExceptionType, InvoiceGuardAction, |
| ) |
|
|
| load_dotenv() |
|
|
| SPACE_URL = "https://piyush-mk-invoice-guard.hf.space" |
| API_BASE_URL = os.getenv("API_BASE_URL", "https://api.openai.com/v1") |
| MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4.1-mini") |
| API_KEY = os.getenv("HF_TOKEN") or os.getenv("API_KEY") or os.getenv("OPENAI_API_KEY") or "" |
|
|
| TASKS = [ |
| "task_1_clean_match", |
| "task_2_partial_receipt", |
| "task_3_price_variance", |
| "task_4_duplicate_invoice", |
| "task_5_mixed_discrepancy", |
| "task_6_false_positive_duplicate", |
| "task_7_retroactive_price", |
| "task_8_split_invoice_pattern", |
| "task_9_clean_from_risky_vendor", |
| "task_10_rounding_false_alarm", |
| "task_11_authorized_overship", |
| "task_12_corrected_resubmission", |
| ] |
|
|
| _MODELS_MAX_COMPLETION_TOKENS = {"gpt-5.4", "gpt-5.4-mini", "gpt-5.4-nano", "gpt-5", "gpt-5-mini", "gpt-5.1"} |
|
|
| def _tok_kwarg(limit=512): |
| for p in _MODELS_MAX_COMPLETION_TOKENS: |
| if MODEL_NAME.startswith(p): |
| return {"max_completion_tokens": limit} |
| return {"max_tokens": limit} |
|
|
| SYSTEM_PROMPT = """You are a senior accounts payable analyst. You will be given an invoice case to investigate and resolve. |
| |
| The environment tells you your goal, available actions, and decision options. Read the goal carefully. |
| |
| WORKFLOW: |
| 1. Investigate: inspect documents (PO, GRN, vendor profile, policy rules), run comparisons (quantity, price, totals), check for duplicates. |
| 2. Resolve: submit_final_resolution with your decision, exception type, evidence references, and explanation. |
| |
| Complete a thorough investigation before resolving. Inspect at least: purchase order, goods receipt note, compare quantity, compare price, policy rules, duplicate check, and vendor profile. |
| |
| RESPONSE FORMAT: |
| - Respond with ONLY a valid JSON object. No markdown, no commentary. |
| - Investigation example: {"action_type": "inspect_purchase_order"} |
| - Resolution example: {"action_type": "submit_final_resolution", "final_decision": "approve_for_payment", "exception_type": "clean_match", "evidence_references": ["inspect_purchase_order", "compare_quantity"], "explanation": "All documents match within tolerance."} |
| |
| RULES: |
| - Pay close attention to POLICY findings -- they tell you when escalation is required. |
| - When multiple issues exist, escalation takes priority over hold. |
| - Check PO references carefully before concluding an invoice is a duplicate. |
| - Include all investigation actions you performed in evidence_references. |
| - Cite specific numbers in your explanation. |
| - NEVER repeat an action you already took. |
| - When remaining_steps is 3 or fewer, submit immediately with what you have. |
| """ |
|
|
|
|
| def build_observation_prompt(obs, is_first=False): |
| parts = [ |
| f"Case: {obs.case_id} | Difficulty: {obs.difficulty} | Steps remaining: {obs.remaining_steps}", |
| f"Invoice: {obs.invoice_summary}", |
| ] |
| if is_first and obs.goal: |
| parts.append(f"\n{obs.goal}") |
| if obs.revealed_documents: |
| parts.append(f"Documents reviewed: {', '.join(obs.revealed_documents)}") |
| if obs.findings: |
| parts.append("Findings:") |
| for i, f in enumerate(obs.findings, 1): |
| parts.append(f" {i}. {f}") |
| if obs.last_action_result: |
| parts.append(f"Last result: {obs.last_action_result}") |
| if obs.warnings: |
| parts.append(f"Warnings: {'; '.join(obs.warnings)}") |
| if obs.remaining_steps <= 2: |
| parts.append(">>> YOU MUST submit_final_resolution NOW. Decide based on what you have. <<<") |
| return "\n".join(parts) |
|
|
|
|
| def parse_llm_response(text): |
| text = text.strip() |
| if "```json" in text: |
| text = text.split("```json")[1].split("```")[0].strip() |
| elif "```" in text: |
| text = text.split("```")[1].split("```")[0].strip() |
| try: |
| return json.loads(text) |
| except json.JSONDecodeError: |
| pass |
| for line in text.split("\n"): |
| line = line.strip() |
| if line.startswith("{"): |
| try: |
| return json.loads(line) |
| except json.JSONDecodeError: |
| continue |
| return {"action_type": "summarize_findings"} |
|
|
|
|
| def build_action(params): |
| action_type = params.get("action_type", "summarize_findings") |
| try: |
| ActionType(action_type) |
| except ValueError: |
| action_type = "summarize_findings" |
| kwargs = {"action_type": action_type} |
| if params.get("final_decision"): |
| try: |
| kwargs["final_decision"] = DecisionType(params["final_decision"]) |
| except ValueError: |
| pass |
| if params.get("exception_type"): |
| try: |
| kwargs["exception_type"] = ExceptionType(params["exception_type"]) |
| except ValueError: |
| pass |
| if params.get("evidence_references"): |
| kwargs["evidence_references"] = list(params["evidence_references"]) |
| if params.get("explanation"): |
| kwargs["explanation"] = str(params["explanation"]) |
| return InvoiceGuardAction(**kwargs) |
|
|
|
|
| async def run_task(env, llm, task_id): |
| result = await env.reset(task_id=task_id) |
| obs = result.observation |
| obs.done = result.done |
| obs.reward = result.reward |
| messages = [{"role": "system", "content": SYSTEM_PROMPT}] |
| steps = 0 |
| rewards = [] |
|
|
| print(f" [RESET] case={obs.case_id} difficulty={obs.difficulty} steps={obs.remaining_steps}") |
|
|
| while not obs.done: |
| user_msg = build_observation_prompt(obs, is_first=(steps == 0)) |
| messages.append({"role": "user", "content": user_msg}) |
|
|
| try: |
| api_kwargs = {"model": MODEL_NAME, "messages": messages, "temperature": 0.0, **_tok_kwarg()} |
| try: |
| api_kwargs["response_format"] = {"type": "json_object"} |
| response = llm.chat.completions.create(**api_kwargs) |
| except Exception: |
| del api_kwargs["response_format"] |
| response = llm.chat.completions.create(**api_kwargs) |
| assistant_msg = response.choices[0].message.content or "" |
| except Exception as e: |
| print(f" [LLM ERROR] {e}") |
| assistant_msg = '{"action_type": "summarize_findings"}' |
|
|
| messages.append({"role": "assistant", "content": assistant_msg}) |
|
|
| params = parse_llm_response(assistant_msg) |
| action = build_action(params) |
|
|
| result = await env.step(action) |
| obs = result.observation |
| obs.done = result.done |
| obs.reward = result.reward |
| reward = result.reward or 0.0 |
| rewards.append(reward) |
| steps += 1 |
|
|
| status = "ERR" if obs.last_action_error else "ok" |
| print(f" [STEP {steps:2d}] {action.action_type.value:35s} reward={reward:+.2f} remain={obs.remaining_steps} {status}") |
|
|
| grader = getattr(obs, "grader_result", {}) or {} |
| score = grader.get("score", 0.0) if isinstance(grader, dict) else 0.0 |
| decision = params.get("final_decision", "none") |
|
|
| return {"task_id": task_id, "steps": steps, "score": score, "decision": decision, "rewards": rewards} |
|
|
|
|
| async def main(): |
| print("=" * 65) |
| print(" JUDGE SIMULATION -- Live HF Space + LLM Agent") |
| print(f" Space: {SPACE_URL}") |
| print(f" Model: {MODEL_NAME}") |
| print(f" API: {API_BASE_URL}") |
| print("=" * 65) |
|
|
| llm = OpenAI(base_url=API_BASE_URL, api_key=API_KEY) |
| env = InvoiceGuardEnv(base_url=SPACE_URL) |
|
|
| results = [] |
| try: |
| async with env: |
| for task_id in TASKS: |
| print(f"\n--- {task_id} ---") |
| start = time.time() |
| try: |
| r = await run_task(env, llm, task_id) |
| r["time"] = time.time() - start |
| results.append(r) |
| print(f" >> score={r['score']:.4f} decision={r['decision']} time={r['time']:.1f}s") |
| except Exception as e: |
| print(f" >> FAILED: {e}") |
| results.append({"task_id": task_id, "score": 0.0, "steps": 0, "error": str(e)}) |
| except Exception as e: |
| print(f"\nConnection error: {e}") |
| sys.exit(1) |
|
|
| print(f"\n\n{'='*65}") |
| print(" RESULTS SUMMARY") |
| print(f"{'='*65}") |
| scores = [] |
| for r in results: |
| s = r.get("score", 0.0) |
| scores.append(s) |
| dec = r.get("decision", "n/a") |
| print(f" {r['task_id']:40s} score={s:.4f} decision={dec}") |
| avg = sum(scores) / len(scores) if scores else 0.0 |
| print(f"\n Average score: {avg:.4f}") |
| print(f" Tasks passed: {sum(1 for s in scores if s >= 0.5)}/{len(scores)}") |
| print(f"{'='*65}") |
|
|
|
|
| if __name__ == "__main__": |
| asyncio.run(main()) |
|
|
