latent.agents.observability.sinks¶
General-purpose event sinks for any AgentEvent stream.
Sinks observe events without affecting the stream. Pass them to
:class:AgentSinkMiddleware to attach logging or tracing to any agent.
Example::
from latent.agents.observability.sinks import StructlogSink
from latent.agents.observability.sink_middleware import AgentSinkMiddleware
agent = MyAgent()
wrapped = AgentSinkMiddleware(agent, sinks=[StructlogSink()])
async for event in wrapped.stream(messages):
...
Classes¶
AgentEventSink¶
Protocol for sinks that consume any AgentEvent.
OpenTelemetrySink¶
Emits any AgentEvent as an OpenTelemetry span event.
No-op when opentelemetry-api is not installed or no span is active.
StructlogSink¶
StructlogSink(logger_name: str = 'latent.agents', level: int = logging.INFO, max_text_chars: int = 500)
Logs any AgentEvent as a structured log record.
Uses :func:dataclasses.asdict for serialisation so all event fields are
captured without requiring a custom to_dict method on each event type.