pyagent-trace API Reference

Event Bus

pyagent_trace.events.TraceEvent dataclass

Contract: every trace event pyagent emits.

Attributes:

Name	Type	Description
timestamp	float	Unix timestamp of the event.
event_type	str	One of pattern_start, pattern_end, agent_start, agent_end, llm_call, llm_response, routing_decision, compression, error, cost_record.
agent_name	str	Name of the agent involved (empty string if N/A).
pattern_type	str	Name of the pattern involved (empty string if N/A).
payload	dict[str, Any]	Event-specific data (tokens, cost, duration, model, messages, etc.).

pyagent_trace.events.TraceEventBus

Pub/sub event bus — the contract between trace producers and studio consumers.

Usage

bus = TraceEventBus() sub_id = bus.subscribe(lambda event: print(event)) bus.emit(TraceEvent(timestamp=time.time(), event_type="llm_call", ...)) bus.unsubscribe(sub_id)

emit(event)

Emit event to all matching subscribers (sync).

emit_async(event) async

Emit event to all matching subscribers, awaiting any coroutine callbacks.

subscribe(callback)

Subscribe to all trace events. Returns subscription ID.

subscribe_filter(event_types, callback)

Subscribe to specific event types only. Returns subscription ID.

unsubscribe(subscription_id)

Remove a subscription by ID.

Exporters

pyagent_trace.exporters.base.TraceExporter

Bases: Protocol

Portal-agnostic exporter contract.

Any class implementing these three methods can receive pyagent trace events. Built-in: ConsoleExporter, JsonlExporter, OTelExporter, LangfuseExporter. Users can implement custom exporters for any backend.

export_event(event)

Export a single trace event to the backend.

flush()

Flush any buffered events to the backend.

shutdown()

Shutdown the exporter, flushing remaining events.

pyagent_trace.exporters.console.ConsoleExporter

Export trace events to stdout (or any file-like object).

Usage

exporter = ConsoleExporter() bus.subscribe(exporter.export_event)

export_event(event)

Print a formatted trace event.

flush()

Flush the output stream.

shutdown()

Flush remaining output.

pyagent_trace.exporters.jsonl.JsonlExporter

Export trace events to a JSONL file.

Usage

exporter = JsonlExporter("traces/run_001.jsonl") bus.subscribe(exporter.export_event)

export_event(event)

Append a trace event as a JSON line.

flush()

Flush buffered writes to disk.

shutdown()

Flush and close the file.

pyagent_trace.exporters.langfuse.LangfuseExporter

Export trace events to Langfuse.

Usage

exporter = LangfuseExporter( public_key="pk-...", secret_key="sk-...", host="https://cloud.langfuse.com", ) bus.subscribe(exporter.export_event)

Requires: pip install pyagent-trace[langfuse]

export_event(event)

Export a trace event to Langfuse.

flush()

Flush pending events to Langfuse.

shutdown()

Flush and shutdown the Langfuse client.

Spans

pyagent_trace.spans.PatternSpanEmitter

Emit OTel spans for pattern executions.

Usage

emitter = PatternSpanEmitter() with emitter.pattern_span("debate", {"rounds": 3}) as span: # ... pattern logic ... emitter.set_result(span, result)

agent_span(agent_name, parent_span=None, attributes=None)

Start a span for an individual agent call.

pattern_span(pattern_type, attributes=None)

Start a span for a pattern execution.

set_compression_info(span, input_tokens, output_tokens, savings_pct) staticmethod

Set compression attributes on a span.

set_error(span, error) staticmethod

Record an error on a span.

set_pattern_result(span, output_length, rounds=None, consensus=None, escalated=False, duration_ms=0.0, token_estimate=0, cost_estimate=0.0) staticmethod

Set result attributes on a pattern span.

set_routing_info(span, difficulty, selected_model, cost_estimate, category='') staticmethod

Set routing attributes on a span.

pyagent_trace.attributes.PyAgentAttributes

Attribute key constants for pyagent OTel spans.

Decorators

pyagent_trace.decorators.traced_pattern(cls)

Class decorator: auto-emit OTel spans for every pattern.run() call.

Usage

@traced_pattern class MyPattern(Pattern): ...

pyagent_trace.decorators.traced_agent(agent)

Wrap an Agent instance to emit OTel spans on every run() call.

Usage

agent = traced_agent(Agent("my_agent", llm))

Cost Tracking

pyagent_trace.cost.CostTracker

Track costs across an entire workflow.

Usage

tracker = CostTracker() tracker.record("debate", "bull_agent", "gpt-4o", 500, 200, 0.003) tracker.record("debate", "bear_agent", "gpt-4o-mini", 500, 200, 0.0004) print(tracker.summary())

by_agent()

Cost breakdown by agent name.

by_model()

Cost breakdown by model.

by_pattern()

Cost breakdown by pattern type.

record(pattern_type, agent_name, model, input_tokens, output_tokens, cost_usd)

Record a cost entry.

summary()

Full cost summary.

pyagent_trace.cost.CostEntry dataclass

A single cost record.

Record & Replay

pyagent_trace.recorder.Recorder

Record pattern executions for debugging and replay.

Usage

recorder = Recorder() recorder.start("debate") recorder.record_llm_call("bull", messages, response) recorder.save("debug_trace.jsonl")

end(result_output)

Mark the end of a pattern execution.

load(path) classmethod

Load recorded entries from a JSONL file.

record_llm_call(agent_name, messages, response, metadata=None)

Record an LLM call and its response.

save(path)

Save recorded entries to a JSONL file.

start(pattern_type)

Mark the start of a pattern execution.

pyagent_trace.recorder.RecordEntry dataclass

A single recorded event.