Tracer

Capture execution traces and LLM performance metrics for your application.

Tracer is the primary way to add observability to AI agents and LLM pipelines. It records spans (units of work), automatically captures inputs/outputs, and exports everything to the Judgment dashboard.

Getting started:

1. Call Tracer.init() to create and activate a tracer.
2. Decorate your functions with @Tracer.observe() to trace them.
3. Optionally wrap LLM clients with Tracer.wrap() for automatic token/cost tracking.

Basic setup and usage:

from judgeval import Tracer

tracer = Tracer.init(project_name="search-assistant")

@Tracer.observe(span_type="tool")
def search(query: str) -> str:
    return vector_db.search(query)

@Tracer.observe(span_type="agent")
async def answer(question: str) -> str:
    context = search(question)
    return await llm.generate(question, context)

Wrap an LLM client for automatic instrumentation:

from openai import OpenAI

openai = Tracer.wrap(OpenAI())

Attributes

init()

Create and activate a new Tracer.

This is the recommended way to initialize tracing. Credentials are read from environment variables (JUDGMENT_API_KEY, JUDGMENT_ORG_ID, JUDGMENT_API_URL) when not passed explicitly. If credentials are missing, the tracer still works but spans won’t be exported.

tracer = Tracer.init(
    project_name="search-assistant",
    environment="production",
)

def init(project_name=None, api_key=None, organization_id=None, api_url=None, environment=None, set_active=True, serializer=safe_serialize, resource_attributes=None, sampler=None, span_limits=None, span_processors=None) -> Tracer:

Parameters

Returns

Tracer - A configured and active Tracer instance.

set_active()

Set this tracer as the globally active tracer.

def set_active() -> bool:

Returns

bool - True if the tracer was successfully activated.

get_span_exporter()

Return the span exporter for this tracer.

Returns a no-op exporter when monitoring is disabled.

def get_span_exporter() -> JudgmentSpanExporter:

Returns

JudgmentSpanExporter - The JudgmentSpanExporter (or no-op variant) for this tracer.

get_span_processor()

Return the span processor for this tracer.

Returns a no-op processor when monitoring is disabled.

def get_span_processor() -> JudgmentSpanProcessor:

Returns

JudgmentSpanProcessor - The JudgmentSpanProcessor (or no-op variant) for this tracer.

get_current_span()

Return the currently active span from the Judgment tracer provider.

def get_current_span() -> opentelemetry.trace.Span:

Returns

opentelemetry.trace.Span - The active Span object.

force_flush()

Send all pending spans to Judgment immediately.

Call this before your process exits (e.g. in a serverless function) to ensure no spans are lost. Does not shut down the tracer.

def lambda_handler(event, context):
    result = process(event)
    Tracer.force_flush()
    return result

def force_flush(timeout_millis=30000) -> bool:

Parameters

Returns

bool - True if all spans were flushed within the timeout.

shutdown()

Flush pending spans and shut down the tracer.

Call this on application exit to ensure all data is exported before the process terminates.

def shutdown(timeout_millis=30000) -> None:

Parameters

Returns

None

registerOTELInstrumentation()

Register a third-party OpenTelemetry instrumentor with Judgment.

Use this to route spans from libraries like opentelemetry-instrumentation-requests through the Judgment trace pipeline.

def registerOTELInstrumentation(instrumentor) -> None:

Parameters

Returns

None

getOTELTracer()

Get the underlying OpenTelemetry Tracer instance.

def getOTELTracer() -> opentelemetry.trace.Tracer:

Returns

opentelemetry.trace.Tracer - The OpenTelemetry Tracer.

start_span()

Start a new span that must be ended manually with span.end().

Prefer the span context manager for automatic lifecycle management.

def start_span(name, attributes=None) -> opentelemetry.trace.Span:

Parameters

Returns

opentelemetry.trace.Span - The newly started Span.

start_as_current_span()

Start a span and set it as the current span in the context.

def start_as_current_span(name, attributes=None) -> Iterator[Span]:

Parameters

Returns

Iterator[Span]

continue_trace()

Continue a distributed trace from an upstream service.

Extracts W3C trace context and Judgment baggage from carrier and makes it the active context for the duration of the block. Any span started inside — including @Tracer.observe functions — becomes a child of the upstream parent, stitching your service into the caller’s trace.

Use this at the entry point of an inbound request (HTTP handler, message queue consumer, RPC dispatcher, etc.) to join the trace started by the upstream caller.

FastAPI:

@Tracer.observe(span_type="agent")
def handle(payload): ...

@app.post("/run")
async def run(request: Request):
    with Tracer.continue_trace(request.headers):
        return handle(await request.json())

Propagating in the opposite direction (outbound):

from judgeval.trace.propagation import inject

headers = {}
inject(headers)
httpx.post(downstream_url, headers=headers, json=payload)

def continue_trace(carrier) -> Iterator[Any]:

Parameters

Returns

Iterator[Any]

scoped_context()

Temporarily apply Judgment context to spans created in the block.

This is useful before integration-created spans exist, such as around a wrapped LLM client call. Context is restored when the block exits.

def scoped_context(*, session_id=None, customer_id=None, customer_user_id=None, attributes=None) -> Iterator[None]:

Parameters

Returns

Iterator[None]

start_linked_trace()

Start a linked trace rooted at a new span.

The new span is the root of a fresh trace. It links back to the current span via an OpenTelemetry Link and stores explicit cross-trace source/target IDs on the linked root and invocation spans.

def start_linked_trace(name, attributes=None, *, span_type='span') -> Iterator[Span]:

Parameters

Returns

Iterator[Span]

span()

Open a child span using a with block.

Use this for tracing a section of code that isn’t a standalone function. Exceptions are automatically recorded on the span.

with Tracer.span("process-results"):
    results = parse(raw_data)
    Tracer.set_attribute("result_count", len(results))

def span(span_name) -> Iterator[Span]:

Parameters

Returns

Iterator[Span]

observe()

Decorator that automatically traces a function call.

Wraps any sync or async function in a span. When fork=True and an active parent span exists, eligible calls run in a fresh linked trace while a parent-side invocation span remains on the current trace. Generator and async-generator functions stay on the normal observation path. Inputs and outputs are captured automatically. Works with or without parentheses.

Basic usage:

@Tracer.observe(span_type="tool")
def search(query: str) -> list[str]:
    return vector_db.search(query)

Async functions work the same way:

@Tracer.observe(span_type="agent")
async def answer(question: str) -> str:
    context = search(question)
    return await llm.generate(question, context)

Fork a call into a linked trace:

@Tracer.observe(span_type="agent", fork=True)
def delegate(task: str) -> str:
    return run_subsystem(task)

Without parentheses (uses default settings):

@Tracer.observe
def my_function():
    ...

def observe(func=None, span_type='span', span_name=None, record_input=True, record_output=True, disable_generator_yield_span=False, fork=False) -> C | Callable[[C], C]:

Parameters

Returns

C | Callable[[C], C]

wrap()

Wrap an LLM client for automatic tracing of all API calls.

Supported providers: OpenAI, Anthropic, Together AI, and Google GenAI. Once wrapped, every API call made through the client is recorded as a span with model name, token counts, and cost.

from openai import OpenAI
from anthropic import Anthropic

openai = Tracer.wrap(OpenAI())
anthropic = Tracer.wrap(Anthropic())

def wrap(client) -> TClient:

Parameters

Returns

TClient - The same client instance, now instrumented with tracing.

set_span_kind()

Set the judgment.span_kind attribute on the current span.

def set_span_kind(kind) -> None:

Parameters

Returns

None

set_llm_span()

def set_llm_span() -> None:

Returns

None

set_tool_span()

def set_tool_span() -> None:

Returns

None

set_general_span()

def set_general_span() -> None:

Returns

None

set_attribute()

Attach a custom key-value pair to the current span.

Use this to record application-specific metadata that you want to see in the Judgment dashboard. Non-primitive values (dicts, lists, objects) are serialized to strings automatically.

Tracer.set_attribute("user_tier", "premium")
Tracer.set_attribute("search_results_count", len(results))

def set_attribute(key, value) -> None:

Parameters

Returns

None

set_attributes()

Set multiple custom attributes on the current span at once.

def set_attributes(attributes) -> None:

Parameters

Returns

None

set_input()

Manually set the input for the current span.

Use when @observe(record_input=False) is set but you want to record a sanitized or transformed version of the input.

def set_input(input_data) -> None:

Parameters

Returns

None

set_output()

Manually set the output for the current span.

Use when @observe(record_output=False) is set but you want to record a sanitized or transformed version of the output.

def set_output(output_data) -> None:

Parameters

Returns

None

recordLLMMetadata()

Record model, token usage, and cost on the current span.

If you’re using Tracer.wrap() this is called automatically. Use this method when you need to record metadata for a custom LLM integration.

@Tracer.observe(span_type="llm")
def call_custom_model(prompt: str) -> str:
    response = my_model.generate(prompt)
    Tracer.recordLLMMetadata({
        "model": "my-model-v2",
        "output_tokens": response.usage.output,
        "total_cost_usd": response.usage.cost,
    })
    return response.text

def recordLLMMetadata(metadata) -> None:

Parameters

Returns

None

set_customer_id()

Associate the current trace with a customer.

Once set, this ID propagates to all child spans and enables per-customer analytics in the Judgment dashboard. Call this early in your request handler.

@Tracer.observe(span_type="agent")
def handle_request(user_id: str, question: str):
    Tracer.set_customer_id(user_id)
    return answer(question)

def set_customer_id(customer_id) -> None:

Parameters

Returns

None

set_customer_user_id()

Set the customer user ID on the current span and propagate to children.

def set_customer_user_id(customer_user_id) -> None:

Parameters

Returns

None

set_session_id()

Associate the current trace with a conversation session.

Groups multiple requests into a session in the Judgment dashboard. Propagates to all child spans. Call this early in your request handler.

@Tracer.observe(span_type="agent")
def handle_message(session_id: str, message: str):
    Tracer.set_session_id(session_id)
    return chatbot.respond(message)

def set_session_id(session_id) -> None:

Parameters

Returns

None

set_propagating_attribute()

Set an attribute and propagate it to all child spans via baggage.

Unlike set_attribute (single span only). The judgment. prefix is reserved and ignored.

def set_propagating_attribute(key, value) -> None:

Parameters

Returns

None

tag()

Add tags to the current trace for filtering in the dashboard.

Tags are sent asynchronously and appear in the Judgment monitoring view. Useful for marking traces by feature, experiment, or user segment.

Tracer.tag("rag-pipeline")
Tracer.tag(["experiment-v2", "premium-user"])

def tag(tags) -> None:

Parameters

Returns

None

async_evaluate()

Run a hosted evaluation on this span when it completes.

The evaluation is queued and processed server-side by the Judgment platform after the span ends. Use this to score live traffic without blocking your application.

@Tracer.observe(span_type="agent")
def answer(question: str) -> str:
    response = llm.generate(question)
    Tracer.async_evaluate(
        "faithfulness",
        {"input": question, "actual_output": response},
    )
    return response

def async_evaluate(judge, example=None) -> None:

Parameters

Returns

None