From 372d2dcff3cbc9cdd4a24d77dd480beb29495746 Mon Sep 17 00:00:00 2001 From: Thibault Date: Wed, 15 Jul 2026 00:10:03 +0200 Subject: [PATCH 1/4] feat: add provider-neutral LLM decision ledger --- scripts/python/llm_decision_ledger.py | 182 ++++++++++++++++++++++++++ 1 file changed, 182 insertions(+) create mode 100644 scripts/python/llm_decision_ledger.py diff --git a/scripts/python/llm_decision_ledger.py b/scripts/python/llm_decision_ledger.py new file mode 100644 index 0000000..4641503 --- /dev/null +++ b/scripts/python/llm_decision_ledger.py @@ -0,0 +1,182 @@ +"""Provider-neutral decision ledger for LLM engineering workflows. + +This module does not proxy model traffic. It records routing intent, outcomes, +and evidence so an existing gateway such as LiteLLM, OpenRouter, a direct SDK, +or an internal platform can make auditable and continuously improving choices. +""" + +from __future__ import annotations + +import hashlib +import json +import sqlite3 +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Iterable, Optional + + +@dataclass(frozen=True) +class Decision: + request_id: str + task_type: str + selected_model: str + alternative_models: tuple[str, ...] + reason: str + estimated_cost_usd: float + risk_level: str = "medium" + + +@dataclass(frozen=True) +class Outcome: + request_id: str + success: bool + latency_ms: int + actual_cost_usd: float + quality_score: Optional[float] = None + reviewer: str = "automatic" + notes: str = "" + + +class DecisionLedger: + """Append-only SQLite ledger with simple model evidence summaries.""" + + def __init__(self, database: str | Path = "llm_decisions.sqlite3") -> None: + self.database = str(database) + self._initialize() + + def _connect(self) -> sqlite3.Connection: + connection = sqlite3.connect(self.database) + connection.row_factory = sqlite3.Row + return connection + + def _initialize(self) -> None: + with self._connect() as connection: + connection.executescript( + """ + CREATE TABLE IF NOT EXISTS decisions ( + request_id TEXT PRIMARY KEY, + created_at TEXT NOT NULL, + task_type TEXT NOT NULL, + selected_model TEXT NOT NULL, + alternative_models TEXT NOT NULL, + reason TEXT NOT NULL, + estimated_cost_usd REAL NOT NULL CHECK(estimated_cost_usd >= 0), + risk_level TEXT NOT NULL, + integrity_hash TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS outcomes ( + request_id TEXT PRIMARY KEY REFERENCES decisions(request_id), + created_at TEXT NOT NULL, + success INTEGER NOT NULL, + latency_ms INTEGER NOT NULL CHECK(latency_ms >= 0), + actual_cost_usd REAL NOT NULL CHECK(actual_cost_usd >= 0), + quality_score REAL, + reviewer TEXT NOT NULL, + notes TEXT NOT NULL + ); + """ + ) + + @staticmethod + def _canonical_payload(decision: Decision) -> str: + payload = asdict(decision) + payload["alternative_models"] = list(decision.alternative_models) + return json.dumps(payload, sort_keys=True, separators=(",", ":")) + + @classmethod + def integrity_hash(cls, decision: Decision) -> str: + return hashlib.sha256(cls._canonical_payload(decision).encode("utf-8")).hexdigest() + + def record_decision(self, decision: Decision) -> str: + if not decision.request_id.strip(): + raise ValueError("request_id cannot be empty") + if decision.estimated_cost_usd < 0: + raise ValueError("estimated_cost_usd cannot be negative") + + digest = self.integrity_hash(decision) + with self._connect() as connection: + connection.execute( + """ + INSERT INTO decisions VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + decision.request_id, + datetime.now(timezone.utc).isoformat(), + decision.task_type, + decision.selected_model, + json.dumps(decision.alternative_models), + decision.reason, + decision.estimated_cost_usd, + decision.risk_level, + digest, + ), + ) + return digest + + def record_outcome(self, outcome: Outcome) -> None: + if outcome.latency_ms < 0 or outcome.actual_cost_usd < 0: + raise ValueError("latency and cost must be non-negative") + if outcome.quality_score is not None and not 0 <= outcome.quality_score <= 1: + raise ValueError("quality_score must be between 0 and 1") + + with self._connect() as connection: + exists = connection.execute( + "SELECT 1 FROM decisions WHERE request_id = ?", (outcome.request_id,) + ).fetchone() + if not exists: + raise KeyError(f"unknown request_id: {outcome.request_id}") + connection.execute( + """ + INSERT INTO outcomes VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + outcome.request_id, + datetime.now(timezone.utc).isoformat(), + int(outcome.success), + outcome.latency_ms, + outcome.actual_cost_usd, + outcome.quality_score, + outcome.reviewer, + outcome.notes, + ), + ) + + def model_evidence(self, task_type: Optional[str] = None) -> list[dict[str, object]]: + filters = "WHERE d.task_type = ?" if task_type else "" + params: Iterable[object] = (task_type,) if task_type else () + query = f""" + SELECT + d.selected_model AS model, + COUNT(*) AS samples, + AVG(o.success) AS success_rate, + AVG(o.latency_ms) AS average_latency_ms, + AVG(o.actual_cost_usd) AS average_cost_usd, + AVG(o.quality_score) AS average_quality_score + FROM decisions d + JOIN outcomes o ON o.request_id = d.request_id + {filters} + GROUP BY d.selected_model + ORDER BY average_quality_score DESC, success_rate DESC, average_cost_usd ASC + """ + with self._connect() as connection: + return [dict(row) for row in connection.execute(query, tuple(params)).fetchall()] + + def verify(self, request_id: str) -> bool: + with self._connect() as connection: + row = connection.execute( + "SELECT * FROM decisions WHERE request_id = ?", (request_id,) + ).fetchone() + if row is None: + return False + decision = Decision( + request_id=row["request_id"], + task_type=row["task_type"], + selected_model=row["selected_model"], + alternative_models=tuple(json.loads(row["alternative_models"])), + reason=row["reason"], + estimated_cost_usd=row["estimated_cost_usd"], + risk_level=row["risk_level"], + ) + return self.integrity_hash(decision) == row["integrity_hash"] From 2410bb68fdabc4015d11a778e444076d6d2345d4 Mon Sep 17 00:00:00 2001 From: Thibault Date: Wed, 15 Jul 2026 00:10:29 +0200 Subject: [PATCH 2/4] test: cover LLM decision ledger --- .../python/tests/test_llm_decision_ledger.py | 61 +++++++++++++++++++ 1 file changed, 61 insertions(+) create mode 100644 scripts/python/tests/test_llm_decision_ledger.py diff --git a/scripts/python/tests/test_llm_decision_ledger.py b/scripts/python/tests/test_llm_decision_ledger.py new file mode 100644 index 0000000..4920299 --- /dev/null +++ b/scripts/python/tests/test_llm_decision_ledger.py @@ -0,0 +1,61 @@ +import tempfile +import unittest +from pathlib import Path + +from llm_decision_ledger import Decision, DecisionLedger, Outcome + + +class DecisionLedgerTests(unittest.TestCase): + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + self.ledger = DecisionLedger(Path(self.temp_dir.name) / "ledger.sqlite3") + self.decision = Decision( + request_id="req-001", + task_type="code-review", + selected_model="model-a", + alternative_models=("model-b",), + reason="Best historical quality under the cost ceiling", + estimated_cost_usd=0.02, + risk_level="high", + ) + + def tearDown(self): + self.temp_dir.cleanup() + + def test_records_and_verifies_decision(self): + digest = self.ledger.record_decision(self.decision) + self.assertEqual(64, len(digest)) + self.assertTrue(self.ledger.verify("req-001")) + + def test_records_outcome_and_builds_evidence(self): + self.ledger.record_decision(self.decision) + self.ledger.record_outcome( + Outcome( + request_id="req-001", + success=True, + latency_ms=800, + actual_cost_usd=0.018, + quality_score=0.9, + ) + ) + evidence = self.ledger.model_evidence("code-review") + self.assertEqual(1, len(evidence)) + self.assertEqual("model-a", evidence[0]["model"]) + self.assertAlmostEqual(0.9, evidence[0]["average_quality_score"]) + + def test_rejects_outcome_without_decision(self): + with self.assertRaises(KeyError): + self.ledger.record_outcome( + Outcome("missing", True, 10, 0.0, quality_score=1.0) + ) + + def test_rejects_invalid_quality_score(self): + self.ledger.record_decision(self.decision) + with self.assertRaises(ValueError): + self.ledger.record_outcome( + Outcome("req-001", True, 10, 0.0, quality_score=1.5) + ) + + +if __name__ == "__main__": + unittest.main() From 9177825c3141a49b0851b976adf974c90cbab5e8 Mon Sep 17 00:00:00 2001 From: Thibault Date: Wed, 15 Jul 2026 00:10:57 +0200 Subject: [PATCH 3/4] docs: explain complementary decision intelligence layer --- docs/llm-decision-ledger.md | 77 +++++++++++++++++++++++++++++++++++++ 1 file changed, 77 insertions(+) create mode 100644 docs/llm-decision-ledger.md diff --git a/docs/llm-decision-ledger.md b/docs/llm-decision-ledger.md new file mode 100644 index 0000000..21a29a9 --- /dev/null +++ b/docs/llm-decision-ledger.md @@ -0,0 +1,77 @@ +# LLM Decision Ledger + +## Positioning + +This component is deliberately **not** another model proxy or provider marketplace. +LiteLLM, OpenRouter, direct provider SDKs, and internal gateways remain responsible +for authentication, transport, retries, streaming, and model invocation. + +The Decision Ledger adds a provider-neutral intelligence and governance layer: + +1. Record the model selected for a task and the alternatives considered. +2. Preserve the reason, estimated cost, and risk level before execution. +3. Attach real latency, cost, success, and reviewed quality after execution. +4. Build evidence by task type instead of relying only on generic benchmarks. +5. Verify that stored routing decisions were not silently modified. + +## Why it is different + +A router answers: **Which endpoint receives this request now?** + +The ledger answers: + +- Why was this model appropriate for this precise business task? +- Did the choice deliver the expected quality, cost, and latency? +- Is there enough internal evidence to automate this choice later? +- Can an auditor or customer understand the decision after the fact? + +This makes the project complementary to existing gateways and useful for AI +engineering teams operating several providers, local models, or sensitive clients. + +## Minimal integration + +```python +from llm_decision_ledger import Decision, DecisionLedger, Outcome + +ledger = DecisionLedger("data/llm_decisions.sqlite3") +ledger.record_decision( + Decision( + request_id="ticket-1842", + task_type="powershell-security-review", + selected_model="internal-secure-model", + alternative_models=("provider-model-a", "provider-model-b"), + reason="Customer data must remain local; model passed prior security reviews", + estimated_cost_usd=0.01, + risk_level="high", + ) +) + +# Invoke the model through LiteLLM, OpenRouter, a direct SDK, or another gateway. + +ledger.record_outcome( + Outcome( + request_id="ticket-1842", + success=True, + latency_ms=920, + actual_cost_usd=0.009, + quality_score=0.88, + reviewer="security-reviewer", + ) +) + +print(ledger.model_evidence("powershell-security-review")) +``` + +## SaaS direction + +A first sellable product can expose this ledger through an API and dashboard with: + +- evidence cards per task, customer, and model; +- explainable routing recommendations before execution; +- shadow comparisons that never send production traffic automatically; +- human review workflows for high-risk outputs; +- exportable governance reports for customers and audits; +- adapters for LiteLLM, OpenRouter, Azure OpenAI, local Ollama, and direct SDKs. + +The commercial differentiator is not cheaper API forwarding. It is **decision +intelligence for reliable multi-model AI engineering**. From 5fcb2be29d28e15e2ce43d11f7409587a985f9ef Mon Sep 17 00:00:00 2001 From: Thibault Date: Fri, 17 Jul 2026 10:48:04 +0200 Subject: [PATCH 4/4] fix: make decision ledger tests importable in CI --- scripts/python/tests/test_llm_decision_ledger.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/scripts/python/tests/test_llm_decision_ledger.py b/scripts/python/tests/test_llm_decision_ledger.py index 4920299..6ec3ebd 100644 --- a/scripts/python/tests/test_llm_decision_ledger.py +++ b/scripts/python/tests/test_llm_decision_ledger.py @@ -1,7 +1,11 @@ +import sys import tempfile import unittest from pathlib import Path +# unittest discovery starts from the repository root, so expose the sibling module. +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + from llm_decision_ledger import Decision, DecisionLedger, Outcome