Chaim
338a8a947f
Grounded (INV-AH) multi-instance synthesis with drift guard + chair gate (pending_review, G10). Single path used by backfill, MCP tool, nightly drain. HELD from production run pending the principles-redesign (rename+cull, #152). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
115 lines
4.6 KiB
Python
115 lines
4.6 KiB
Python
"""Three independent-lineage LLM judges — the shared primitive (G2).
|
|
|
|
Extracted from scripts/halacha_panel_approve.py so the panel-extraction regime
|
|
(#152) and the existing approval-triage share ONE implementation of the judges
|
|
(no parallel HTTP/auth paths). Diversity of lineage is the point — cross-model
|
|
agreement is the reliable signal (gold-set AC1=0.92):
|
|
|
|
• claude — Opus via claude_session (local CLI, zero marginal cost) [Anthropic]
|
|
• deepseek — api.deepseek.com (deepseek-chat) [DeepSeek]
|
|
• gemini — generativelanguage (gemini-2.5-flash, #1 LegalBench) [Google]
|
|
|
|
Every judge has the SAME signature ``(system, user) -> dict | None`` and returns
|
|
None on ANY failure (missing key, HTTP error, bad JSON) — callers must tolerate a
|
|
missing judge (a 2/3 panel is still actionable).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
from legal_mcp.services import claude_session
|
|
|
|
|
|
def _env_key(name: str, *files: str) -> str:
|
|
for f in files:
|
|
p = Path(f).expanduser()
|
|
if p.exists():
|
|
for line in p.read_text().splitlines():
|
|
if line.startswith(name + "="):
|
|
return line.split("=", 1)[1].strip()
|
|
return os.environ.get(name, "")
|
|
|
|
|
|
DEEPSEEK_KEY = _env_key("DEEPSEEK_API_KEY", "~/.hermes/profiles/deepseek/.env", "~/.env")
|
|
# canonical Infisical name is GOOGLE_GEMINI_API_KEY (/external-apis/gemini); accept
|
|
# the bare GEMINI_API_KEY too for back-compat.
|
|
GEMINI_KEY = _env_key("GOOGLE_GEMINI_API_KEY", "~/.env") or _env_key("GEMINI_API_KEY", "~/.env")
|
|
|
|
JUDGE_NAMES = ("claude", "deepseek", "gemini")
|
|
|
|
|
|
def available() -> dict[str, bool]:
|
|
return {"claude": True, "deepseek": bool(DEEPSEEK_KEY), "gemini": bool(GEMINI_KEY)}
|
|
|
|
|
|
async def judge_claude(system: str, user: str, *, max_tokens: int = 2000) -> dict | list | None:
|
|
try:
|
|
# tools="" → no tool_use, so a pure text→JSON extraction never trips
|
|
# error_max_turns (and wastes no retries on a web-search detour).
|
|
return await claude_session.query_json(user, system=system, tools="")
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
async def judge_deepseek(
|
|
client: httpx.AsyncClient, system: str, user: str, *, max_tokens: int = 2000,
|
|
) -> dict | list | None:
|
|
if not DEEPSEEK_KEY:
|
|
return None
|
|
try:
|
|
r = await client.post(
|
|
"https://api.deepseek.com/v1/chat/completions",
|
|
headers={"Authorization": f"Bearer {DEEPSEEK_KEY}", "Content-Type": "application/json"},
|
|
json={"model": "deepseek-chat", "temperature": 0, "max_tokens": max_tokens,
|
|
"response_format": {"type": "json_object"},
|
|
"messages": [{"role": "system", "content": system},
|
|
{"role": "user", "content": user}]},
|
|
timeout=120,
|
|
)
|
|
r.raise_for_status()
|
|
return json.loads(r.json()["choices"][0]["message"]["content"])
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
async def judge_gemini(
|
|
client: httpx.AsyncClient, system: str, user: str, *, max_tokens: int = 8000,
|
|
) -> dict | list | None:
|
|
if not GEMINI_KEY:
|
|
return None
|
|
try:
|
|
r = await client.post(
|
|
f"https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent?key={GEMINI_KEY}",
|
|
headers={"Content-Type": "application/json"},
|
|
json={"system_instruction": {"parts": [{"text": system}]},
|
|
"contents": [{"parts": [{"text": user}]}],
|
|
# thinkingBudget=0 disables gemini-2.5-flash's "thinking", which
|
|
# otherwise eats the output budget on large inputs → empty parts
|
|
# → finishReason MAX_TOKENS → the judge silently dropped out.
|
|
"generationConfig": {"temperature": 0, "maxOutputTokens": max_tokens,
|
|
"responseMimeType": "application/json",
|
|
"thinkingConfig": {"thinkingBudget": 0}}},
|
|
timeout=120,
|
|
)
|
|
r.raise_for_status()
|
|
parts = (r.json().get("candidates") or [{}])[0].get("content", {}).get("parts")
|
|
if not parts:
|
|
return None
|
|
return json.loads(parts[0]["text"])
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def to_bool(d: dict | None, key: str) -> bool | None:
|
|
"""Robust bool coercion for a judge JSON field (handles he/en truthy strings)."""
|
|
if not isinstance(d, dict) or key not in d:
|
|
return None
|
|
v = d[key]
|
|
if isinstance(v, bool):
|
|
return v
|
|
return str(v).strip().lower() in ("true", "1", "yes", "כן")
|