Pattern Tracing

TracedPattern wraps any BasePattern with OpenTelemetry spans so every run appears in your observability backend (Arize Phoenix, Jaeger, Honeycomb, Grafana Tempo, …).

When no OTEL backend is configured the wrapper is a zero-overhead pass-through — it uses the no-op tracer from brain.obs.tracing.

Usage

from brain.patterns import ReActAgent
from brain.patterns.tracing import TracedPattern

agent = TracedPattern(
    ReActAgent(tools={"search": my_search}),
    name="research_agent",           # span name (default: class name)
    attributes={"env": "prod",       # extra span attributes
                "team": "platform"},
    record_steps=True,               # add span events for each step (default: True)
)

result = agent.run("What is the latest on AI safety?")
# → emits a "research_agent" span with step events
print(result.metadata.get("trace_id"))  # hex trace ID if OTEL is active

Span attributes

Attribute	Value
pattern.name	Class name of the wrapped pattern
pattern.task	Task string (first 500 chars)
pattern.ok	True/False
pattern.iterations	Number of steps
pattern.answer	Final answer (first 500 chars)
pattern.latency_ms	Wall-clock run time in ms
pattern.step_count	len(result.steps)
pattern.error	Error message (if any)

Plus any attributes you pass to the constructor.

Enabling OTEL

from brain.obs.tracing import init_tracing

init_tracing(
    project_name="my_project",
    endpoint="http://localhost:6006/v1/traces",  # e.g. Arize Phoenix
    dev_mode=True,   # SimpleSpanProcessor — immediate export
)

Or via environment variables:

export PHOENIX_COLLECTOR_ENDPOINT=http://localhost:6006/v1/traces
export PHOENIX_PROJECT_NAME=my_project
export SB_TRACING_DEV_MODE=true

trace_pattern decorator/factory

from brain.patterns.tracing import trace_pattern

# Wrap an instance
agent = trace_pattern(name="prod_agent", attributes={"version": "0.3.0"})(my_agent)

# Or use as a class decorator (wraps run() on every instance)
@trace_pattern(name="my_react", attributes={"team": "ml"})
class MyReActAgent(ReActAgent):
    pass

API Reference

brain.patterns.tracing.TracedPattern

TracedPattern(pattern: BasePattern, name: str | None = None, attributes: dict[str, Any] | None = None, record_steps: bool = True)

Bases: BasePattern

Wrap any BasePattern with OpenTelemetry spans.

Transparent proxy — all method calls are forwarded to the inner pattern. The run() method is intercepted to create a root span and record step-level events.

Parameters:

Name	Type	Description	Default
pattern	BasePattern	The pattern to wrap.	required
name	str | None	Span name. Defaults to the wrapped pattern's class name.	None
attributes	dict[str, Any] | None	Extra OTEL attributes added to the root span.	None
record_steps	bool	If True (default), add span events for each Step in the result. Set to False if step volume is very high.	True

Source code in brain/patterns/tracing.py

def __init__(
    self,
    pattern: BasePattern,
    name: str | None = None,
    attributes: dict[str, Any] | None = None,
    record_steps: bool = True,
) -> None:
    self.pattern = pattern
    self._span_name = name or type(pattern).__name__
    self._attributes: dict[str, Any] = attributes or {}
    self._record_steps = record_steps

run

run(task: str, **kwargs: Any) -> PatternResult

Run the wrapped pattern inside an OTEL span.

Parameters:

Name	Type	Description	Default
task	str	The question or instruction for the agent.	required

Returns:

Type	Description
PatternResult	The PatternResult from the wrapped pattern, unchanged.
PatternResult	result.metadata["trace_id"] is set to the span's trace ID
PatternResult	(as a hex string) when OTEL is available.

Source code in brain/patterns/tracing.py

def run(self, task: str, **kwargs: Any) -> PatternResult:
    """Run the wrapped pattern inside an OTEL span.

    Args:
        task: The question or instruction for the agent.

    Returns:
        The ``PatternResult`` from the wrapped pattern, unchanged.
        ``result.metadata["trace_id"]`` is set to the span's trace ID
        (as a hex string) when OTEL is available.
    """
    tracer = get_tracer(_TRACER_NAME)

    t0 = time.perf_counter()

    try:
        with tracer.start_as_current_span(self._span_name) as span:
            # Root span attributes
            _set_attr(span, "pattern.name", type(self.pattern).__name__)
            _set_attr(span, "pattern.task", task[:500])
            for k, v in self._attributes.items():
                _set_attr(span, k, v)

            result = self.pattern.run(task, **kwargs)

            latency_ms = int((time.perf_counter() - t0) * 1000)
            _set_attr(span, "pattern.ok", result.ok)
            _set_attr(span, "pattern.iterations", result.iterations)
            _set_attr(span, "pattern.answer", result.answer[:500])
            _set_attr(span, "pattern.latency_ms", latency_ms)
            _set_attr(span, "pattern.step_count", len(result.steps))
            if result.error:
                _set_attr(span, "pattern.error", result.error)

            if self._record_steps:
                for step in result.steps:
                    span.add_event(
                        f"step.{step.index}",
                        attributes={
                            "thought": step.thought[:200],
                            "action": step.action,
                            "action_input": step.action_input[:200],
                            "observation": step.observation[:200],
                        },
                    )

            # Inject trace context into result metadata
            try:
                ctx = span.get_span_context()
                if ctx and ctx.is_valid:
                    result.metadata["trace_id"] = format(ctx.trace_id, "032x")
                    result.metadata["span_id"] = format(ctx.span_id, "016x")
            except Exception:  # noqa: BLE001
                pass

            return result

    except Exception as exc:  # noqa: BLE001
        # If tracing itself fails, run the pattern without tracing
        logger.debug("TracedPattern: tracing error, falling back: %s", exc)
        return self.pattern.run(task, **kwargs)

brain.patterns.tracing.trace_pattern

trace_pattern(name: str | None = None, attributes: dict[str, Any] | None = None, record_steps: bool = True) -> Any

Decorator / factory that wraps a pattern instance in TracedPattern.

Can be used as a decorator on a class, or called directly:

.. code-block:: python

from brain.patterns.tracing import trace_pattern

@trace_pattern(name="my_agent", attributes={"team": "platform"})
class MyAgent(ReActAgent):
    pass

# or on an instance:
agent = trace_pattern(name="prod_agent")(ReActAgent(tools=...))

Source code in brain/patterns/tracing.py

def trace_pattern(
    name: str | None = None,
    attributes: dict[str, Any] | None = None,
    record_steps: bool = True,
) -> Any:
    """Decorator / factory that wraps a pattern instance in TracedPattern.

    Can be used as a decorator on a class, or called directly:

    .. code-block:: python

        from brain.patterns.tracing import trace_pattern

        @trace_pattern(name="my_agent", attributes={"team": "platform"})
        class MyAgent(ReActAgent):
            pass

        # or on an instance:
        agent = trace_pattern(name="prod_agent")(ReActAgent(tools=...))

    """

    def _wrap(pattern_or_class: Any) -> Any:
        if isinstance(pattern_or_class, type):
            # Class decorator: patch __init__ to auto-wrap each instance
            original_init = pattern_or_class.__init__

            def new_init(self: Any, *args: Any, **kw: Any) -> None:
                original_init(self, *args, **kw)
                original_run = self.run

                def _traced(task: str, **run_kw: Any) -> PatternResult:
                    class _P(BasePattern):
                        def run(self_inner: "_P", t: str, **k: Any) -> PatternResult:
                            return original_run(t, **k)

                    return TracedPattern(
                        _P(),
                        name=name or type(self).__name__,
                        attributes=attributes,
                        record_steps=record_steps,
                    ).run(task, **run_kw)

                self.run = _traced

            pattern_or_class.__init__ = new_init
            return pattern_or_class

        # Instance: wrap directly
        return TracedPattern(
            pattern_or_class,
            name=name,
            attributes=attributes,
            record_steps=record_steps,
        )

    return _wrap