#!/usr/bin/env python3
"""aria-llm-providers — Multi-Provider LLM-Adapter (KAR-121).
Wrappt Anthropic + OpenAI + DeepSeek + Gemini hinter einheitlicher API.
Auto-Trace via aria_llm_trace.log_call. Cost-Computation aus Pricing-Table.
API:
from aria_llm_providers import complete, judge
out = complete(provider="openai", model="gpt-5", prompt="...", max_tokens=400)
# out = {"text": "...", "prompt_tokens": ..., "completion_tokens": ...,
# "cost_usd": ..., "latency_ms": ..., "raw": {...}}
CLI:
python3 aria-llm-providers.py complete --provider gemini --model gemini-2.5-flash --prompt "hi"
python3 aria-llm-providers.py judge --worker claude --judge gpt-5 \
--input "..." --output "..." --rubric "..."
python3 aria-llm-providers.py models # list available models per provider
"""
from __future__ import annotations
import argparse
import dataclasses
import hashlib
import json
import os
import sys
import time
import urllib.request
from pathlib import Path
from typing import Any
# ----- Load all keys from /root/.aria-secrets/ -------------------------------
SECRETS_DIR = Path("/root/.aria-secrets")
ARIA_ENV = Path("/root/aria/.env")
def _load_envs() -> dict[str, str]:
env = dict(os.environ)
for envfile in [ARIA_ENV] + sorted(SECRETS_DIR.glob("*.env")):
if envfile.exists():
for line in envfile.read_text(encoding="utf-8", errors="replace").splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
k, _, v = line.partition("=")
env[k.strip()] = v.strip().strip('"').strip("'")
return env
_ENV = _load_envs()
def _key(name: str) -> str:
return _ENV.get(name, "")
# ----- Pricing-Table ($/1M tokens) ------------------------------------------
# Stand 2026-05-13 — verifiziere via WebSearch wenn Cost-Anomalie auftritt.
PRICING = {
# provider/model: (input_per_M, output_per_M)
("anthropic", "claude-opus-4-7"): (15.00, 75.00),
("anthropic", "claude-sonnet-4-6"): (3.00, 15.00),
("anthropic", "claude-haiku-4-5"): (0.80, 4.00),
("openai", "gpt-5"): (5.00, 15.00), # estimate, verify
("openai", "gpt-4o"): (2.50, 10.00),
("openai", "text-embedding-3-large"): (0.13, 0.0),
("openai", "text-embedding-3-small"): (0.02, 0.0),
("deepseek", "deepseek-v4-pro"): (0.55, 1.65), # public DeepSeek-V4 pricing
("deepseek", "deepseek-v4-flash"): (0.14, 0.28),
("google", "gemini-2.5-pro"): (1.25, 5.00),
("google", "gemini-2.5-flash"): (0.10, 0.40),
}
DEFAULT_MODELS = {
"anthropic": "claude-opus-4-7",
"openai": "gpt-5",
"deepseek": "deepseek-v4-pro",
"google": "gemini-2.5-pro",
}
FAMILY_OF = {
"claude": "anthropic", "claude-opus-4-7": "anthropic", "claude-sonnet-4-6": "anthropic",
"claude-haiku-4-5": "anthropic",
"gpt-5": "openai", "gpt-4o": "openai", "text-embedding-3-large": "openai",
"deepseek-v4-pro": "deepseek", "deepseek-v4-flash": "deepseek",
"gemini-2.5-pro": "google", "gemini-2.5-flash": "google",
}
# ----- Provider-Adapters ----------------------------------------------------
def _http_post_json(url: str, headers: dict, body: dict, timeout: int = 60,
retries: int = 3) -> dict:
"""HTTP POST mit Retry für 5xx + Timeouts (Phase-2.4 Outage-Fallback)."""
data = json.dumps(body).encode()
delay = 1.0
last_err = None
for attempt in range(retries + 1):
try:
req = urllib.request.Request(url, data=data,
headers={**headers, "Content-Type": "application/json"},
method="POST")
with urllib.request.urlopen(req, timeout=timeout) as resp:
return json.loads(resp.read())
except urllib.error.HTTPError as exc:
last_err = exc
# 5xx → retry, 4xx → don't retry (auth/quota etc.)
if 500 <= exc.code < 600 and attempt < retries:
sys.stderr.write(f"[provider] {exc.code} retry {attempt + 1}/{retries} in {delay:.1f}s\n")
time.sleep(delay)
delay *= 2
continue
raise
except (urllib.error.URLError, TimeoutError) as exc:
last_err = exc
if attempt < retries:
sys.stderr.write(f"[provider] network err retry {attempt + 1}/{retries} in {delay:.1f}s\n")
time.sleep(delay)
delay *= 2
continue
raise
raise last_err or RuntimeError("retry-loop exhausted")
# Mechanical-Validation (Phase 2.3) — vor Judge-Call billige Heuristik-Checks
# Aus Kais' Architektur-Doc Sektion 3: spart 30-40% Judge-Calls bei Code-Heavy-Outputs
def mechanical_validate(output_text: str, kind: str = "json") -> dict:
"""Pre-Judge mechanical-check. Returns {is_valid, errors, kind}.
Wenn invalid: caller kann Judge-Call überspringen + direkt fail-Verdict.
"""
errors = []
valid = True
if kind == "json":
# Strip code-fences
text = output_text.strip()
if text.startswith("```"):
text = "\n".join(text.split("\n")[1:])
if text.endswith("```"):
text = text[:-3]
try:
json.loads(text)
except json.JSONDecodeError as exc:
valid = False
errors.append(f"json-parse: {exc}")
elif kind == "python":
try:
import ast
ast.parse(output_text)
except SyntaxError as exc:
valid = False
errors.append(f"python-syntax: {exc}")
elif kind == "markdown_frontmatter":
# Brain-Note-Check: has --- frontmatter + title field?
if not output_text.lstrip().startswith("---"):
valid = False
errors.append("missing-frontmatter")
elif "\ntitle:" not in output_text[:500]:
valid = False
errors.append("missing-title")
elif kind == "sql":
# cheap sanity: contains SELECT/INSERT/UPDATE/CREATE/DROP and balanced parens
upper = output_text.upper()
if not any(k in upper for k in ("SELECT", "INSERT", "UPDATE", "DELETE", "CREATE", "DROP", "ALTER")):
valid = False
errors.append("no-sql-keyword")
if output_text.count("(") != output_text.count(")"):
valid = False
errors.append("unbalanced-parens")
else:
# unknown kind → pass-through, validator not applicable
pass
return {"is_valid": valid, "errors": errors, "kind": kind}
def _complete_anthropic(model: str, prompt: str, max_tokens: int) -> dict:
api_key = _key("ANTHROPIC_API_KEY")
if not api_key:
# Aria läuft selbst auf OAuth — kein API-Key in env nötig wenn wir intern bleiben.
# Für inter-LLM-Tests via API: Key nötig.
raise RuntimeError("ANTHROPIC_API_KEY missing (place in /root/.aria-secrets/anthropic.env)")
result = _http_post_json(
"https://api.anthropic.com/v1/messages",
headers={"x-api-key": api_key, "anthropic-version": "2023-06-01"},
body={"model": model, "max_tokens": max_tokens,
"messages": [{"role": "user", "content": prompt}]},
)
text = "".join(b.get("text", "") for b in result.get("content", []) if b.get("type") == "text")
usage = result.get("usage", {})
return {
"text": text, "raw": result,
"prompt_tokens": usage.get("input_tokens", 0),
"completion_tokens": usage.get("output_tokens", 0),
}
def _complete_openai(model: str, prompt: str, max_tokens: int) -> dict:
api_key = _key("OPENAI_API_KEY")
if not api_key:
raise RuntimeError("OPENAI_API_KEY missing")
result = _http_post_json(
"https://api.openai.com/v1/chat/completions",
headers={"Authorization": f"Bearer {api_key}"},
body={"model": model, "max_completion_tokens": max_tokens,
"messages": [{"role": "user", "content": prompt}]},
)
text = result["choices"][0]["message"]["content"] or ""
usage = result.get("usage", {})
return {
"text": text, "raw": result,
"prompt_tokens": usage.get("prompt_tokens", 0),
"completion_tokens": usage.get("completion_tokens", 0),
}
def _complete_deepseek(model: str, prompt: str, max_tokens: int) -> dict:
api_key = _key("DEEPSEEK_API_KEY")
if not api_key:
raise RuntimeError("DEEPSEEK_API_KEY missing")
result = _http_post_json(
"https://api.deepseek.com/v1/chat/completions",
headers={"Authorization": f"Bearer {api_key}"},
body={"model": model, "max_tokens": max_tokens,
"messages": [{"role": "user", "content": prompt}]},
)
text = result["choices"][0]["message"]["content"] or ""
usage = result.get("usage", {})
return {
"text": text, "raw": result,
"prompt_tokens": usage.get("prompt_tokens", 0),
"completion_tokens": usage.get("completion_tokens", 0),
}
def _complete_google(model: str, prompt: str, max_tokens: int) -> dict:
api_key = _key("GEMINI_API_KEY")
if not api_key:
raise RuntimeError("GEMINI_API_KEY missing")
# max_output_tokens via generationConfig
result = _http_post_json(
f"https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent?key={api_key}",
headers={},
body={"contents": [{"parts": [{"text": prompt}]}],
"generationConfig": {"maxOutputTokens": max_tokens}},
)
candidate = (result.get("candidates") or [{}])[0]
text = "".join(p.get("text", "") for p in (candidate.get("content") or {}).get("parts", []))
usage = result.get("usageMetadata", {})
return {
"text": text, "raw": result,
"prompt_tokens": usage.get("promptTokenCount", 0),
"completion_tokens": usage.get("candidatesTokenCount", 0),
}
_PROVIDER_FN = {
"anthropic": _complete_anthropic,
"openai": _complete_openai,
"deepseek": _complete_deepseek,
"google": _complete_google,
}
# ----- Public API -----------------------------------------------------------
def compute_cost(provider: str, model: str, prompt_tokens: int, completion_tokens: int) -> float:
price = PRICING.get((provider, model))
if not price:
return 0.0
in_p, out_p = price
return (prompt_tokens * in_p + completion_tokens * out_p) / 1_000_000
def complete(*, provider: str, model: str | None = None,
prompt: str, max_tokens: int = 1024,
task_class: str | None = None, log_trace: bool = True) -> dict:
"""Universal LLM-Completion. Auto-Trace via aria_llm_trace."""
if model is None:
model = DEFAULT_MODELS[provider]
fn = _PROVIDER_FN.get(provider)
if not fn:
raise ValueError(f"unknown provider: {provider}")
t0 = time.monotonic()
result = fn(model, prompt, max_tokens)
latency_ms = int((time.monotonic() - t0) * 1000)
cost_usd = compute_cost(provider, model,
result["prompt_tokens"], result["completion_tokens"])
out = {
**result,
"provider": provider, "model": model,
"latency_ms": latency_ms,
"cost_usd": cost_usd,
}
if log_trace:
try:
sys.path.insert(0, "/root/aria/scripts")
from aria_llm_trace import log_call, make_hash # type: ignore
log_call(
task_class=task_class,
worker_model=model, worker_family=provider,
prompt_tokens=result["prompt_tokens"],
completion_tokens=result["completion_tokens"],
cost_usd=cost_usd, latency_ms=latency_ms,
input_hash=make_hash(prompt), output_hash=make_hash(result["text"]),
privacy_tier="cloud", # router-tag wenn vorhanden
)
except Exception as exc: # noqa: BLE001
sys.stderr.write(f"[providers] trace-log failed (non-fatal): {exc}\n")
return out
def judge(*, worker_provider: str, worker_model: str,
judge_provider: str, judge_model: str | None = None,
input_text: str, output_text: str, rubric: str,
max_tokens: int = 600, mechanical_kind: str | None = None) -> dict:
"""Cross-Family-Judge (KAR-121). Erzwingt Judge ≠ Worker-Family.
Phase 2.3: Wenn `mechanical_kind` gesetzt (e.g. "json", "python"), läuft
erst Mechanical-Validation. Wenn FAIL → fail-Verdict ohne LLM-Call.
Returns dict mit verdict (pass/fail), confidence (0-1), reason, sowie
Worker- und Judge-Cost-Decomposition.
"""
if worker_provider == judge_provider:
raise ValueError(
f"Cross-Family-Constraint: judge ({judge_provider}) muss ≠ worker ({worker_provider}) sein"
)
if judge_model is None:
judge_model = DEFAULT_MODELS[judge_provider]
# Mechanical-Validation FIRST (spart Judge-Call wenn invalid)
if mechanical_kind:
mv = mechanical_validate(output_text, kind=mechanical_kind)
if not mv["is_valid"]:
return {
"judge_provider": judge_provider, "judge_model": judge_model,
"judge_cost_usd": 0.0, "judge_latency_ms": 0,
"verdict": "fail",
"confidence": 1.0,
"scores": None,
"reason": f"mechanical-validation failed ({mechanical_kind}): {mv['errors']}",
"raw": {"mechanical_only": True, "errors": mv["errors"]},
}
judge_prompt = f"""You are an evaluation judge. A worker model produced an output for a given input.
Score the output against the rubric. Return ONLY a JSON object — no prose.
Schema:
{{ "verdict": "pass" | "fail", "scores": {{...rubric_keys: 0-3}}, "confidence": 0.0-1.0, "reason": "1 sentence" }}
{rubric}
{input_text}
{output_text}
Respond with the JSON object only."""
j_out = complete(provider=judge_provider, model=judge_model, prompt=judge_prompt,
max_tokens=max_tokens, task_class="judge", log_trace=True)
# Parse JSON verdict
text = j_out["text"].strip()
# Drop markdown fences if present
if text.startswith("```"):
text = "\n".join(text.split("\n")[1:-1] if text.endswith("```") else text.split("\n")[1:])
try:
verdict_obj = json.loads(text)
except json.JSONDecodeError as exc:
verdict_obj = {"verdict": "parse_error", "confidence": 0.0,
"reason": f"json-parse-failed: {exc}", "raw_response": text[:300]}
return {
"judge_provider": judge_provider, "judge_model": judge_model,
"judge_cost_usd": j_out["cost_usd"], "judge_latency_ms": j_out["latency_ms"],
"verdict": verdict_obj.get("verdict"),
"confidence": verdict_obj.get("confidence"),
"scores": verdict_obj.get("scores"),
"reason": verdict_obj.get("reason"),
"raw": verdict_obj,
}
# ----- CLI ------------------------------------------------------------------
def _cli_complete(args) -> int:
out = complete(provider=args.provider, model=args.model,
prompt=args.prompt, max_tokens=args.max_tokens,
task_class=args.task_class)
if args.json:
print(json.dumps(out, indent=2, ensure_ascii=False)[:5000])
else:
print(out["text"])
sys.stderr.write(
f"[providers] {args.provider}/{out['model']} · "
f"tokens {out['prompt_tokens']}/{out['completion_tokens']} · "
f"cost ${out['cost_usd']:.6f} · {out['latency_ms']}ms\n"
)
return 0
def _cli_judge(args) -> int:
# Determine families
worker_fam = FAMILY_OF.get(args.worker, args.worker)
judge_fam = FAMILY_OF.get(args.judge, args.judge)
j = judge(worker_provider=worker_fam, worker_model=args.worker,
judge_provider=judge_fam, judge_model=args.judge,
input_text=args.input, output_text=args.output, rubric=args.rubric)
print(json.dumps(j, indent=2, ensure_ascii=False))
return 0
def _cli_models(args) -> int:
print(json.dumps({
"defaults": DEFAULT_MODELS,
"pricing_per_M_tokens": {f"{p}/{m}": price for (p, m), price in PRICING.items()},
"keys_present": {
"anthropic": bool(_key("ANTHROPIC_API_KEY")),
"openai": bool(_key("OPENAI_API_KEY")),
"deepseek": bool(_key("DEEPSEEK_API_KEY")),
"google": bool(_key("GEMINI_API_KEY")),
},
}, indent=2))
return 0
def main() -> int:
parser = argparse.ArgumentParser(description="Aria Multi-Provider LLM Adapter (KAR-121)")
sub = parser.add_subparsers(dest="cmd", required=True)
p_c = sub.add_parser("complete", help="single completion")
p_c.add_argument("--provider", required=True, choices=["anthropic", "openai", "deepseek", "google"])
p_c.add_argument("--model")
p_c.add_argument("--prompt", required=True)
p_c.add_argument("--max-tokens", type=int, default=400)
p_c.add_argument("--task-class")
p_c.add_argument("--json", action="store_true")
p_c.set_defaults(func=_cli_complete)
p_j = sub.add_parser("judge", help="cross-family eval-judge")
p_j.add_argument("--worker", required=True, help="worker-model id (e.g. claude-opus-4-7)")
p_j.add_argument("--judge", required=True, help="judge-model id (e.g. gpt-5)")
p_j.add_argument("--input", required=True)
p_j.add_argument("--output", required=True)
p_j.add_argument("--rubric", required=True)
p_j.set_defaults(func=_cli_judge)
p_m = sub.add_parser("models", help="list available providers + pricing + key status")
p_m.set_defaults(func=_cli_models)
args = parser.parse_args()
return args.func(args)
if __name__ == "__main__":
sys.exit(main())