llm-observatory 0.1.0

Context0 SDK — one import, all calls logged

Context0 — Python SDK

Drop-in replacements for OpenAI, Azure OpenAI, and Anthropic clients. One import change, all LLM calls logged automatically.

Quick Start

from llm_observatory import configure, OpenAI

configure(api_key="sk-proj-xxx")

client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)

Supported Clients

from llm_observatory import (
    OpenAI,            # drop-in for openai.OpenAI
    AsyncOpenAI,       # drop-in for openai.AsyncOpenAI
    AzureOpenAI,       # drop-in for openai.AzureOpenAI
    AsyncAzureOpenAI,  # drop-in for openai.AsyncAzureOpenAI
    Anthropic,         # drop-in for anthropic.Anthropic
    AsyncAnthropic,    # drop-in for anthropic.AsyncAnthropic
)

All clients support chat.completions.create() (OpenAI/Azure) or messages.create() (Anthropic), including streaming. OpenAI clients also support responses.create().

Streaming

for chunk in client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
    stream=True,
):
    print(chunk.choices[0].delta.content, end="")

Streaming captures time_to_first_token_ms and tokens_per_second automatically.

Metadata

Attach metadata at the client level or per-request:

# Client-level — applies to all subsequent calls
client.set_metadata(user_id="u-123", session_id="s-456", tags=["prod"], custom={"team": "ml"})

# Per-request — merged with client-level metadata
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[...],
    metadata={"user_id": "u-789"},  # overrides client-level user_id
)

Configuration

configure(
    api_key="sk-proj-xxx",          # or set LLM_OBSERVATORY_API_KEY
    base_url="http://localhost:8080", # or set LLM_OBSERVATORY_BASE_URL
    enabled=True,                    # or set LLM_OBSERVATORY_ENABLED=false to disable
    redact_messages=False,           # redact all message content from logged events
    flush_interval=5.0,             # seconds between batch flushes
    flush_batch_size=50,            # max events per batch
    max_queue_size=1000,            # max queued events (oldest dropped on overflow)
)

Environment Variables

Variable	Description
LLM_OBSERVATORY_API_KEY	API key (used if not passed to configure())
LLM_OBSERVATORY_BASE_URL	Ingestion endpoint URL
LLM_OBSERVATORY_ENABLED	Set to false to disable all logging

Flushing

Events are batched and sent asynchronously in a background thread. To force-flush:

client.flush()  # blocks until all buffered events are sent

Design Principles

• Not in the critical path. LLM calls go directly to the provider. Logging is async in a background thread.
• Never crashes user code. Event capture is try/catch wrapped. Exceptions from the LLM provider are re-raised unchanged.
• Transparent. Same interface as the original clients. Returns original responses unchanged.

Structure

sdk-python/
├── pyproject.toml
├── src/llm_observatory/
│   ├── __init__.py          # Public API: configure, OpenAI, Anthropic, etc.
│   ├── client.py            # Client wrappers (OpenAI, Azure, Anthropic)
│   ├── streaming.py         # Stream iterator wrappers
│   ├── transport.py         # Async batch flusher (daemon thread)
│   ├── event.py             # Event schema + redaction
│   └── config.py            # Configuration
└── tests/

Install (dev)

cd sdk-python
pip install -e ".[dev]"
pytest tests/ -v