Documentation Index
Fetch the complete documentation index at: https://docs.uselumina.io/docs/llms.txt
Use this file to discover all available pages before exploring further.
Complete reference for lumina-sdk.
init_lumina(config)
Initialize the Lumina client and return a Lumina instance. Also stores the instance as a module-level singleton accessible via get_lumina().
Parameters:
| Parameter | Type | Required | Description |
|---|
config | dict | No | Configuration overrides (merged with env vars) |
config["endpoint"] | str | No | OTLP endpoint URL |
config["service_name"] | str | No | Service identifier |
config["api_key"] | str | No | Bearer token for authentication |
config["environment"] | str | No | "live" or "test" |
config["enabled"] | bool | No | Enable/disable tracing |
Returns: Lumina
Example:
from lumina import init_lumina
lumina = init_lumina({
"endpoint": "http://localhost:9411/v1/traces",
"service_name": "my-service",
})
get_lumina()
Retrieve the current module-level singleton (set by the last init_lumina() call).
Returns: Lumina | None
Example:
from lumina import get_lumina
lumina = get_lumina()
if lumina:
lumina.trace_llm(...)
Trace an LLM call with automatic attribute extraction. Handles both OpenAI and Anthropic response shapes.
Works with both sync and async callables — pass either a regular function or a coroutine function.
Parameters:
| Parameter | Type | Required | Description |
|---|
fn | Callable[[], T] | Yes | Callable returning an LLM response |
name | str | No | Span name (default: "llm.request") |
system | str | No | Provider name: "openai", "anthropic", etc. |
prompt | str | No | Input prompt text |
metadata | dict | No | Custom key-value attributes |
tags | list[str] | No | Tags for filtering |
Returns: T (sync) or Coroutine[T] (async)
Automatically extracted from response:
- Model name (
gen_ai.response.model)
- Prompt and completion tokens (
gen_ai.usage.*)
- Response text (
gen_ai.completion)
- Cost in USD (
lumina.cost_usd)
Example:
response = lumina.trace_llm(
lambda: openai_client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
),
name="chat",
system="openai",
prompt=prompt,
metadata={"user_id": "user-123"},
tags=["production"],
)
Create a span for any block of code. Use this to create parent spans for hierarchical (multi-span) traces.
Works with both sync and async callables.
Parameters:
| Parameter | Type | Required | Description |
|---|
name | str | Yes | Span name |
fn | Callable[[Span], T] | Yes | Callable receiving the active span |
metadata | dict | No | Custom key-value attributes |
tags | list[str] | No | Tags for filtering |
Returns: T (sync) or Coroutine[T] (async)
Example:
def run(span):
span.set_attribute("query", query)
return do_work()
result = lumina.trace("rag_pipeline", run)
lumina.flush()
Flush all buffered spans to the collector immediately.
Returns: Coroutine[None] (must be awaited)
import asyncio
asyncio.run(lumina.flush())
lumina.shutdown()
Flush all buffered spans and shut down the SDK.
Returns: Coroutine[None] (must be awaited)
import asyncio
asyncio.run(lumina.shutdown())
Span API
The span object passed to trace() callbacks is a standard OpenTelemetry Span.
span.set_attribute(key, value)
Add an attribute to the span.
span.set_attribute("user_id", "user-123")
span.set_attribute("cost", 0.003)
span.add_event(name, attributes?)
Record a timestamped event on the span.
span.add_event("retrieval_started")
span.add_event("checkpoint", {"progress": 50})
span.set_status(status_code, description?)
Set the span status.
from opentelemetry.trace import StatusCode
span.set_status(StatusCode.OK)
span.set_status(StatusCode.ERROR, "Failed to process request")
span.record_exception(exception)
Record an exception on the span.
try:
do_work()
except Exception as e:
span.record_exception(e)
raise