Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.vals.ai/llms.txt

Use this file to discover all available pages before exploring further.

​
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, get_current_span, SpanType, SpanLevel

trace = get_client("my-app")
By default, traces are sent to the "default-project" project. To send traces to a different project, pass the project_slug parameter: 
trace = get_client("my-app", project_slug="my-project")

​
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. You can optionally provide a custom name for the span: 
@trace.span(name="custom_span_name")
def preprocess_query(query: str) -> str:
    return query.strip().lower()
If no name is provided, the function name is used: 
@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.

​
Usage Tracking

When using SpanType.LLM, you can track token usage with the usage parameter, which accepts a Usage TypedDict with the following fields:
  • in_tokens: Number of input tokens
  • out_tokens: Number of output tokens
  • reasoning_tokens: Number of reasoning tokens (for models that support it)
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()