Decorators

from muxx import observe
from openai import OpenAI

client = OpenAI()

@observe()
def get_completion(prompt: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

@observe("summarize-text")
def summarize(text: str) -> str:
    # Function name in dashboard will be "summarize-text"
    pass

@observe("user-chat", metadata={"feature": "support"})
def handle_chat(user_id: str, message: str):
    pass

from muxx import trace

@trace("document-pipeline")
def process_document(doc_id: str):
    # All operations inside create a new trace
    extract_text(doc_id)
    analyze_content(doc_id)
    generate_summary(doc_id)

@trace(
    name="user-request",
    metadata={"environment": "production"},
    capture_input=True,   # Log function arguments
    capture_output=True   # Log return value
)
def handle_request(request_data: dict):
    pass

from muxx import trace, span

@trace("multi-step-process")
def process():

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

    @span("step-2")
    def step_two():
        pass

    step_one()
    step_two()

from muxx import generation

@generation(model="gpt-4o", provider="openai")
def call_llm(prompt: str):
    # Useful when using a custom LLM client
    # that Muxx doesn't automatically detect
    pass

from muxx import trace, span, observe

@trace("api-request")
def handle_api_request(request):

    @span("validate")
    def validate():
        pass

    @span("process")
    @observe()  # This span is also observed
    def process():
        # LLM call here
        pass

    validate()
    return process()

from muxx import trace, span

@trace("async-pipeline")
async def async_pipeline():

    @span("async-step")
    async def async_step():
        await some_async_operation()

    await async_step()

from muxx import trace, observe

class DocumentProcessor:

    @trace("process-document")
    def process(self, document: str):
        return self._summarize(document)

    @observe("summarize")
    def _summarize(self, text: str):
        # LLM call
        pass

from muxx import observe

@observe("risky-operation")
def risky_operation():
    raise ValueError("Something went wrong")
    # Error is captured in the trace with full stack trace

from muxx import observe, get_current_span

@observe("operation")
def operation():
    try:
        # risky code
        pass
    except Exception as e:
        span = get_current_span()
        span.set_error(e, metadata={"recovery_attempted": True})
        raise