Telemetry

Optional OpenTelemetry tracing for the emotional_memory pipeline.

Install the [otel] extra to activate spans:

uv pip install "emotional-memory[otel]"

Without the extra, all helpers are zero-overhead no-ops — no import errors, no performance cost.

Package logging (configure_logging) is separate and always available — see Logging.

Usage

from emotional_memory.telemetry import traced_span

with traced_span("my.operation", attributes={"key": "value"}) as span:
    ...  # span is None when [otel] extra is absent

Use traced_span in your own code to add child spans inside the engine's root spans. If the body raises, the exception is recorded on the span and status is set to ERROR before re-raising.

API

emotional_memory.telemetry.traced_span(name: str, attributes: Mapping[str, Any] | None = None) -> Iterator[Any | None]

Emit an OTel span if the [otel] extra is installed; otherwise no-op.

Source code in src/emotional_memory/telemetry.py

@contextlib.contextmanager
def traced_span(
    name: str,
    attributes: Mapping[str, Any] | None = None,
) -> Iterator[Any | None]:
    """Emit an OTel span if the ``[otel]`` extra is installed; otherwise no-op."""
    tracer = _get_tracer()
    if tracer is None:
        yield None
        return

    attrs = dict(attributes) if attributes else None
    with tracer.start_as_current_span(name, attributes=attrs) as span:
        try:
            yield span
        except Exception as exc:
            span.record_exception(exc)
            trace_module = importlib.import_module("opentelemetry.trace")
            span.set_status(trace_module.Status(trace_module.StatusCode.ERROR, str(exc)))
            raise