Skip to main content
These recipes cover the core tracing patterns. For conceptual background, see Observability & Tracing.

Setup

from abvdev import ABV, observe

abv = ABV(api_key="sk-abv-...")

Decorator Pattern

Wrap existing functions without modifying their internals. Input/output are captured automatically.
from abvdev import observe

@observe()
def process_query(query: str) -> str:
    # Your logic here
    return f"Processed: {query}"

@observe(name="llm-generation", as_type="generation")
def generate_response(prompt: str) -> str:
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

# Just call normally - tracing happens automatically
result = process_query("What is AI?")
response = generate_response("Explain machine learning")

Context Manager Pattern

Automatic lifecycle management with explicit control over span attributes.
from abvdev import ABV

abv = ABV(api_key="sk-abv-...")

with abv.start_as_current_span(name="user-request") as span:
    span.update(input={"query": "What is the capital of France?"})

    # Nested generation
    with abv.start_as_current_generation(
        name="llm-call",
        model="gpt-4",
        model_parameters={"temperature": 0.7}
    ) as gen:
        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": "What is the capital of France?"}]
        )
        gen.update(
            output=response.choices[0].message.content,
            usage_details={
                "prompt_tokens": response.usage.prompt_tokens,
                "completion_tokens": response.usage.completion_tokens,
            }
        )

    span.update(output="Paris")

Manual Spans

Full control over span creation, nesting, and lifecycle.
from abvdev import ABV

abv = ABV(api_key="sk-abv-...")

# Create root span
span = abv.start_span(name="pipeline")
span.update(input={"query": "Search for documents"})

try:
    # Create child spans
    retriever = abv.start_span(name="retrieve-docs")
    retriever.update(input={"query": "Search for documents"})
    docs = search_documents("Search for documents")
    retriever.update(output={"doc_count": len(docs)})
    retriever.end()

    # Generation span
    gen = abv.start_span(name="generate-answer")
    gen.update(
        input={"docs": docs},
        metadata={"model": "gpt-4"}
    )
    answer = generate_answer(docs)
    gen.update(output=answer)
    gen.end()

    span.update(output=answer)
finally:
    span.end()
    abv.flush()

Add User & Session Context

Track users and group related traces into sessions.
@observe()
def handle_request(user_id: str, session_id: str, query: str):
    # Update trace-level context
    abv.update_current_trace(
        user_id=user_id,
        session_id=session_id,
        metadata={"source": "web-app"},
        tags=["production", "chat"]
    )

    return process_query(query)

handle_request("user-123", "session-456", "Hello!")

Observation Types

Use specific types for better categorization in the ABV UI.
TypeUse Case
spanGeneral operations (default)
generationLLM API calls
embeddingText embeddings
retrieverDocument/vector search
toolExternal API calls, function invocations
agentAgent workflows
chainMulti-step pipelines
evaluatorQuality assessment
guardrailSafety/validation checks
# Retriever
with abv.start_as_current_observation(as_type="retriever", name="search") as obs:
    docs = vector_db.search(query)
    obs.update(output={"documents": docs})

# Tool
with abv.start_as_current_observation(as_type="tool", name="api-call") as obs:
    result = external_api.call(params)
    obs.update(output=result)

# Agent
with abv.start_as_current_observation(as_type="agent", name="assistant") as obs:
    response = agent.run(user_input)
    obs.update(output=response)

Error Handling

Capture errors with appropriate log levels.
@observe()
def risky_operation(data):
    try:
        result = process(data)
        return result
    except Exception as e:
        abv.update_current_span(
            level="ERROR",
            status_message=str(e),
            output={"error": str(e)}
        )
        raise

# Or with context manager
with abv.start_as_current_span(name="operation") as span:
    try:
        result = process(data)
        span.update(output=result)
    except Exception as e:
        span.update(level="ERROR", status_message=str(e))
        raise

Get Trace URL

Share traces by getting the direct URL.
@observe()
def my_function():
    trace_id = abv.get_current_trace_id()
    trace_url = abv.get_trace_url(trace_id=trace_id)
    print(f"View trace: {trace_url}")
    return "done"