Tracing

Trace: "process-document"
|-- Span: "parse-document"
|-- Span: "extract-entities"
|   |-- Generation: GPT-4o call
|-- Span: "summarize"
|   |-- Generation: GPT-4o call
|-- Span: "format-output"

from muxx import trace

@trace("my-operation")
def my_function():
    # All LLM calls inside are grouped under this trace
    pass

from muxx import Muxx

muxx = Muxx()

with muxx.trace("my-operation"):
    # All LLM calls inside are grouped under this trace
    pass

from muxx import Muxx

muxx = Muxx()

trace = muxx.start_trace("my-operation")
try:
    # Your code here
    pass
finally:
    trace.end()

from muxx import trace, span

@trace("document-processing")
def process_document(doc: str):

    @span("extraction")
    def extract():
        # LLM call for extraction
        pass

    @span("summarization")
    def summarize(data):
        # LLM call for summarization
        pass

    data = extract()
    return summarize(data)

from muxx import Muxx

muxx = Muxx()

with muxx.trace("document-processing"):
    with muxx.span("extraction"):
        # Extraction logic
        pass

    with muxx.span("summarization"):
        # Summarization logic
        pass

from muxx import trace, span

@trace("complex-operation")
def complex_operation():

    @span("step-1")
    def step_one():

        @span("step-1a")
        def sub_step():
            # Nested operation
            pass

        sub_step()

    step_one()

@trace("user-request", metadata={
    "user_id": "user_123",
    "session_id": "sess_abc",
    "feature": "chat"
})
def handle_request():
    pass

@span("llm-call", metadata={
    "model": "gpt-4o",
    "purpose": "summarization"
})
def make_llm_call():
    pass

from muxx import get_current_trace, get_current_span

@trace("my-trace")
def my_function():
    current_trace = get_current_trace()
    print(f"Trace ID: {current_trace.id}")

    with span("my-span"):
        current_span = get_current_span()
        print(f"Span ID: {current_span.id}")

import asyncio
from muxx import trace, span

@trace("async-operation")
async def async_operation():

    @span("async-step")
    async def async_step():
        await asyncio.sleep(1)

    await async_step()