Lightweight SDK for capturing and buffering LLM calls (OpenAI for now) for audit, compliance, and analytics.

LLMs like ChatGPT are powerful but opaque. ObservAI adds transparency and observability to your AI systems by:

• Capturing every prompt/response made to OpenAI
• Logging metadata such as latency, timestamp, route, and user
• Supporting crash-safe local buffering with SQLite
• Sending logs to a remote collector or local file
• Providing optional background flushing with low overhead

Ideal for:

• Auditing LLM usage
• Debugging hallucinations or failure cases
• Proving compliance (e.g., GDPR, HIPAA)
• Generating analytics on prompt effectiveness

• Capture prompts/responses from OpenAI automatically
• SQLite fallback for crash resilience
• Easy integration with openai SDK
• Local log file or remote collector
• Optional background flushing

pip install observai

import observai as oai

# initialize with defaults (no background flusher)
oai.init(
    user_id="alice@example.com",
    log_path="observai_events.log",
    endpoint="http://localhost:9000/events/batch",
    headers={"X-Observai-Key": "dev-secret-key"},
    flush_interval=2.0,
    background=True,  # enables automatic flush
)

# optionally tag with a user id if not set above
oai.set_user("alice@example.com")

# ... then anywhere in your code, whenever you call OpenAI:
from openai import OpenAI
client = OpenAI(api_key="YOUR_API_KEY")
resp = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role":"user","content":"Hello"}]
)

# at process exit we auto‐flush; you can also manually:
oai.flush(final=True)

You can attach custom metadata to each OpenAI request captured by ObservAI using the observai_context decorator. This is useful for tagging requests with experiment IDs, user actions, or any business context you want to track.

Import the decorator:

from observai.core.decorators import observai_context

Attach a fixed dictionary of metadata to all calls of a function:

@observai_context(metadata={"experiment": "A"})
def my_function():
    # Your OpenAI call here
    pass

Attach metadata that depends on the function's arguments:

def dynamic_metadata(*args, **kwargs):
    return {"user_action": kwargs.get("action", "unknown")}

@observai_context(metadata_func=dynamic_metadata)
def another_function(action):
    # Your OpenAI call here
    pass

@observai_context({"metadata": 1})
def call_openai():
    response = openai.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": "Hello there monkey ?"}],
    )
    print("Assistant:", response.choices[0].message.content)

• The decorator injects your metadata into the ObservAI context for the duration of the function call.
• This metadata will appear in the metadata field of each captured event (see observai_events.log or your collector endpoint).
• Works with both synchronous and asynchronous functions.

You can override defaults in init():

By default, ObservAI auto-detects and patches all supported SDKs (currently: OpenAI).

An adapter patches a third-party SDK (like OpenAI) to automatically intercept API calls.

To explicitly specify only the OpenAI adapter:

from observai.adapters.openai import OpenAIChatAdapter

observai.init(adapters=[OpenAIChatAdapter()])

✅ Compatible with openai>=1.0.0

• Source layout: follows the src/ layout best practice
• Package metadata: see pyproject.toml
• Example script: example.py
• Smoke test: tests/test_integration.py
• Pre-commit: run pre-commit install && pre-commit run --all-files

For each LLM call:

• API route (e.g. chat.completions.create)
• Model name
• Input prompt (truncated if configured)
• Output response (truncated if configured)
• Latency (ms)
• Timestamp
• User ID (if configured)
• SHA256 hash of prompt + response

• flush(final=True) sends events to the configured collector endpoint.
• flush(final=False) writes events to a local JSON-lines fallback file.
• Auto-flush mode (enabled via background=True) periodically calls flush() in a background thread.
• All buffered events are flushed on process exit via atexit.

• Anthropic adapters ??