Skip to main content
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:
ParameterTypeRequiredDescription
configdictNoConfiguration overrides (merged with env vars)
config["endpoint"]strNoOTLP endpoint URL
config["service_name"]strNoService identifier
config["api_key"]strNoBearer token for authentication
config["environment"]strNo"live" or "test"
config["enabled"]boolNoEnable/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(...)

​
lumina.trace_llm(fn, *, name, system, prompt, metadata, tags)

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:
ParameterTypeRequiredDescription
fnCallable[[], T]YesCallable returning an LLM response
namestrNoSpan name (default: "llm.request")
systemstrNoProvider name: "openai", "anthropic", etc.
promptstrNoInput prompt text
metadatadictNoCustom key-value attributes
tagslist[str]NoTags 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"],
)

​
lumina.trace(name, fn, *, metadata, tags)

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:
ParameterTypeRequiredDescription
namestrYesSpan name
fnCallable[[Span], T]YesCallable receiving the active span
metadatadictNoCustom key-value attributes
tagslist[str]NoTags 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