File size: 5,003 Bytes
af8b301
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
143
144
145
146
147
148
149
150
"""
Shared utility functions used across the project.
"""

from __future__ import annotations

import json
import logging
import os
from pathlib import Path
from typing import Any, Dict, List, Optional


# ---------------------------------------------------------------------------
# Logging setup
# ---------------------------------------------------------------------------

def setup_logging(level: str = "INFO") -> None:
    """Configure root logger with a clean format."""
    logging.basicConfig(
        level=getattr(logging, level.upper(), logging.INFO),
        format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
        datefmt="%H:%M:%S",
    )


# ---------------------------------------------------------------------------
# Environment helpers
# ---------------------------------------------------------------------------

def get_env(key: str, default: Optional[str] = None, required: bool = False) -> Optional[str]:
    """
    Retrieve an environment variable with optional validation.

    Args:
        key: Environment variable name
        default: Default value if not set
        required: Raise ValueError if not set and no default
    """
    value = os.getenv(key, default)
    if required and not value:
        raise ValueError(
            f"Required environment variable '{key}' is not set. "
            f"Check your .env file."
        )
    return value


# ---------------------------------------------------------------------------
# JSON helpers
# ---------------------------------------------------------------------------

def load_json(path: Path | str) -> Any:
    """Load and return JSON from a file."""
    path = Path(path)
    if not path.exists():
        raise FileNotFoundError(f"JSON file not found: {path}")
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)


def save_json(data: Any, path: Path | str, indent: int = 2) -> None:
    """Save data as JSON to a file, creating parent dirs as needed."""
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=indent, ensure_ascii=False)


# ---------------------------------------------------------------------------
# Text helpers
# ---------------------------------------------------------------------------

def truncate(text: str, max_chars: int = 200, suffix: str = "...") -> str:
    """Truncate text to max_chars, appending suffix if truncated."""
    if len(text) <= max_chars:
        return text
    return text[:max_chars - len(suffix)] + suffix


def count_words(text: str) -> int:
    return len(text.split())


# ---------------------------------------------------------------------------
# Config loader
# ---------------------------------------------------------------------------

def load_config(config_path: Path | str = ".env") -> Dict[str, str]:
    """
    Load key=value pairs from a .env file into a dict.
    Does NOT set environment variables — use python-dotenv for that.
    """
    config: Dict[str, str] = {}
    path = Path(config_path)
    if not path.exists():
        return config
    with open(path, "r") as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            if "=" in line:
                key, _, value = line.partition("=")
                config[key.strip()] = value.strip().strip('"').strip("'")
    return config


# ---------------------------------------------------------------------------
# Eval dataset helpers
# ---------------------------------------------------------------------------

def load_eval_dataset(paths: List[Path | str]) -> List[dict]:
    """Load and merge multiple eval JSON files into a single list."""
    combined: List[dict] = []
    for path in paths:
        try:
            data = load_json(path)
            if isinstance(data, list):
                combined.extend(data)
            elif isinstance(data, dict) and "prompts" in data:
                combined.extend(data["prompts"])
        except Exception as e:
            logging.getLogger(__name__).warning("Could not load eval file %s: %s", path, e)
    return combined


def format_cost_table(
    oss_tokens: int,
    frontier_tokens: int,
    oss_cost_per_1k: float = 0.0,
    frontier_cost_per_1k: float = 0.0006,
) -> str:
    """
    Format a simple cost comparison table as a markdown string.

    Default costs:
      - OSS (self-hosted): $0.00 per 1k tokens
      - Llama 3.1-8B via Groq: $0.00 (free tier) / ~$0.00005 per 1k tokens (paid)
    """
    oss_cost = (oss_tokens / 1000) * oss_cost_per_1k
    frontier_cost = (frontier_tokens / 1000) * frontier_cost_per_1k

    return (
        f"| Model       | Tokens | Cost (USD) |\n"
        f"|-------------|--------|------------|\n"
        f"| OSS (local) | {oss_tokens:,} | ${oss_cost:.4f} |\n"
        f"| Frontier    | {frontier_tokens:,} | ${frontier_cost:.4f} |\n"
    )