best_engine_ai_helper.observe module

observe — a local, append-only activity/cost ledger for llm.chat().

Every best_engine_ai_helper.llm.chat() call already emits a small event dict to any observer registered via llm.add_observer(); nothing consumes it by default. This module is that consumer: call enable() once (the CLI’s activity command and the MCP server do this at startup; the FastAPI app does NOT — GET /api/activity reads the ledger directly instead, see api.activity(), so it works even under a plain uvicorn launch that never called enable()) and every subsequent call — local (always free) or cloud (a real paid request through OpenAI/Mistral/OpenRouter/Together/Azure/Anthropic/Gemini, see best_engine_ai_helper.engine’s mode: cloud) — is appended to a SQLite database at ~/.best-engine-ai-helper/usage.db.

The point is the “one company, several users” case: a shared machine or a small internal server fields calls from more than one person, and someone needs to answer “who is calling what, how often, and at what cost” without standing up a separate telemetry stack. This ledger is local-only (no network call, no third-party service) and off by default.

Cost is a best-effort estimate, not a provider-verified bill. Most providers report real token usage per call; when that field is present it is used directly, otherwise cost is estimated from character counts via a fixed chars-per-token ratio (see _CHARS_PER_TOKEN) against the paid price table, read through cloud_catalog.load_cloud_catalog() so that a catalog update --cloud refresh reaches the bill and not only the ranking. Local backends (Ollama, vLLM) always cost 0.0 — there is no paid API call to price. A cloud-backend call for a model the table does not price gets cost_usd: None (unknown), never a fabricated number.

What a local 0.0 means is “nothing was billed”, not “nothing was spent”: electricity, disk and the slower model’s time are real and this ledger has no way to measure them. Ledger.counterfactual() answers the question that is answerable from what is recorded — what the same workload would have cost at a provider’s list price.

Author

Warith Harchaoui <warithmetics@deraison.ai>

class best_engine_ai_helper.observe.Ledger(db_path=None)

Bases: object

Append-only SQLite sink for llm.chat() observer events.

Parameters:

db_path (str or Path or None) – Where to store the database. Defaults to ~/.best-engine-ai-helper/usage.db; :memory: is accepted for tests. The parent directory is created if missing.

close()

Close the underlying SQLite connection.

Return type:

None

counterfactual()

What the free calls in this ledger would have cost at list price.

The field around this project is full of local-versus-cloud break-even calculators: type in a guessed monthly token volume, read back a payback period. The guess is the weak part — nobody knows their volume before they have run the workload, and the number that comes out inherits whatever was typed in.

This ledger already holds the workload, measured. Crossing it with the same price table estimate_cost_usd() bills paid calls against turns the question around: not “what might you spend”, but “what did the calls you already made avoid”. Two sources, both already here, and no invented volume.

Three things it deliberately does not claim:

• It is not a saving. Running locally costs electricity, disk and the time a slower model takes. This figure is an avoided bill, not a profit, and the ledger has no business inventing a watt.

• It is not the same answer. A paid model given the same prompts would not emit the same tokens, so the output volume is this machine’s, carried across as the closest thing to a workload anyone has measured.

• The token counts are approximate. A local backend reports no usage block, so the characters recorded per call go through _CHARS_PER_TOKEN exactly as they do for an unreported paid call.

Returns:

free_calls, in_tokens, out_tokens (both estimated), and models: one row per priced catalog entry (model, provider, cost_usd), cheapest first. Empty models when nothing in the catalog carries a price.

Return type:

dict

Examples

>>> led = Ledger(":memory:")
>>> led.record({"backend": "ollama", "model": "qwen3:8b", "kind": "llm",
...             "in_chars": 4000, "out_chars": 4000, "latency_ms": 1.0,
...             "ok": True})
>>> what_if = led.counterfactual()
>>> what_if["free_calls"], what_if["in_tokens"], what_if["out_tokens"]
(1, 1000, 1000)
>>> cheapest = what_if["models"][0]
>>> cheapest["cost_usd"] >= 0.0
True
>>> led.close()

An empty ledger has nothing to price, and says so with zeros rather than with a catalog of zero-cost rows:

>>> empty = Ledger(":memory:")
>>> empty.counterfactual()["models"]
[]
>>> empty.close()

record(event)

Persist one llm.chat() event, enriched with user and cost.

Parameters:

event (dict) – One event dict as emitted by llm.chat().

Return type:

None

summary()

Aggregate the ledger into the figures a usage dashboard needs.

Returns:

total_calls, total_cost_usd (0.0 for an empty ledger, None if any recorded call has an unpriced model — an unknown component must never be silently dropped from the total), error_rate (0-1), by_user / by_model (call count + cost per key, sorted by call count descending), and recent_errors (last 10 failed calls: ts, user, model, error).

Return type:

dict

best_engine_ai_helper.observe.active_ledger()

Return the active ledger, or None if enable() was never called.

Return type:

Ledger | None

class best_engine_ai_helper.observe.as_user(name)

Bases: object

Context manager: attribute every ledger entry inside the block to name.

Parameters:

name (str) – Identity to record for calls made inside this block.

Examples

>>> with as_user("alice"):
...     current_user()
'alice'
>>> current_user() != "alice"
True

best_engine_ai_helper.observe.current_user()

Resolve the identity to attribute the next ledger entry to.

Precedence: as_user() scope > BEST_ENGINE_USER env var > OS login name > "unknown" (only when even the OS refuses to say).

Returns:

The resolved user identity. Never empty.

Return type:

str

Examples

>>> isinstance(current_user(), str) and current_user() != ""
True

best_engine_ai_helper.observe.disable()

Stop recording and close the active ledger, if any.

Clears every registered llm observer, not only this module’s — today the ledger is the only observer type in the suite, so this is equivalent in practice, but a caller with its own observers registered separately would need to re-register them after calling this.

Return type:

None

best_engine_ai_helper.observe.enable(db_path=None)

Start recording every llm.chat() call to a ledger.

Idempotent: calling this again returns the already-active ledger rather than registering a second observer.

Parameters:

db_path (str or Path or None) – Forwarded to Ledger. Ignored if a ledger is already active.

Returns:

The active ledger, for direct querying (Ledger.summary()).

Return type:

Ledger

best_engine_ai_helper.observe.estimate_cost_usd(event)

Estimate the USD cost of one llm.chat() event.

Parameters:

event (dict) – One event dict as emitted by llm.chat() (backend, model, in_chars, out_chars, and — when the transport’s provider reports them — in_tokens/out_tokens).

Returns:

0.0 for a local backend (always free). For a paid backend: the cost from pricing.yaml using the provider’s own reported token counts when the event carries them (in_tokens/out_tokens, exact), falling back to the character-count heuristic otherwise (approximate — see _CHARS_PER_TOKEN). None when the model is not in the table — an unpriced model must never silently show as free.

Return type:

float or None

Examples

>>> estimate_cost_usd({"backend": "ollama", "model": "qwen3:8b",
...                     "in_chars": 100, "out_chars": 100})
0.0
>>> estimate_cost_usd({"backend": "openai", "model": "not-in-table",
...                     "in_chars": 100, "out_chars": 100}) is None
True

best_engine_ai_helper.observe.is_enabled()

Return whether a ledger is currently recording.

Return type:

bool