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**. 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"] 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..6ec3ebd --- /dev/null +++ b/scripts/python/tests/test_llm_decision_ledger.py @@ -0,0 +1,65 @@ +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 + + +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()