Merge pull request #94 from ClickHouse/feat/agent-arena-online-evaluation

feat(agent-arena): add the continuous online evaluation loop

Author: sishuo-yang

workshops/agent_arena/.env.example

@@ -7,12 +7,19 @@ export ARENA_RO_PASSWORD=
 # OpenRouter (LLM provider)
 export OPENROUTER_API_KEY=sk-or-...
 export OPENROUTER_BASE_URL=https://openrouter.ai/api/v1
+# policy-v2 is the current default; Module 03 explicitly overrides it to
+# policy-v1 when starting the stale seeded release.
+export AGENT_ARENA_POLICY_VERSION=policy-v2
+# Non-secret defaults for the calibrated workshop winner.
+export WINNER_MODEL=qwen3.7-flash
+export WINNER_PROMPT=P2_fewshot
+export WINNER_CONFIG_ID=qwen3.7-flash__P2_fewshot
 # Instructor only (public workshops): a Management key from
 # https://openrouter.ai/settings/management-keys — used by
 # scripts/provision_workshop_keys.py to mint capped learner keys. Admin-level:
 # NEVER give this to learners or commit it. Learners only get OPENROUTER_API_KEY.
 # export OPENROUTER_PROVISIONING_KEY=sk-or-v1-...
-# LangFuse Cloud (eval store + tracing)
+# Langfuse Cloud (eval store + tracing)
 export LANGFUSE_BASE_URL=https://us.cloud.langfuse.com
 export LANGFUSE_PUBLIC_KEY=pk-lf-...
 export LANGFUSE_SECRET_KEY=sk-lf-...

workshops/agent_arena/README.md

@@ -6,6 +6,7 @@
 import sys
 import time
 import urllib.request
+from collections.abc import Mapping
 from urllib.error import HTTPError
 from dataclasses import dataclass
 
@@ -58,6 +59,45 @@ def _parse_response(resp: dict) -> ConverseResult:
     return ConverseResult(text=text, usage=usage)
 
 
+_REASONING_KEYS = frozenset({"max_tokens", "effort", "exclude", "enabled"})
+_REASONING_EFFORTS = frozenset({
+    "max", "xhigh", "high", "medium", "low", "minimal", "none",
+})
+
+_REASONING_MAPPING_ERROR = "reasoning must be a mapping"
+_REASONING_FIELDS_ERROR = "reasoning contains unsupported fields"
+_REASONING_MAX_TOKENS_ERROR = "reasoning max_tokens has an invalid value"
+_REASONING_EFFORT_ERROR = "reasoning effort has an invalid value"
+_REASONING_EXCLUDE_ERROR = "reasoning exclude has an invalid value"
+_REASONING_ENABLED_ERROR = "reasoning enabled has an invalid value"
+_REASONING_MODE_ERROR = "reasoning mode fields are mutually exclusive"
+
+
+def _validate_reasoning(reasoning: object) -> dict:
+    if not isinstance(reasoning, Mapping):
+        raise TypeError(_REASONING_MAPPING_ERROR)
+    if not set(reasoning).issubset(_REASONING_KEYS):
+        raise ValueError(_REASONING_FIELDS_ERROR)
+    if "max_tokens" in reasoning:
+        max_tokens = reasoning["max_tokens"]
+        if (isinstance(max_tokens, bool) or
+                not isinstance(max_tokens, int) or
+                max_tokens <= 0):
+            raise ValueError(_REASONING_MAX_TOKENS_ERROR)
+    if ("effort" in reasoning and
+            (not isinstance(reasoning["effort"], str) or
+             reasoning["effort"] not in _REASONING_EFFORTS)):
+        raise ValueError(_REASONING_EFFORT_ERROR)
+    if "exclude" in reasoning and type(reasoning["exclude"]) is not bool:
+        raise ValueError(_REASONING_EXCLUDE_ERROR)
+    if "enabled" in reasoning and type(reasoning["enabled"]) is not bool:
+        raise ValueError(_REASONING_ENABLED_ERROR)
+    mode_fields = {"max_tokens", "effort", "enabled"}.intersection(reasoning)
+    if len(mode_fields) > 1:
+        raise ValueError(_REASONING_MODE_ERROR)
+    return dict(reasoning)
+
+
 class OpenRouterClient:
     def __init__(self, base_url: str, api_key: str):
         self._base = base_url.rstrip("/")
@@ -72,6 +112,8 @@ def converse(self, model_id: str, system: str, messages: list[dict],
         max_tok = inference.get("max_tokens", inference.get("maxTokens"))
         if max_tok is not None:
             body["max_tokens"] = max_tok
+        if "reasoning" in inference:
+            body["reasoning"] = _validate_reasoning(inference["reasoning"])
         req = urllib.request.Request(
             f"{self._base}/chat/completions",
             data=json.dumps(body).encode(),

workshops/agent_arena/agents/llm.py

@@ -61,12 +61,15 @@ def run_agent(question: str, model_cfg: ModelCfg, prompt_cfg: PromptCfg,
                                    attempt + 1, usage_total, "ok", transcript)
             except Exception as e:  # noqa: BLE001
                 last_err, hint = str(e), "sql_exec_error"
-        if attempt < max_retries and prompt_cfg.self_correct:
-            transcript.append({"role": "user",
-                               "content": f"That query failed: {last_err}. "
-                                          "Return a corrected ClickHouse SQL query."})
-            messages = messages + correction_turn(last_sql or "", last_err or "")
-            continue
+        if attempt < max_retries:
+            if not resp.text.strip():
+                continue
+            if prompt_cfg.self_correct:
+                transcript.append({"role": "user",
+                                   "content": f"That query failed: {last_err}. "
+                                              "Return a corrected ClickHouse SQL query."})
+                messages = messages + correction_turn(last_sql or "", last_err or "")
+                continue
         break
 
     return AgentResult(last_sql, None, None, last_err,

workshops/agent_arena/agents/loop.py

@@ -0,0 +1,46 @@
+from dataclasses import dataclass
+from pathlib import Path
+import re
+
+import yaml
+
+
+_VERSION = re.compile(r"^policy-v[0-9]+$")
+
+
+@dataclass(frozen=True)
+class PolicyContext:
+    version: str
+    metrics: dict[str, str]
+    rendered: str
+
+
+def load_policy(version: str, root: Path = Path("policies")) -> PolicyContext:
+    if not _VERSION.fullmatch(version):
+        raise ValueError(f"unknown policy version: {version}")
+
+    path = root / f"{version}.yaml"
+    if not path.is_file():
+        raise ValueError(f"unknown policy version: {version}")
+
+    raw = yaml.safe_load(path.read_text())
+    if not isinstance(raw, dict):
+        raise ValueError("policy document must be a mapping")
+    metrics = raw.get("metrics")
+    if raw.get("version") != version:
+        raise ValueError(f"policy version mismatch in {path}")
+    if not isinstance(metrics, dict) or not metrics or not all(
+        isinstance(name, str) and isinstance(definition, str) and definition.strip()
+        for name, definition in metrics.items()
+    ):
+        raise ValueError("metrics must be a non-empty mapping of names to definitions")
+
+    lines = [f"# Business metric policy ({version})"]
+    lines.extend(
+        f"- {name}: {definition.strip()}" for name, definition in metrics.items()
+    )
+    return PolicyContext(version, metrics, "\n".join(lines))
+
+
+def with_policy(schema_context: str, policy: PolicyContext) -> str:
+    return f"{schema_context.rstrip()}\n\n{policy.rendered}\n"

workshops/agent_arena/agents/policy.py

@@ -23,6 +23,8 @@ openrouter:
   inference:
     temperature: 0.0
     max_tokens: 2048          # headroom for reasoning models
+    reasoning:
+      max_tokens: 512         # leave output budget for the final SQL answer
 
 langfuse:
   host: ${LANGFUSE_BASE_URL}

workshops/agent_arena/config.yaml

@@ -12,6 +12,7 @@ class GoldenQuestion:
     tags: list = field(default_factory=list)
     notes: str = ""
     fewshot_holdout: bool = False
+    float_dp: int | None = None
 
 
 def load_golden(path: str = "golden/questions.yaml") -> list[GoldenQuestion]:
@@ -22,6 +23,7 @@ def load_golden(path: str = "golden/questions.yaml") -> list[GoldenQuestion]:
         ordered=q.get("ordered", False), golden_sql=q["golden_sql"],
         tags=q.get("tags", []), notes=q.get("notes", ""),
         fewshot_holdout=q.get("fewshot_holdout", False),
+        float_dp=q.get("float_dp"),
     ) for q in raw]
 
 

workshops/agent_arena/eval/golden.py

@@ -11,12 +11,19 @@
 from agents.chclient import ROClickHouseClient
 from agents.llm import OpenRouterClient, cost_usd
 from agents.loop import run_agent
+from agents.policy import load_policy, with_policy
 from eval.golden import GoldenQuestion, load_golden, fewshot_examples
 from eval.langfuse_adapter import LangfuseTracer, emit_agent_trace
 from eval.serialize import result_payload, golden_payload
 
 
-def main() -> None:
+DEFAULT_REQUIRED_SCORE_NAMES = (
+    "correctness",
+    "agent-arena-llm-judge",
+)
+
+
+def parse_args(argv=None):
     ap = argparse.ArgumentParser()
     ap.add_argument("--run-id", default=None)
     ap.add_argument("--eval-timeout", type=int, default=180,
@@ -28,10 +35,41 @@ def main() -> None:
     ap.add_argument("--models-file", default="",
                     help="JSON file of model specs [{id,name,family,price_per_1m_in,price_per_1m_out}] "
                          "to run instead of config.yaml models (used by the web UI)")
-    args = ap.parse_args()
+    ap.add_argument("--policy-version", default="policy-v2")
+    ap.add_argument("--wait-for-score", action="append", default=[])
+    return ap.parse_args(argv)
+
+
+def effective_run_id(run_id: str, policy_version: str) -> str:
+    return f"{run_id}--{policy_version}"
+
+
+def _dataset_item(q: GoldenQuestion, *, rows, cols, default_float_dp: int) -> dict:
+    float_dp = q.float_dp if q.float_dp is not None else default_float_dp
+    return {
+        "id": q.id,
+        "question": q.question,
+        "expected_output": golden_payload(
+            golden_sql=q.golden_sql,
+            rows=rows,
+            cols=cols,
+            ordered=q.ordered,
+        ),
+        "metadata": {
+            "tier": q.tier,
+            "ordered": q.ordered,
+            "float_dp": float_dp,
+        },
+    }
+
+
+def main(argv=None) -> None:
+    args = parse_args(argv)
 
     cfg = load_config()
-    run_id = args.run_id or f"run-{uuid.uuid4().hex[:8]}"
+    release = args.run_id or f"run-{uuid.uuid4().hex[:8]}"
+    policy = load_policy(args.policy_version)
+    run_id = effective_run_id(release, policy.version)
 
     # The web UI can hand the harness an arbitrary set of models (from the live
     # OpenRouter catalog) via a JSON file, overriding config.yaml's model list.
@@ -56,7 +94,7 @@ def main() -> None:
     llm = OpenRouterClient(cfg.openrouter.base_url, cfg.openrouter.api_key)
 
     with open("schema/schema_context.md") as f:
-        schema_ctx = f.read()
+        schema_ctx = with_policy(f.read(), policy)
 
     all_questions = load_golden()
     # Hold out the few-shot example questions so P2 is not graded on its own
@@ -71,15 +109,15 @@ def main() -> None:
         golden_cache[q.id] = (gr.rows, gr.cols)
 
     # Upload the golden set as a Langfuse Dataset (each config becomes a Run).
-    tracer.ensure_dataset([{
-        "id": q.id, "question": q.question,
-        "expected_output": golden_payload(golden_sql=q.golden_sql,
-                                           rows=golden_cache[q.id][0],
-                                           cols=golden_cache[q.id][1],
-                                           ordered=q.ordered),
-        "metadata": {"tier": q.tier, "ordered": q.ordered,
-                     "float_dp": cfg.eval.float_dp},
-    } for q in questions])
+    tracer.ensure_dataset([
+        _dataset_item(
+            q,
+            rows=golden_cache[q.id][0],
+            cols=golden_cache[q.id][1],
+            default_float_dp=cfg.eval.float_dp,
+        )
+        for q in questions
+    ])
 
     model_names, prompt_names = cfg.resolved_grid()
     if args.models:
@@ -126,9 +164,11 @@ def task(*, item, _m=mname, _p=pname, _mc=mcfg, _pc=pcfg,
                     payload, c, latency_ms, ar = run_one(_m, _p, _mc, _pc, _ex, _cid, q)
                     trace_id = emit_agent_trace(
                         trace_name="agent_run", session_id=_sid,
-                        tags=[_cid, f"run:{run_id}", _m, _p],
+                        tags=[_cid, f"run:{run_id}", policy.version, _m, _p],
                         metadata={"config_id": _cid, "question_id": q.id,
-                                  "model": _m, "prompt": _p, "run_id": run_id},
+                                  "model": _m, "prompt": _p, "run_id": run_id,
+                                  "policy_version": policy.version,
+                                  "release": release},
                         question=q.question, model=_m,
                         transcript=ar.transcript, sql=ar.sql,
                         output_payload=payload, usage=ar.usage)
@@ -148,7 +188,13 @@ def task(*, item, _m=mname, _p=pname, _mc=mcfg, _pc=pcfg,
     tracer.flush()
 
     if pending_trace_ids:
-        _wait_for_scores(tracer, pending_trace_ids, args.eval_timeout, args.eval_poll)
+        _wait_for_scores(
+            tracer,
+            pending_trace_ids,
+            args.eval_timeout,
+            args.eval_poll,
+            args.wait_for_score,
+        )
     print(f"done. run_id={run_id}", flush=True)
 
 
@@ -193,25 +239,39 @@ def _question_for_item(item, qmap) -> GoldenQuestion:
                           golden_sql=str(golden_sql))
 
 
-def _wait_for_scores(tracer, trace_ids, timeout, poll) -> None:
-    """Wait until both required asynchronous Langfuse evaluators have finished."""
+def _wait_for_scores(tracer, trace_ids, timeout, poll, required_names=None) -> None:
+    """Compatibility wrapper for the two default asynchronous evaluators."""
+    _wait_for_required_scores(
+        tracer, trace_ids, timeout, poll, required_names=required_names
+    )
+
+
+def _wait_for_required_scores(
+    tracer, trace_ids, timeout, poll, required_names=None
+) -> None:
+    """Wait for the default scores and every additional exact score name."""
+    required = list(dict.fromkeys([
+        *DEFAULT_REQUIRED_SCORE_NAMES,
+        *(required_names or []),
+    ]))
     remaining = set(trace_ids)
     print(f"grading via LangFuse evaluators — waiting on {len(remaining)} traces "
           f"(timeout {timeout}s)…", flush=True)
     deadline = time.time() + timeout
-    last_missing = {tid: {"correctness", "llm_judge"} for tid in remaining}
+    last_missing = {tid: set(required) for tid in remaining}
     while remaining and time.time() < deadline:
         tids = list(remaining)
         with ThreadPoolExecutor(max_workers=min(8, len(tids))) as pool:
             score_lists = list(pool.map(tracer.fetch_trace_scores, tids))
         done = []
         for tid, scores in zip(tids, score_lists):
-            corr, judge, _outcome = _classify_scores(scores)
-            missing = set()
-            if corr is None:
-                missing.add("correctness")
-            if judge is None:
-                missing.add("llm_judge")
+            available = {
+                score.get("name")
+                for score in scores or []
+                if score.get("name")
+                and (score.get("value") is not None or score.get("string") is not None)
+            }
+            missing = set(required) - available
             last_missing[tid] = missing
             if not missing:
                 done.append(tid)
@@ -224,7 +284,7 @@ def _wait_for_scores(tracer, trace_ids, timeout, poll) -> None:
         missing_names = sorted({name for tid in remaining for name in last_missing[tid]})
         raise RuntimeError(
             f"{len(remaining)} traces are missing {', '.join(missing_names)} after {timeout}s; "
-            "configure both Langfuse evaluators for Experiments on arena-golden. "
+            "configure the required Langfuse evaluators for Experiments on arena-golden. "
             "See eval/langfuse_evaluators/README.md."
         )
     print(f"Langfuse scored all {len(trace_ids)} traces; leaderboard ready", flush=True)