Skip to main content

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.

Advanced patterns for complex workflows.

​
Multi-Span Tracing

Track complex workflows with hierarchical spans. Any trace or trace_llm call made inside a parent trace callback automatically becomes a child span: 
from lumina import init_lumina

lumina = init_lumina({"endpoint": "http://localhost:9411/v1/traces", "service_name": "my-app"})

def run_rag_pipeline(span):
    span.set_attribute("query", user_query)

    # Child span 1: vector search
    def retrieve(s):
        docs = vector_db.search(user_query, top_k=3)
        s.set_attribute("docs_retrieved", len(docs))
        return docs

    docs = lumina.trace("retrieval", retrieve)

    # Child span 2: LLM synthesis
    response = lumina.trace_llm(
        lambda: llm.generate(build_prompt(user_query, docs)),
        name="synthesis",
        system="anthropic",
    )

    return response

result = lumina.trace("rag_pipeline", run_rag_pipeline)

​
Multi-Span with Async

The same pattern works fully in async contexts: 
import asyncio
from lumina import init_lumina

lumina = init_lumina({"endpoint": "http://localhost:9411/v1/traces", "service_name": "my-app"})

async def run_pipeline(span):
    span.set_attribute("query", user_query)

    async def retrieve(s):
        docs = await vector_db.asearch(user_query)
        s.set_attribute("docs_retrieved", len(docs))
        return docs

    docs = await lumina.trace("retrieval", retrieve)

    response = await lumina.trace_llm(
        lambda: async_llm.generate(build_prompt(user_query, docs)),
        name="synthesis",
        system="openai",
    )

    return response

result = asyncio.run(lumina.trace("rag_pipeline", run_pipeline))

​
Span Attributes

Add custom attributes to enrich any span: 
def my_operation(span):
    span.set_attribute("user_id", "user-123")
    span.set_attribute("priority", "high")
    span.add_event("processing_started")

    # Your code here

    span.add_event("processing_completed")
    return result

lumina.trace("operation", my_operation)

​
Graceful Shutdown

Flush all buffered spans before your process exits: 
import asyncio
import signal
from lumina import init_lumina

lumina = init_lumina({"endpoint": "http://localhost:9411/v1/traces", "service_name": "my-app"})

def handle_shutdown(sig, frame):
    asyncio.run(lumina.shutdown())
    exit(0)

signal.signal(signal.SIGTERM, handle_shutdown)
signal.signal(signal.SIGINT, handle_shutdown)
For scripts, call flush() at the end: 
import asyncio

# ... your traced code ...

asyncio.run(lumina.flush())

​
Next Steps

API Reference

Complete API documentation

Multi-Span Guide

Detailed multi-span tracing guide