llm-observatory 0.2.7

Context0 SDK — one import, all calls logged

llm-observatory

Drop-in LLM observability for Python. One import change, all calls logged automatically.

Install

pip install llm-observatory

Usage

from llm_observatory import configure, OpenAI

configure(api_key="your-api-key")

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

Supports OpenAI, AsyncOpenAI, AzureOpenAI, AsyncAzureOpenAI, Anthropic, and AsyncAnthropic.

Wrapping third-party clients

Already using another LLM wrapper (e.g. Langfuse)? Use wrap() to add Context0 observability on top of any OpenAI- or Anthropic-compatible client — no need to change your existing setup:

from langfuse.openai import OpenAI as LangfuseOpenAI
from llm_observatory import configure, wrap

configure(api_key="your-api-key")

client = wrap(LangfuseOpenAI(api_key="sk-..."))

# Both Langfuse AND Context0 capture this call
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)

wrap() works with any client that has a .chat (OpenAI-shaped) or .messages (Anthropic-shaped) interface. If the client is already a Context0 wrapper, it's returned as-is — no double-wrapping.

Langfuse integration

Already using Langfuse? wrap_langfuse() adds a Context0 SpanProcessor to Langfuse's OTel TracerProvider so that all Langfuse observations are automatically forwarded to Context0 — with zero code changes to your existing Langfuse setup.

Requires Langfuse Python SDK v3.0.0+ (OTel-native).

Setup

from llm_observatory import configure, wrap_langfuse

configure(api_key="your-api-key")

# Call once after Langfuse is initialized — adds Context0 processor
wrap_langfuse()

What gets captured

After calling wrap_langfuse(), every Langfuse observation is automatically forwarded to Context0:

from langfuse.openai import OpenAI as LangfuseOpenAI
from langfuse import propagate_attributes

client = LangfuseOpenAI(api_key="sk-...")

# Per-request metadata (user_id, session_id, tags) — captured automatically
with propagate_attributes(user_id="alice", session_id="sess-1", tags=["prod"]):
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "What is observability?"}],
    )

# @observe decorator — captured automatically
from langfuse.decorators import observe

@observe(as_type="generation")
def my_llm_call(prompt):
    return client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
    )

# Manual traces with start_as_current_observation — captured automatically
from langfuse import Langfuse

langfuse = Langfuse()
with langfuse.start_as_current_observation(as_type="span", name="rag-pipeline"):
    # Nested spans create parent-child relationships in Context0
    with langfuse.start_as_current_observation(as_type="span", name="retrieval"):
        docs = vector_db.search(query)

    with langfuse.start_as_current_observation(as_type="generation", name="completion"):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": f"{docs}\n\n{query}"}],
        )

Works with all Langfuse v3+ patterns: @observe, LangfuseOpenAI, start_as_current_observation(), propagate_attributes(), and framework integrations (LangChain, LlamaIndex).

Tracing

Group related LLM calls into traces to see your full pipeline as a tree.

@observe decorator

Automatically creates a trace (top-level) or span (nested). Captures function args as input and return value as output.

import llm_observatory as obs

client = obs.OpenAI()

@obs.observe
def retrieve_docs(query: str):
    return vector_db.search(query)

@obs.observe
def rag_pipeline(question: str):
    docs = retrieve_docs(question)       # child span
    response = client.chat.completions.create(  # auto-captured as generation
        model="gpt-4o",
        messages=[{"role": "user", "content": f"{docs}\n\n{question}"}],
    )
    return response.choices[0].message.content

Options:

• @obs.observe(name="custom-name") — override the span name (default: function name)
• @obs.observe(capture_input=False) — don't log args (for sensitive data)
• @obs.observe(capture_output=False) — don't log return value

Works with both sync and async functions.

Context managers

Use trace() and span() for inline code blocks that aren't standalone functions:

with obs.trace(name="rag-pipeline", input={"query": q}) as t:
    with obs.span(name="vector-search") as s:
        results = search(q)
        s.set_output(results)

    response = client.chat.completions.create(model="gpt-4o", messages=[...])
    t.set_output(response.choices[0].message.content)

Mix freely — @observe and with span() compose within the same trace.

Cross-service propagation

Pass trace context across services (e.g., API to SQS worker) via W3C traceparent:

# Producer — get traceparent inside a trace
with obs.trace(name="api-request"):
    traceparent = obs.get_current_traceparent()
    sqs.send_message(Body=json.dumps({"traceparent": traceparent, ...}))

# Consumer — restore context
msg = json.loads(sqs_message["Body"])
with obs.trace(name="worker", traceparent=msg["traceparent"]):
    client.chat.completions.create(...)  # appears as child in same trace

Use obs.emit_completed_span() to backfill spans with explicit start/end times (e.g., queue wait duration).