Skip to Content
ObservabilityTelemetry

Telemetry

Telemetry tells you what an agent did and how long it took: every run, turn, model call and tool call opens a timed span, and token counts, cost and durations are emitted as metrics. Pass a telemetry sink to Agent(telemetry=...) to see them. Loomflow ships six: five that need no collector (including the no-op default and a fan-out helper), and OTelTelemetry, which sends everything to any OpenTelemetry-compatible backend when you’re ready for production.

from loomflow import Agent
from loomflow.observability import ConsoleTelemetry
 
agent = Agent(
    "...",
    model="claude-opus-4-7",
    telemetry=ConsoleTelemetry(),
)

Run the agent and span lines print to stderr as they happen. No collector, no SDK, no extras to install.

To browse runs in a web page, with the messages each model call sent and each tool call’s arguments and result, turn on Studio: it records alongside whatever telemetry you configure here.

The five sinks that need no collector

SinkWhat it doesBest for
NoTelemetryNo-op default. Yields empty spans and drops metrics at zero overhead.Production paths with telemetry off
InMemoryTelemetryAccumulates spans + metrics in lists. Inspect via .spans() / .metrics().Unit tests, exploration
ConsoleTelemetryPrints each span + metric to a stream (sys.stderr by default) with nested-trace indentation.”Tail my agent in dev”
FileTelemetryAppend-only JSONL on disk. One JSON record per span / metric. Parseable by jq, Splunk, Datadog log pipelines.Long-running dev / staging, post-mortems
MultiTelemetryFan-out. Forwards every span + metric to every sink in declaration order. Trace IDs are shared.Watch live AND inspect after
from loomflow.observability import (
    NoTelemetry,
    InMemoryTelemetry,
    ConsoleTelemetry,
    FileTelemetry,
    MultiTelemetry,
)

All five implement the same Telemetry protocol as OTelTelemetry, so swapping sinks never touches agent code.

Spans

SpanOpensAttributes
loom.runeach agent.run() / agent.stream()session_id, max_turns, model, architecture, plus gen_ai.* (invoke_agent)
loom.turneach iteration of the loopturn, session_id
loom.model.completeeach model call from agent.run()model, turn, session_id, tool_count, plus gen_ai.*
loom.model.streameach model call from agent.stream()same as loom.model.complete
loom.tooleach tool calltool, call_id, turn, loom.permission (allow, deny, ask), loom.approval (approved, declined) after an ask, plus gen_ai.*
loom.workflow.stepeach step of a Workflowstep, user_id, session_id, pattern

These are the names the in-memory, console and file sinks record. OTelTelemetry exports the GenAI spans under their OpenTelemetry spec names instead: model calls as chat {model} (kind CLIENT), tool calls as execute_tool {tool}, and the run as invoke_agent. The loomflow name moves to the loom.span.name attribute, so a query on loom.span.name = "loom.model.complete" still finds them. loom.turn and loom.workflow.step keep their names. See GenAI semantic conventions.

For multi-agent architectures, the worker’s spans nest under the supervisor’s tool call:

loom.run
├── loom.turn (turn=1)
│   ├── loom.model.complete
│   └── loom.tool (tool=delegate)
│       └── loom.run (the worker)
│           └── loom.turn ... (recursive)
└── loom.turn (turn=2)
    └── loom.model.complete

Every sink preserves the parent_span_id link, including FileTelemetry’s JSONL output, so you can rebuild this tree offline.

Metrics

MetricTypeUnitTags
loom.tokens.inputcounter{token}model
loom.tokens.outputcounter{token}model
loom.cost.usdcountermodel (only emitted when the call has a cost)
loom.tool.duration_mshistogrammstool, ok, denied
loom.session.duration_mshistogrammsinterrupted
loom.budget.exceededcounterreason
loom.auto_extract.duration_mshistogrammsstatus
loom.auto_extract.invocationscounterstatus

Each model and tool call also emits the OpenTelemetry GenAI client metrics, with the spec’s units and bucket boundaries:

MetricTypeUnitTags
gen_ai.client.inference.durationhistogramsgen_ai.operation.name, gen_ai.provider.name, gen_ai.request.model
gen_ai.client.inference.usage.input_tokenscounter{token}the same, plus gen_ai.token.modality
gen_ai.client.inference.usage.output_tokenscounter{token}the same, plus gen_ai.token.modality
gen_ai.client.inference.usage.cache_read.input_tokenscounter{token}as above (only when the call read the cache)
gen_ai.client.inference.usage.cache_write.input_tokenscounter{token}as above (only when the call wrote the cache)
gen_ai.client.inference.operation.input_tokenshistogram{token}gen_ai.operation.name, gen_ai.provider.name, gen_ai.request.model
gen_ai.client.inference.operation.output_tokenshistogram{token}the same
gen_ai.execute_tool.durationhistogramsgen_ai.operation.name, gen_ai.tool.name, error.type on failure

The gen_ai.* input-token counts include cached tokens; the loom.tokens.input count doesn’t.

Metrics carry no session_id, turns or user_id tags: one series per session or per user is unbounded cardinality. session_id stays on the agent spans; user_id is on the workflow step spans and in every audit entry.

The gen_ai.* instruments have fixed types. For the rest, the type follows the metric name: names ending in _ms, _seconds or _bytes become histograms (with unit ms, s or By), loom.tokens.* are token counters, and everything else is a counter. There is one emit_metric() call, and the in-memory, console and file sinks tag each captured metric with the same instrument_kind, so test assertions match what OTLP would have emitted.

NoTelemetry (the default)

from loomflow import Agent, NoTelemetry
 
agent = Agent("...", model="claude-opus-4-7", telemetry=NoTelemetry())   # explicit
agent = Agent("...", model="claude-opus-4-7")                            # implicit (same)

NoTelemetry makes every span / metric call a no-op. The agent loop detects this at construction and skips the async with telemetry.trace(...) context managers entirely on the hot path.

Worked example: inspect spans and metrics in-process

The simplest way to see what an agent emits is InMemoryTelemetry. There’s no collector, SDK or log file; you get the captured records as plain dataclasses:

import asyncio
from loomflow import Agent, ScriptedModel, ScriptedTurn, ToolCall, tool
from loomflow.observability import InMemoryTelemetry
 
@tool
async def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b
 
async def main():
    telemetry = InMemoryTelemetry()
 
    # ScriptedModel keeps the run deterministic, with no network call
    model = ScriptedModel([
        ScriptedTurn(tool_calls=[
            ToolCall(id="c1", tool="add", args={"a": 2, "b": 3})
        ]),
        ScriptedTurn(text="The sum is 5."),
    ])
 
    agent = Agent(
        "Arithmetic assistant.",
        model=model,
        tools=[add],
        telemetry=telemetry,
    )
    await agent.run("What is 2 + 3?", user_id="alice")
 
    # Spans carry timing + parent-span linkage
    for s in telemetry.spans():
        attrs = ", ".join(f"{k}={v}" for k, v in s.attributes.items())
        print(f"{s.name:<22}  ({s.duration_ms:5.1f}ms)  [{attrs}]")
 
    # Metrics carry their auto-detected instrument kind
    for m in telemetry.metrics():
        print(f"{m.name:<28}  {m.value:<8}  ({m.instrument_kind})")
 
asyncio.run(main())

Output (trimmed; spans are recorded as they close, so children come before their parents):

loom.model.complete     (  0.0ms)  [model=scripted, turn=1, session_id=sess_..., tool_count=1, gen_ai.operation.name=chat, ...]
loom.tool               (  0.1ms)  [tool=add, call_id=c1, turn=1, gen_ai.operation.name=execute_tool, ...]
loom.turn               (  0.3ms)  [turn=1, session_id=sess_...]
loom.model.complete     (  0.0ms)  [model=scripted, turn=2, session_id=sess_..., tool_count=1, ...]
loom.turn               (  0.0ms)  [turn=2, session_id=sess_...]
loom.run                (  0.8ms)  [session_id=sess_..., max_turns=50, model=scripted, architecture=react, gen_ai.operation.name=invoke_agent, ...]
loom.tokens.input             0         (counter)
loom.tokens.output            0         (counter)
gen_ai.client.inference.duration  4.1e-05  (histogram)
gen_ai.client.inference.usage.input_tokens  0  (counter)
...
loom.tool.duration_ms         0.05      (histogram)
gen_ai.execute_tool.duration  5.3e-05   (histogram)
loom.tokens.input             0         (counter)
...
loom.session.duration_ms      0.68      (histogram)

Token counts are 0 because these ScriptedTurns don’t set usage=; a real model reports its own.

CapturedSpan carries name, trace_id, span_id, parent_span_id, started_at, ended_at, duration_ms, attributes, and an exception repr if the body raised. CapturedMetric carries name, value, instrument_kind, attributes, and emitted_at. Both are frozen dataclasses. Assert on them directly in tests.

Tail in stderr with ConsoleTelemetry

Swap the sink to see spans appear live as the agent runs, with nested-trace indentation:

import sys
from loomflow.observability import ConsoleTelemetry
 
agent = Agent(
    "...",
    model="claude-opus-4-7",
    telemetry=ConsoleTelemetry(stream=sys.stderr, show_metrics=False),
)

Each span prints when it closes, indented under its parent (attributes trimmed here):

    • loom.model.complete  (0.0ms)  model=scripted, turn=1, session_id=sess_..., tool_count=1, ...
    • loom.tool  (0.1ms)  tool=add, call_id=c1, turn=1, ...
  • loom.turn  (0.5ms)  turn=1, session_id=sess_...
    • loom.model.complete  (0.0ms)  model=scripted, turn=2, ...
  • loom.turn  (0.1ms)  turn=2, session_id=sess_...
• loom.run  (1.0ms)  session_id=sess_..., max_turns=50, model=scripted, architecture=react

Indentation follows the active parent span automatically, even when spans open inside anyio.create_task_group() (the framework’s parallel tool-dispatch path).

Write JSONL with FileTelemetry

from loomflow.observability import FileTelemetry
 
agent = Agent(
    "...",
    model="claude-opus-4-7",
    telemetry=FileTelemetry("./traces.jsonl"),
)

Each line is a structured JSON record with "kind": "span" or "kind": "metric" discriminator. Span records carry the parent_span_id linkage needed to reconstruct the trace tree offline. Writes go through anyio.to_thread.run_sync so the event loop never blocks on disk I/O; an internal lock serialises concurrent writes from parallel tool dispatches.

Query offline with jq:

# Spans that took longer than 1s
jq -c 'select(.kind=="span" and .duration_ms > 1000)' traces.jsonl
 
# One session
jq -c 'select(.attributes.session_id=="sess_...")' traces.jsonl
 
# All cost metrics
jq -c 'select(.kind=="metric" and .name=="loom.cost.usd")' traces.jsonl

There’s no built-in rotation. Use logrotate, journald or your platform’s log management to cap file size. The framework deliberately stays out of that policy decision.

FileTelemetry ≠ FileAuditLog. They capture different things: audit log = business events for compliance (“did Alice’s refund go through?”); telemetry = performance / diagnostic spans (“why was this run slow?”). Run both together in production.

Fan out with MultiTelemetry

Watch live AND assert in tests by stacking sinks:

from loomflow import Agent, ScriptedModel, ScriptedTurn, ToolCall, tool
from loomflow.observability import (
    ConsoleTelemetry, InMemoryTelemetry, MultiTelemetry,
)
 
 
@tool
async def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b
 
 
model = ScriptedModel([
    ScriptedTurn(tool_calls=[ToolCall(id="c1", tool="add", args={"a": 2, "b": 3})]),
    ScriptedTurn(text="The sum is 5."),
])
 
in_mem = InMemoryTelemetry()
telemetry = MultiTelemetry([ConsoleTelemetry(), in_mem])
 
agent = Agent("Arithmetic assistant.", model=model, tools=[add], telemetry=telemetry)
await agent.run("What is 2 + 3?")
 
# Watched live in stderr, and still assertable
assert any(s.name == "loom.tool" for s in in_mem.spans())

Span IDs are minted by the first sink and shared with the others, so trace hierarchy stays consistent across every capture. Exceptions inside one sink’s trace propagate after every other sink has had a chance to record its cleanup (finally blocks fire even on exceptional exit thanks to AsyncExitStack).

The framework ships examples/13_telemetry.py running InMemoryTelemetry, ConsoleTelemetry, MultiTelemetry and FileTelemetry end-to-end against a ScriptedModel. No API key required.

Production: OTelTelemetry

For Honeycomb / Datadog / Grafana Tempo / Jaeger / any OTLP-compatible backend, wire the OpenTelemetry SDK and pass it to OTelTelemetry:

from loomflow.observability import OTelTelemetry
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
 
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="https://otlp.your-vendor"))
)
 
metric_reader = PeriodicExportingMetricReader(
    OTLPMetricExporter(endpoint="https://otlp.your-vendor")
)
meter_provider = MeterProvider(metric_readers=[metric_reader])
 
telemetry = OTelTelemetry(
    tracer_provider=tracer_provider,
    meter_provider=meter_provider,
)
 
agent = Agent("...", model="claude-opus-4-7", telemetry=telemetry)

Install the OTel extras:

pip install 'loomflow[otel]'

The adapter is the same Telemetry protocol. Swapping sinks never touches agent code. The one difference you’ll see in the backend is the span names: model, tool and run spans arrive as chat {model}, execute_tool {tool} and invoke_agent, with the loom.* name in the loom.span.name attribute. A failed or refused tool call sets error.type and an ERROR status on its span.

Model and tool spans also carry the OTel GenAI semantic-convention attributes (gen_ai.operation.name, gen_ai.usage.input_tokens, …), so attribute-keyed backends like Langfuse and Datadog LLM Observability recognise them without mapping config. See GenAI semantic conventions.

Custom telemetry

Any class with two methods satisfies the Telemetry protocol: a trace() async context manager that yields a span handle, and an async emit_metric():

import time
from contextlib import asynccontextmanager
 
from loomflow import Agent
from loomflow.core.types import Span
 
 
class PrintTelemetry:
    @asynccontextmanager
    async def trace(self, name: str, **attrs):
        start = time.perf_counter()
        # Emit sites add attributes (token usage, for example) to the
        # yielded span's ``attributes`` dict before it closes.
        span = Span(name=name, trace_id="", span_id="", attributes=dict(attrs))
        try:
            yield span
        finally:
            print(f"{name} took {(time.perf_counter() - start) * 1000:.1f}ms")
 
    async def emit_metric(self, name: str, value: float, **attrs):
        print(f"metric {name}={value}")
 
 
agent = Agent("...", model="echo", telemetry=PrintTelemetry())
await agent.run("hi")

Useful when you have a custom metric backend not covered by OTel (e.g. a homegrown StatsD or Prometheus pushgateway). Wrap it in MultiTelemetry alongside one of the built-in sinks to keep the live-tail / file-on-disk affordances while still sending to your backend.

Telemetry vs the audit log. Spans are for operational observability: latency, error rates, dependencies. The audit log is for compliance observability: who did what, with HMAC signatures. Different lifecycles, different retention, different consumers. Wire both; they don’t overlap.

Last updated on
SandboxesGenAI semantic conventions