Contents

Reference

PyWrapAI-Monitor

Single responsibility: record every LLM call, manage conversation history, and serve analytics. Wraps a GraphLLM — the returned MonitoredGraph has the identical API. Your application code does not change.

Quick start

llm     = LLM(provider="anthropic")
graph   = GraphLLM(llm=llm, enable_system_prompt_versioning=True, prompt_file="support-agent-v1.txt")
monitor = Monitor(db_path="./pywrapai_library.db")

tracked = monitor.wrap(graph, user_id="alice", session_id="chat-001", max_turns=10)

response = tracked.chat("I can't log in to my account.")   # history + persistence automatic

Monitor — constructor

Exactly one of db_path or store — or use Monitor.from_env().

monitor = Monitor(db_path="./pywrapai_library.db")           # SQLite, zero setup

from pywrapai_monitor.db import PostgresStore, MySQLStore
monitor = Monitor(store=PostgresStore("postgresql://user:pass@host/db"))
monitor = Monitor(store=MySQLStore("mysql+pymysql://user:pass@host/db"))

os.environ["PYWRAPAI_DB_URL"] = "postgresql://user:pass@host/db"
monitor = Monitor.from_env()   # picks the backend from the URL scheme

URL prefix	Backend used
`sqlite:/// or plain file path`	`SQLiteStore`
`postgresql:// or postgres://`	`PostgresStore`
`mysql+pymysql:// or mysql://`	`MySQLStore`

All five tables are created automatically on construction. The same Monitor can wrap multiple GraphLLM instances — one per user session in a web server, for example.

Monitor.wrap()

Parameter	Type	Required	What it does
`graph`	GraphLLM	yes	The GraphLLM instance to wrap.
`user_id`	str	yes	Who is talking — from your auth system. Can't be empty.
`session_id`	str	yes	Which conversation — typically a UUID. Can't be empty.
`max_turns`	int \| None	no (20)	Past turns to load as context. None loads full history.
`prompt_name`	str \| None	no	Load this named prompt from the DB registry — per-user pin aware.

Returns a MonitoredGraph. Raises ValueError if user_id or session_id is empty.

MonitoredGraph methods

tracked.chat(input, tools=None, tool_choice=None)

Loads history → GraphLLM.chat() → saves user + assistant rows → saves a token_records row. Six steps, fully automatic.

tracked.stream(input)

Same history loading. Not compatible with tools=. Saves both messages + a token record after the last chunk.

tracked.chat_structured(input, schema)

Same history loading and persistence as chat().

await tracked.achat(...) / tracked.astream(...)

Async versions. DB writes are fire-and-forget in the background via asyncio.create_task(asyncio.to_thread(...)) — the response returns before the write completes.

Sync paths (chat, stream, chat_structured) write synchronously — safe and correct for request/response frameworks with a long-lived event loop.

The database

One store, five tables, schema v4. Opened with PRAGMA journal_mode=WAL so the dashboard can read while the app writes.

conversations

One row per message turn.

user_id, session_id, role, content, rag_context, system_prompt, history_count

token_records

One row per LLM call — drives the whole dashboard.

user_id, session_id, model, input_tokens, output_tokens, cost_usd, duration_ms

system_prompts

Every version of every named prompt.

name, version, text, is_active

user_settings

Which prompt version each user is pinned to.

user_id, prompt_name, prompt_version

tool_calls

One row per tool call in an agent run.

turn_num, tool_name, call_id, arguments, result, error, duration_ms

What's never in history: the system prompt (sent fresh every call) and RAG context (design decision D5 — stale chunks waste tokens) are both recorded on the conversations row for observability, but load_history() only ever reads role and content. history_count records exactly how many turns were sent for that specific call, respecting max_turns at call time — not the total DB row count.

DB-backed prompt registry

An alternative to the file-based PromptRegistry — versions live in the database and can be changed at runtime with no redeploy.

monitor.save_prompt("support-agent", "v1", "You are a helpful support agent.", is_active=True)
monitor.save_prompt("support-agent", "v2", "You are an expert, concise support agent.")
monitor.save_prompt("support-agent", "v2", is_active=True)   # promote v2

text = monitor.get_prompt("support-agent")        # active version
monitor.set_user_prompt("alice", "support-agent", "v2")   # pin Alice to v2

# In wrap() — DB-backed, per-user versioned, resolved automatically
tracked = monitor.wrap(graph, user_id="alice", session_id="s1", prompt_name="support-agent")

The dashboard

monitor.serve(
    port         = 8080,
    open_browser = True,
    auth_token   = "my-secret-token",  # optional — Bearer auth on /api/*
    host         = "127.0.0.1",        # default; use "0.0.0.0" only behind a firewall
)

Parameter	Type	Default	What it does
`port`	int	8080	Port to listen on.
`open_browser`	bool	False	Open the dashboard URL after startup.
`auth_token`	str \| None	None	Requires Authorization: Bearer <token> on /api/*. Timing-safe compare.
`host`	str	"127.0.0.1"	Bind address. Combine with auth_token before switching to "0.0.0.0".

Sections

All-time metrics (calls, cost, users, sessions, tokens) · Today's metrics · a 7-day cost chart and a per-model doughnut chart · per-user and per-session tables · a recent-queries table · auto-refresh every 30s.

Conversations tab adds a filterable session list with a detail panel per message: Message text, Token Usage (model, tokens, cost, duration), System Prompt active at call time, and History Context — the exact turns sent, using history_count rather than a raw row dump. RAG context shows similarity-score bars per chunk.

REST API

`GET /api/overview`	All-time aggregates
`GET /api/today`	Today's aggregates
`GET /api/cost-over-time?days=7`	Daily cost for the last N days
`GET /api/model-usage`	Per-model call counts and costs
`GET /api/users`	Per-user analytics
`GET /api/sessions`	Per-session analytics
`GET /api/recent-queries?limit=20`	Most recent token records
`GET /api/token-records?user_id=&session_id=`	Full session token history — powers the detail panel

Store layer

BaseStore is the abstract interface — 19 methods. Three production-ready implementations ship in the box; implement your own for anything else (DynamoDB, MongoDB).

SQLiteStore

built in

SQLAlchemy Core, WAL mode, ":memory:" shortcut for tests.

PostgresStore

pip install psycopg2-binary

Production backend. Connection pooling.

MySQLStore

pip install pymysql

Production backend, MySQL / MariaDB.

Writes (3)

`save_message()`	One conversation turn — rag_context/system_prompt on user rows only.
`save_token_record()`	One analytics record, includes duration_ms.
`save_tool_step()`	All tool calls/results for one agent turn.

Reads (2)

`load_history()`	Past user/assistant messages as list[Message] — tool rows excluded.
`get_session_token_records()`	Token records for a session, chronological.

Prompt registry (4)

`save_prompt()`	Save a version; is_active=True promotes it.
`get_prompt()`	"active" resolves to the current default; KeyError if missing.
`set_user_prompt()`	Pin a user to a specific version (upsert).
`get_user_prompt_version()`	The user's pinned version, or None.

Dashboard queries (9)

`get_overview() / get_today()`	Aggregate dicts.
`get_cost_over_time(days=7)`	Daily cost series.
`get_model_usage() / get_users() / get_sessions()`	Per-model, per-user, per-session aggregates.
`get_recent_queries(limit=20)`	Most recent token records.
`get_conversation()`	Full session, all messages, dashboard use only.
`get_tool_usage()`	Per-tool call/success/error counts.

Maintenance (1)

purge_before(days) Deletes old conversations/token_records/tool_calls in one transaction.

LibraryStore is a backwards-compatible alias for SQLiteStore (LibraryStore is SQLiteStore). All writes use ON CONFLICT ... DO UPDATE SET — portable across all three backends.

Utility methods

record_embedding_build()

Persist offline indexing cost — call once after retriever.add_documents(), before any wrap() call. Ongoing query-time embedding cost is flushed automatically.

purge_before(days)

Deletes old conversations/token_records/tool_calls rows in one transaction. Raises ValueError if days < 1. Good for a weekly retention cron.

get_tool_usage(user_id=None)

Per-tool call/success/error counts, sorted by total calls descending — from the tool_calls table.

MonitoredAgent — Monitor.wrap_agent()

Parameter	Type	Required	What it does
`graph`	GraphLLM	yes	Graph with the system prompt set.
`registry`	ToolRegistry	yes	Registered tools.
`user_id / session_id`	str	yes	Who, and which agent session.
`config`	AgentConfig \| None	no	Loop configuration — defaults to AgentConfig().
`max_history`	int \| None	no (20)	Prior turns to load as context.

monitored_agent = monitor.wrap_agent(
    graph=graph, registry=registry,
    user_id="alice", session_id="agent-001",
    config=AgentConfig(max_turns=8),
)
result = monitored_agent.run("What's the weather in London?")
# .arun() mirrors it, async, with parallel tool execution

What gets written, per turn

role	content	When
`tool_call`	JSON array of tool calls for the turn	one row per AgentStep
`tool_result`	JSON array of tool results for the turn	one row per AgentStep
`user`	the original input string	once, at the end
`assistant`	the final text answer	once, at the end

One token_records row per turn (tool turns + final answer), not one per run. On AgentMaxTurnsError, the error propagates but every completed step is already saved — partial runs stay observable. load_history() filters tool rows out, so the next call's history is a clean dialogue, not a JSON dump.

Infrastructure

SQLite today — a single self-contained file, zero external infrastructure, backed up with a file copy. Already built on SQLAlchemy Core, so the Stage 2 move to PostgresStore or MySQLStore is a connection-string change, not a rewrite. PyWrapAI itself is stateless; only the store changes.