VisvoAI DocsPersistence
Record what an agent did — every tool call, every model call and its cost — in your own datastore, through two interfaces injected once, with no call-site wrapping.
Anything running agents in production eventually needs the audit trail: what
tools ran, with what arguments, how long they took, what each model call
cost. visvoai-core has no datastore of its own and does not want one —
recording is done through two interfaces you implement, injected once, that
default to a silent no-op. No call site is wrapped, and nothing changes in
the tools themselves.
ToolPersistence
class ToolPersistence:
def on_start(self, *, tool_id, message_id, tool_name, tool_input,
agent_step, execution_phase=None, **kwargs) -> str:
"""Called before execution. Returns the canonical tool_call_id —
you may rewrite/canonicalize it."""
return tool_id
def on_resume(self, tool_id: str) -> str:
"""Called on [checkpoint](https://langchain-ai.github.io/langgraph/reference/checkpoints/) resume; transitions a record to in-progress."""
return tool_id
def on_complete(self, *, tool_id, status, output, duration_ms, **kwargs) -> None: ...
def on_error(self, *, tool_id, error, duration_ms) -> None: ...Every method's default implementation is a no-op that returns ids
unchanged — a BaseAgentTool runs standalone, recording nothing, until you
inject a concrete subclass:
tool_instance._persistence = MyPersistence()BaseAgentTool.execute() is what actually drives this lifecycle (see
Defining tools):
on_start fires before your _execute, on_complete fires after a
successful return, and on_error fires — with the exception re-raised
afterward, never swallowed — if _execute throws.
A real example: SQLite in ~30 lines
import sqlite3
from visvoai.core.persistence import ToolPersistence
class SqliteAudit(ToolPersistence):
def __init__(self, path: str = ":memory:") -> None:
self.db = sqlite3.connect(path)
self.db.execute(
"CREATE TABLE IF NOT EXISTS tool_calls ("
" id TEXT PRIMARY KEY, tool TEXT, input TEXT,"
" status TEXT, duration_ms INT)")
def on_start(self, *, tool_id, message_id, tool_name, tool_input,
agent_step, execution_phase=None, **kw):
self.db.execute("INSERT INTO tool_calls VALUES (?,?,?,?,?)",
(tool_id, tool_name, repr(tool_input), "RUNNING", None))
return tool_id
def on_complete(self, *, tool_id, status, output, duration_ms, **kw):
self.db.execute("UPDATE tool_calls SET status=?, duration_ms=? WHERE id=?",
(status, duration_ms, tool_id))
def on_error(self, *, tool_id, error, duration_ms, **kw):
self.db.execute("UPDATE tool_calls SET status=?, duration_ms=? WHERE id=?",
(f"ERROR: {error}"[:80], duration_ms, tool_id))audit = SqliteAudit()
tool._persistence = audit
tool.execute(tool_call_id="call-1", text="audited")on_start and on_complete accept **kwargs specifically so a subclass
can widen the call sites with surface-specific fields (a message id from
your chat schema, a tenant id, …) without breaking the ToolPersistence
base contract. on_error and on_resume take fixed signatures — a
subclass can still widen them, since Python permits it.
(Full runnable file: visvoai-core/examples/06_sqlite_audit_trail.py.
The hosted platform behind these packages does exactly this into Postgres;
the CLI ships JSONL traces the same way.)
LLMPersistence
A parallel interface for recording per-call model usage/cost, independent of tool tracking:
class LLMPersistence:
def on_call_complete(self, *, message_id, model_name, input_tokens,
output_tokens, total_tokens, action,
estimated_cost_usd, tool_call_id=None) -> None: ...
def on_thinking_log(self, *, message_id, thinking_text,
tool_call_id=None) -> None: ...Pair this with visvoai.ai's usage_from / cost_of
to compute the values you pass in: usage_from(response) gives you the
token counts, cost_of(deployment_id, input, output) gives you the USD
figure for estimated_cost_usd. on_thinking_log is the hook for
recording extended-thinking content surfaced via a Provider's
normalize_content (see Providers & the model registry).
Like ToolPersistence, both methods default to no-ops; nothing is recorded
until you inject a concrete subclass into your own call sites (core itself
doesn't call LLMPersistence from inside build_graph — it's your
_build_agent_node override, or the code around your own model calls,
that invokes it after a response comes back).
ToolResult — the minimal result envelope
visvoai/core/results.py defines the return shape lifecycle tools should
use:
class ToolStatus(str, Enum):
SUCCESS = "SUCCESS"
EMPTY_RESULT = "EMPTY_RESULT"
INVALID_INPUT = "INVALID_INPUT"
TOOL_ERROR = "TOOL_ERROR"
class ToolResult(BaseModel):
tool_name: str
status: ToolStatus
result: str # model-facing text
data: Optional[Dict[str, Any]] = None # canonical payload at data["output"]Four status factory classmethods encode the output contract so payloads land where consumers expect them — prefer these over the raw constructor:
ToolResult.success(tool_name, payload, display=None, **meta)
# data = {"output": payload, **meta}; result = payload
ToolResult.invalid_input(tool_name, whats_wrong, expected=None)
# recoverable bad args, shaped so the model can self-correct
ToolResult.tool_error(tool_name, error, **meta)
# a runtime/boundary failure
ToolResult.empty(tool_name, reason, next_step=None, **meta)
# succeeded but produced nothing — always give a reasonToolResult knows nothing about HITL, citations, artifacts, streaming, or
any datastore — a consumer that needs those subclasses ToolResult and
widens status/adds fields; the core envelope stays minimal
(model_config = ConfigDict(extra="ignore") on the pydantic model means
extra keys from a subclass round-trip without validation errors on the
base shape).
Next: Tool retrieval at scale.