belvedir

The Belvedir SDK for Python: auto-instrument your AI agent for recursive self-improvement.

The Python counterpart of belvedir on npm: same concepts (initialize, sessions, tasks, flush), Python idioms (context managers instead of callbacks).

Already using belvedir-loop? Nothing breaks: belvedir-loop@0.4.0 stays on PyPI and keeps working. It just won't get new releases. import belvedir_loop also still works after you switch, with no deprecation warning: this package ships that module too, and belvedir re-exports it.

Switching from belvedir-loop, remove it first:

pip uninstall belvedir-loop && pip install belvedir

Installing belvedir alongside an existing belvedir-loop<=0.4.0 leaves both distributions claiming the same belvedir_loop/ files. Everything imports and runs, but a later pip uninstall belvedir then deletes files the older belvedir-loop still records as its own, and import belvedir_loop breaks until you reinstall. pip install --force-reinstall belvedir repairs it.

Install

pip install belvedir

Quick Start

Call initialize() once at process startup, before your agent starts making LLM calls. Instrumentation (via OpenLLMetry) patches the installed LLM client libraries at the class level, so clients created earlier are still traced, but initializing first is the safe default:

import os
import belvedir as loop

loop.initialize(
    api_key=os.environ["BELVEDIR_API_KEY"],
    app_name="my-agent",
)

There is no instrumentModules equivalent: Python has no bundler, so auto-instrumentation always sees your installed anthropic / openai / LangChain / etc. packages.

Raw HTTP calls are covered too (0.6.0): if your agent calls an OpenAI-compatible endpoint (the Belvedir router, OpenRouter, vLLM, ...) with plain requests or httpx instead of a client SDK, initialize() captures those calls as full LLM spans, streamed (SSE) responses included, with the assistant's reply accumulated in the background while your app reads the stream. Opt out with instrument_http=False. The raw-HTTP layer watches only POST .../chat/completions: raw HTTP to Anthropic's /v1/messages wire shape, to /embeddings or /responses, and aiohttp requests are not captured; use the official client SDKs for those call sites (they always are).

Sessions: link traces so Belvedir can find tasks

This is what powers tasks and task groups. Wrap each agent run in session so every LLM and tool span is linked under one session id. Optionally wrap distinct units of work in task() for sharper task boundaries:

with loop.session(session_id=chat_id, user_id=user.id):
    with loop.task("send_email"):
        agent.run("email danny the report")

Both are plain context managers; they propagate across await boundaries, so they work inside async def bodies too. Without session you still get raw traces, but they can't be grouped into sessions or segmented into tasks. (with_session is an alias for session, mirroring the JS SDK's withSession.)

Short-lived processes: flush before exiting

Spans are exported in batches. Scripts, batch jobs, and serverless handlers can exit before the batch ships; call flush() before returning:

loop.flush()

Or pass disable_batch=True to initialize() to export every span immediately (useful for local testing).

Background workers and threads

initialize() is per process: a queue worker, forked child, or subprocess never inherits the parent's initialization, so its spans are never exported, and flush() in that process silently no-ops rather than erroring. Call initialize() at each worker process's startup; on prefork servers (gunicorn --preload, multiprocessing fork), initialize after the fork, since an exporter thread started before fork() doesn't survive into the child.

The session context is contextvars-based: it follows await, but never crosses a thread, process, or queue boundary (Celery, Temporal, multiprocessing, thread pools). Pass the session id along as part of the job's payload and re-open the session inside the worker; spans under the same id merge into one session:

# worker (its own process: initialize() ran at worker startup)
def handle_job(payload, session_id):
    with loop.session(session_id=session_id):
        run_agent(payload)  # joins the same session as the producer's spans
    loop.flush()

flush() is synchronous, so calling it immediately before os._exit() is safe; the hazard is exiting without it (os._exit and SIGKILL skip atexit and kill the exporter thread). Two session details that matter in loops: don't nest session() (an inner session replaces the outer context rather than merging), and don't use metadata keys named session_id or user_id (they'd clobber the real ones; other metadata keys land as span attributes).

Report outcomes (ground truth)

When your app knows whether a session actually worked (the order shipped, the test passed, the user accepted), tell Belvedir. Reported outcomes beat the platform's model-judged labels: a "fail" keeps every task in that session out of training data.

loop.flush()  # the session must have arrived first
loop.report_outcome(chat_id, "success" if order_shipped else "fail")

report_outcome never raises and returns whether the report was accepted (False with a warning if Belvedir hasn't seen the session yet; retry after flush(), or report later from a webhook or job).

Import traces from other platforms

Already tracing with LangSmith? Import that history instead of waiting for fresh traffic. Imported traces go through the normal ingest path, so they form sessions, tasks, and task groups, and count toward training, attributed to the API key you pass, like live traffic.

# list what would be imported
python -m belvedir.importers.langsmith \
    --langsmith-key lsv2_pt_... --project my-project \
    --belvedir-key bv_live_... --list

# import the 20 most recent conversations
python -m belvedir.importers.langsmith \
    --langsmith-key lsv2_pt_... --project my-project \
    --belvedir-key bv_live_... --limit 20

Or from Python:

from belvedir.importers.langsmith import LangSmithImporter

imp = LangSmithImporter(api_key="lsv2_pt_...", project="my-project")
conversations = imp.list_conversations(limit=20)
imp.import_conversations(
    [c.id for c in conversations],
    belvedir_api_key="bv_live_...",
)

Conversations are LangSmith traces grouped by their session_id / conversation_id / thread_id metadata; traces without any are imported individually. LLM runs keep their message history, tool runs their names, errors their status. Tasks appear ~30s after an import finishes. Re-importing the same conversation duplicates its spans; import once.

From a file (CSV / JSONL / JSON)

For datasets that never went through an observability platform (exported chat logs, fine-tuning files, spreadsheets):

# preview what the file parses into
python -m belvedir.importers.files \
    --file conversations.jsonl --belvedir-key bv_live_... --list

# import it
python -m belvedir.importers.files \
    --file conversations.jsonl --belvedir-key bv_live_...

Shapes are detected automatically. JSONL: one conversation per line, {"messages": [...]} (OpenAI chat/fine-tune format; LangChain and Anthropic message encodings work too), a bare message list, or a single {"input": ..., "output": ...} exchange (prompt/response and question/answer too); an id/session_id field names the conversation and lines sharing one merge into it. CSV: role + content columns give one message per row (grouped by a session_id-style column, or the whole file as one conversation), or an input/output-style column pair gives one exchange per row. JSON: a list of conversations or one {"messages": ...} object. Conversations without timestamps are stamped relative to import time. Same caveat as above: import a file once.

Configuration

Option	Type	Default	Description
api_key	str	required	Your Belvedir API key (bv_live_...)
base_url	str	https://platform.belvedir.ai	Belvedir platform URL
app_name	str	belvedir-loop-app	Your application name
disable_batch	bool	False	Export each span immediately instead of batching
instrument_http	bool	True	Capture LLM calls made with raw requests/httpx to OpenAI-compatible endpoints (POST .../chat/completions), including streamed responses. Official OpenAI/Anthropic client traffic is skipped (already covered). Set False to opt out

API

Function	Description
initialize(...)	Patches your LLM clients and starts exporting spans. Call once, at startup.
session(session_id, user_id=None, metadata=None)	Context manager: every span produced inside is linked under the session id. Required for tasks and task groups. with_session is an alias.
task(name)	Context manager: optional boundary hint wrapping one unit of agent work in a named span, so segmentation knows where a task starts and ends.
flush()	Forces pending spans to export. Call before a short-lived process exits. Never raises.
report_outcome(session_id, outcome)	Reports ground truth ("success" or "fail") for a session, after flush(). Never raises; returns whether the report was accepted.

Supported Providers

Anything OpenLLMetry instruments, including OpenAI, Anthropic, Cohere, Azure OpenAI, Amazon Bedrock, Google Vertex AI, Replicate, HuggingFace, and frameworks like LangChain and LlamaIndex.

Local Development

If running Belvedir locally:

loop.initialize(
    api_key="your-key",
    base_url="http://localhost:3000",
    app_name="my-app",
    disable_batch=True,  # send spans immediately
)

Troubleshooting

Traces arrive but no tasks or task groups form. Work must run inside session. Tasks appear ~30s after a session goes quiet.

No traces at all. base_url must be a host that serves the ingest API (https://platform.belvedir.ai, the default; https://belvedir.ai still works; the pre-rename platform.fractalresearch.ai is retired); any other host 404s and spans drop silently. For short-lived processes, make sure flush() runs before exit. Initialization and export failures never raise; they log warnings under the belvedir.loop logger, so check your logs.