prism-sdk

LLM Observability in 3 lines. Costs, latency, errors and quality — automatically.

pip install useprism

Quick Start

With Anthropic

import anthropic
import prism

# 1. Init
prism.init("prism_your_api_key")

# 2. Wrap your client
client = prism.wrap(anthropic.Anthropic())

# 3. Use exactly as before — everything is traced automatically
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Summarise this contract..."}],
)

With OpenAI

import openai
import prism

prism.init("prism_your_api_key")
client = prism.wrap(openai.OpenAI())

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

Streaming

# Anthropic — as an iterator
stream = client.messages.create(model="claude-sonnet-4-6", stream=True,
                                max_tokens=1024, messages=[...])
for event in stream:
    ...

# Anthropic — as a context manager
with client.messages.stream(model="claude-sonnet-4-6", max_tokens=1024, messages=[...]) as s:
    for text in s.text_stream:
        print(text, end="")

The trace is recorded when the stream finishes — output tokens are only known at that point. Events pass through untouched.

Streaming with OpenAI: their API only returns usage if you ask for it. Without it, token counts and cost come back as zero, and the trace is flagged with metadata.usage_missing so you can tell "it was free" apart from "it could not be measured":

stream = client.chat.completions.create(
    model="gpt-4o", stream=True, messages=[...],
    stream_options={"include_usage": True},   # without this, no cost data
)

Async clients

import anthropic, prism

prism.init("prism_your_api_key")
client = prism.wrap(anthropic.AsyncAnthropic())

response = await client.messages.create(model="claude-sonnet-4-6",
                                        max_tokens=1024, messages=[...])

# async streaming
stream = await client.messages.create(stream=True, ...)
async for event in stream:
    ...

AsyncOpenAI works the same way.

When a call fails

Your exception is re-raised untouched — Prism never swallows errors. The trace is recorded first, with the provider's status code and the exception type and message under metadata.error, so a failed call shows why rather than just 500.

try:
    client.messages.create(...)
except anthropic.RateLimitError:
    ...   # your handling is unaffected; the trace already recorded 429

With a custom / unsupported client

import prism
from prism.manual import trace

prism.init("prism_your_api_key")

with trace("my-model", endpoint="/api/summarize") as t:
    result = my_custom_llm_call(prompt)
    t.set_tokens(prompt=500, completion=200)
    t.set_response(result.text)

Configuration

prism.init(
    api_key="prism_your_api_key",
    debug=True,           # print logs to stdout
    flush_interval=2.0,   # seconds between batch flushes
    batch_size=20,        # traces per batch request
)

What gets tracked automatically

FastAPI integration example

from fastapi import FastAPI, Request
import anthropic
import prism

prism.init("prism_your_api_key")
app = FastAPI()

@app.post("/api/chat")
async def chat(request: Request):
    body = await request.json()

    # Wrap per-request for endpoint-level tracking
    client = prism.wrap(
        anthropic.Anthropic(),
        endpoint="/api/chat",
    )

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": body["message"]}],
    )
    return {"reply": response.content[0].text}

Overhead

• ~0ms added to your LLM calls (async background thread)
• Traces are batched and sent every 2 seconds
• On shutdown, all pending traces are flushed

License

MIT