supportbench/utils.py

"""Utility helpers for SupportBench."""

from __future__ import annotations

import json
from typing import Any, Dict, Optional


def safe_parse_json(text: str) -> Optional[Dict[str, Any]]:
    """Attempt to extract and parse JSON from a string. Returns None on failure."""
    text = text.strip()
    # Try direct parse
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass
    # Try to find JSON object in text
    start = text.find("{")
    end = text.rfind("}")
    if start != -1 and end != -1 and end > start:
        try:
            return json.loads(text[start : end + 1])
        except json.JSONDecodeError:
            pass
    return None


def clamp(value: float, lo: float, hi: float) -> float:
    return max(lo, min(hi, value))


def format_observation_for_prompt(obs: Dict[str, Any]) -> str:
    """Format an observation dict into a readable prompt block."""
    lines = [
        f"TASK: {obs['task_name']} ({obs['task_id']})",
        f"STATUS: {obs['current_status']}  |  STEP {obs['steps_taken']}/{obs['max_steps']}",
        "",
        "=== CUSTOMER MESSAGE ===",
        obs["customer_message"],
        "",
        "=== CUSTOMER PROFILE ===",
        json.dumps(obs["customer_profile"], indent=2),
        "",
        "=== ORDER INFO ===",
        json.dumps(obs["order_info"], indent=2),
        "",
        "=== POLICY ===",
    ]
    for i, p in enumerate(obs["policy_snippets"], 1):
        lines.append(f"{i}. {p}")
    lines += [
        "",
        "=== TICKET HISTORY ===",
        json.dumps(obs["ticket_history"], indent=2) if obs["ticket_history"] else "(none)",
        "",
        f"LAST RESULT: {obs.get('last_action_result') or '(none)'}",
        f"LAST ERROR: {obs.get('last_action_error') or 'null'}",
        "",
        "AVAILABLE ACTIONS: " + ", ".join(obs["available_actions"]),
    ]
    return "\n".join(lines)