Skip to main content

​
Overview

The Vals SDK includes a tracing feature for monitoring and debugging operations in your applications, such as LLM calls, tool executions, and logic flows. It integrates with OpenTelemetry to export traces to the Vals platform, providing visibility into your code’s execution.

​
Introduction and Setup

Tracing allows you to track the flow of operations, capture inputs/outputs, and monitor performance. To get started, initialize a trace client with a name for your application or module: 
from vals.sdk.tracing import get_client

trace = get_client("my-app")
This creates a tracer that sends spans to your Vals project for visualization.

​
Basic Tracing with Decorators

The simplest way to add tracing is using the @trace.span decorator, which automatically captures function arguments as input and return values as output.

​
Synchronous Functions

@trace.span
def preprocess_query(query: str) -> str:
    return query.strip().lower()

​
Asynchronous Functions

@trace.span
async def fetch_data(query: str) -> dict:
    # Simulate async operation
    await asyncio.sleep(0.1)
    return {"data": "result"}
The decorator handles errors by setting the span’s level to ERROR and recording the exception message.

​
Manual Span Control

For more control, use context managers to create spans manually. This is useful for wrapping specific code blocks or operations. 
with trace.start_as_current_span("custom_operation") as span:
    # Your code here
    result = perform_task()
    span.update(input={"query": "example"}, output=result, level=SpanLevel.INFO)
You can update spans with attributes like input, output, metadata, level, and status_message at any time.

​
Span Types and Hierarchy

Spans have types to categorize operations:
  • LOGIC (default): For general logic, data processing, or workflows.
  • LLM: For language model interactions, with special attributes like model, usage, and reasoning.
  • TOOL: For external tools or API calls.
Specify the type when creating spans: 
# In decorator
@trace.span(span_type=SpanType.TOOL)
async def call_api(query: str) -> str:
    return api_request(query)

# In context manager
with trace.start_as_current_span("llm_query", span_type=SpanType.LLM) as span:
    span.update(
        model="openai/gpt-4",
        input=prompt,
        output=response,
        usage={
            "in_tokens": len(prompt.split()),
            "out_tokens": len(response.split()),
            "reasoning_tokens": len(reasoning.split()),
        },
        reasoning=reasoning,
    )
Spans can be nested to show hierarchies. For example, a parent span for an entire function can contain child spans for sub-operations.

​
Accessing and Updating Current Spans

In nested contexts, retrieve the active span to add metadata: 
def log_stats(count: int):
    current_span = get_current_span()
    current_span.update(metadata={"items_processed": count})
This is useful for updating parent spans from child functions.

​
Advanced Patterns

For multi-threaded environments like ThreadPoolExecutor, propagate context to maintain trace continuity: 
import contextvars
from concurrent.futures import ThreadPoolExecutor

@trace.span
def traced_task(data: str) -> str:
    return process(data)

with ThreadPoolExecutor() as executor:
    future = executor.submit(
        contextvars.copy_context().run,
        traced_task,
        "input_data"
    )
    result = future.result()