Tracing Inside Actions

Tooling for gaining visibility inside actions. You will never instantiate these directly, they are all injected by the framework. This is purely for reference.

class burr.visibility.tracing.TracerFactory(
action: str,
sequence_id: int,
app_id: str,
partition_key: str | None,
lifecycle_adapters: ~burr.lifecycle.internal.LifecycleAdapterSet,
_context_var: ~_contextvars.ContextVar[~burr.visibility.tracing.ActionSpan | None] = <ContextVar name='execution_context' default=None>,
)

Represents a tracer factory to create tracer instances. User never instantiates this directly. Rather, this gets injected by the application. This gives a new span.

Note this carries state – the top level span count. This is important for the sequence id at the root level.

You will only ever see a tracer factory in the context of an action, passed through the __tracer parameter.

@action(reads=[...], writes=[...])
def my_action(state: State, __tracer: TracerFactory) -> tuple[dict, State]:
    context_manager: ActionSpanTracer = __tracer("my_span_name")
    with context_manager:
        ...
class burr.visibility.tracing.ActionSpanTracer(action: str, action_sequence_id: int, span_name: str, lifecycle_adapters: ~burr.lifecycle.internal.LifecycleAdapterSet, app_id: str, partition_key: str | None, span_dependencies: ~typing.List[str], top_level_span_count: int = 0, context_var=<ContextVar name='execution_context' default=None>)

Context manager for use within tracing actions. This has the role solely of delegating to hooks – it does not do anything except manage context and pass those to the hooks.

Note that a new instance of this will be passed to every action that is traced. This allows us to reset this based on state. This can handle both synchronous and asynchronous contexts.

You will be using this through the action API. When you declare __tracer as a parameter, it gives you a callable that, when called with a span name, returns an ActionSpanTracer

@action(reads=[...], writes=[...])
def my_action(state: State, __tracer: TracerFactory) -> tuple[dict, State]:
    context_manager: ActionSpanTracer = __tracer("my_span_name")
    with context_manager:
        ...

The following hooks are respected:

  • pre_span_start and async pre_span_start

  • post_span_end and async post_span_end

log_attribute(key: str, value: Any)

Logs a single attribute to the UI. Note that this must be paired with a tracker or a tracking hook to be useful, otherwise it will be a no-op.

Parameters:

  • key – Name of the attribute (must be unique per action/span)

  • value – Value of the attribute.

log_attributes(**attributes)

Logs a set of attributes to the UI. Note that this must be paired with a tracker or a tracking hook to be useful, otherwise it will be a no-op.

Parameters:

attributes – Attributes to log

class burr.visibility.tracing.ActionSpan(
action: str,
action_sequence_id: int,
name: str,
parent: ForwardRef('ActionSpan') | None,
sequence_id: int = 0,
child_count: int = 0,
)
classmethod create_initial(
action: str,
name: str,
sequence_id: int,
action_sequence_id: int,
) → ActionSpan

Creates the initial action span for an action. This should be only called if the current action span is none.

Parameters:

  • action

  • name

  • sequence_id

Returns:

class burr.visibility.tracing.trace(
capture_inputs: bool = True,
capture_outputs: bool = True,
input_filterlist: ~typing.List[str] | None = None,
span_name: str | None = None,
_context_var: ~_contextvars.ContextVar[~burr.visibility.tracing.TracerFactory | None] = <ContextVar name='tracer_context' default=None>,
)
__init__(
capture_inputs: bool = True,
capture_outputs: bool = True,
input_filterlist: ~typing.List[str] | None = None,
span_name: str | None = None,
_context_var: ~_contextvars.ContextVar[~burr.visibility.tracing.TracerFactory | None] = <ContextVar name='tracer_context' default=None>,
)

trace() can wrap any function and uses the tracer to create a span and log attributes. This also (by default) logs the inputs/outputs of a function as attributes. Be careful not to include sensitive data in the inputs/outputs, but if you do, you have the input_filterlist to exclude it.

This works with sync/async

Take the following code:

from burr.visibility import trace

@trace()
def call_llm(prompt):
    return _query(...)

@trace()
def generate_text(prompt: str) -> str:
    result = call_llm(prompt)
    return f"<p>{result}</p>"

@action(reads=["prompt"], writes=["response"])
def prompt_action(state: State) -> State:
    response = generate_text(state["prompt"])
    return state.update(response=response)

Every time prompt_action is called (within the context of prompt_action), it will generate a trace that looks like the following:

—— prompt action ——————————————–
—– generate_text ———————————–

—– call_llm ———————————–

If it is called outside the context of a Burr action, it will be effectively a no-op.

Parameters:

  • capture_inputs – Whether to capture inputs as attributes (defaults to True). Note that this only works with keyword-argument bindable functions.

  • capture_outputs – Whether to capture outputs as attributes (defaults to True)

  • input_filterlist – A list of inputs to filter out (defaults to filtering nothing. Use if you have sensitive data)

  • span_name – Name of the span, will default to the function name

  • _context_var – Context var to use for the tracer factory, used purely for internal testing