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, the FastAPI app, and the MCP server all do this at startup) and every subsequent call — local or, once the cloud branch’s paid transports land, 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. Providers report real token usage per call; until that lands (tracked on the cloud branch), cost is estimated from character counts via a fixed chars-per-token ratio (see _CHARS_PER_TOKEN) against the bundled pricing.yaml table. Local backends (Ollama, vLLM) always cost 0.0 — there is no paid API call to price. A cloud-backend call for a model absent from pricing.yaml gets cost_usd: None (unknown), never a fabricated number.

Author

Warith Harchaoui <warith.harchaoui@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

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 — on the cloud branch, once a 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