Observability

The runtime emits OpenTelemetry spans for every turn, LLM call, tool call and durable step, following the GenAI semantic conventions. Langfuse, LangSmith, Phoenix, Datadog and the like render tokens, models and tool calls from them. Without the otel extra, the hooks are inactive.

Setup

pip install "pyrula[agents,otel]"

Call once at startup, before the worker starts (later calls no-op):

from pyrula.agents.observability import configure

configure()

With no arguments it honors the standard OTLP variables. Or pass endpoint=, headers=, service_name= (default "pyrula"):

export OTEL_EXPORTER_OTLP_ENDPOINT=https://collector.example.com
export OTEL_EXPORTER_OTLP_HEADERS="authorization=Bearer <token>"

Shortcuts fill in endpoint and auth for common backends:

from pyrula.agents.observability import configure_langfuse

# reads LANGFUSE_PUBLIC_KEY / LANGFUSE_SECRET_KEY / LANGFUSE_HOST (default cloud.langfuse.com)
configure_langfuse()

from pyrula.agents.observability import configure_langsmith

# reads LANGSMITH_API_KEY (or LANGCHAIN_API_KEY) and LANGSMITH_PROJECT
configure_langsmith()

from pyrula.agents.observability import configure_datadog

# exports to the local Datadog Agent's OTLP intake (enable otlp_config in the Agent)
configure_datadog()

Keys can be passed explicitly (configure_langfuse(public_key=..., secret_key=...)). To print spans to stdout:

configure(console=True)

If your application already registers an OTel tracer provider, don’t call configure(); spans go through the global provider.

Traced operations

invoke_agent billing_agent
├── chat claude-sonnet-4-6          gen_ai.usage.input_tokens=812 output_tokens=94
│   └── execute_tool lookup_invoice  gen_ai.tool.name=lookup_invoice
└── chat claude-sonnet-4-6          gen_ai.usage.input_tokens=931 output_tokens=187

Spans carry gen_ai.operation.name, gen_ai.request.model, gen_ai.provider.name and token usage. Durable steps and stream batches get pyrula.* spans in the same trace.

Prompt and completion content

Prompts, completions and tool arguments and results are recorded on spans only when content capture is on. configure() leaves it off unless OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT is set.

configure(capture_content=True)

The Langfuse, LangSmith and Datadog shortcuts turn it on by default; pass capture_content=False to turn it off:

configure_datadog(capture_content=False)

Captured content and exception messages in span status are secret-redacted first (API keys, tokens, passwords, credentials in URLs), then truncated at 32 KB.

Metrics

configure() also exports gen_ai.client.token.usage (histogram by model, provider and token type) and gen_ai.client.operation.duration (seconds, tagged error.type on failure). Langfuse and LangSmith only ingest traces, so their shortcuts disable metrics; pass metrics=False for any other trace-only backend.

Tracing across Kafka

KafkaAgentRunner propagates W3C trace context through Kafka headers, so a pipeline of agents is one trace. A consumed record starts a process {topic} span parented to the upstream traceparent; records the turn produces carry that span’s context; DLQ records keep the source traceparent. Without OTel the runner forwards the inbound traceparent verbatim.

For manual produce paths:

from pyrula.agents.observability.otel import inject_current_context

headers = {k: v.encode() for k, v in inject_current_context().items()}

Conversations and usage in SQL

Pyrula can project the event log into two read-only tables. The event log stays the system of record; the tables can be rebuilt at any time, and agent code never reads them.

• pyrula_agent_messages: one row per conversation turn (role, JSONB content).
• pyrula_agent_usage: one row per LLM call: input_tokens, output_tokens, total_tokens (when the event carries it), cache_creation_input_tokens, cache_read_input_tokens.

A sub-agent shares its parent’s run; its rows carry its call id in subagent_call_id, and the parent’s have subagent_call_id IS NULL.

Enable projection

project=True on the agent Postgres store runs the projection migrations once and registers the projectors; the worker drains new events every 60 seconds.

from pyrula.agents import PyrulaWorker
from pyrula.agents.stores.postgres import PostgresStore

store = PostgresStore(
    dsn="postgresql://localhost/pyrula",
    valkey_url="redis://localhost:6379",
    project=True,
)
worker = PyrulaWorker(agents=[support], store=store, llm=...)
worker.run()

The projection has its own migration chain (pyrula_agent_alembic_version), separate from the workflow archive’s and your application’s alembic_version. On first use it checks pyrula_agent_schema_compat and raises ArchiveSchemaMismatch if the database is missing, from another baseline, or needs a newer pyrula.

Queries

Tokens per model, last 7 days:

SELECT model,
       SUM(input_tokens)  AS in_tok,
       SUM(output_tokens) AS out_tok,
       SUM(cache_read_input_tokens) AS cached_in_tok
FROM pyrula_agent_usage
WHERE created_at > NOW() - INTERVAL '7 days'
GROUP BY model
ORDER BY in_tok + out_tok DESC;

One run’s transcript:

SELECT role, content
FROM pyrula_agent_messages
WHERE name = 'support' AND run_id = '<run-id>'
ORDER BY msg_seq;

Cost

Pyrula records tokens, not money, in both the tables and the spans. Multiply by current rates, let your tracing backend price them, or use Pyrula Cloud’s project-level price table.

Backfill

Runs archived before projection was on are filled in with pyrula reproject:

pyrula reproject                                 # every archived run, every agent
pyrula reproject --agent support                 # one agent
pyrula reproject --agent support --run-id <id>   # one run (--run-id needs --agent)