latent.agents.observability.sink_middleware¶

AgentSinkMiddleware: pipes all agent stream events to a list of sinks.

This is the general counterpart to GuardrailMiddleware — it adds observability (logging, tracing) to any agent without touching its logic.

Example::

from latent.agents.observability.sink_middleware import AgentSinkMiddleware
from latent.agents.observability.sinks import StructlogSink, OpenTelemetrySink

agent = MyAgent()
wrapped = AgentSinkMiddleware(agent, sinks=[StructlogSink(), OpenTelemetrySink()])
async for event in wrapped.stream(messages):
    ...  # every TextDelta, ToolCall, LLMCallEnd, etc. flows to both sinks

Classes¶

AgentSinkMiddleware¶

AgentSinkMiddleware(agent: Any, sinks: list[AgentEventSink])

Wraps any agent and emits all stream events to a list of sinks.

Methods¶

AgentSinkMiddleware.stream¶

stream(messages: list[Message], config: Any = None) -> AsyncIterator[AgentEvent]