kodeagent.tracer#

Interfaces for hierarchical tracing and agent observability.

Supports multiple observability backends (e.g., Langfuse and LangSmith) with a unified API for creating traces, spans, and generations. Provides no-op implementations when tracing is disabled.

create_tracer_manager

Factory function to create a tracer manager based on the specified type.

AbstractObservation

Abstract interface for trace observations.

AbstractTracerManager

Abstract interface for tracer management.

LangSmithObservation

LangSmith implementation of observation.

LangSmithTracerManager

LangSmith implementation of TracerManager.

LangfuseObservation

Langfuse implementation of observation.

LangfuseTracerManager

Langfuse implementation of TracerManager.

NoOpObservation

No-op observation implementation.

NoOpTracerManager

No-op tracer manager implementation.

Interfaces for hierarchical tracing and agent observability.

Supports multiple observability backends (e.g., Langfuse and LangSmith) with a unified API for creating traces, spans, and generations. Provides no-op implementations when tracing is disabled.

class kodeagent.tracer.AbstractObservation[source]#

Bases: ABC

Abstract interface for trace observations.

Represents a single node in a hierarchical trace tree. Can be a top-level trace, a nested span, or an LLM generation. Implements context manager protocol for use with ‘with’ statements.

abstract end(**kwargs: Any) → None[source]#

Explicitly signal the end of the observation.

Records final state and duration. Called automatically when using context manager protocol.

Parameters:

**kwargs – Provider-specific properties (output, result, error, etc).

abstract update(**kwargs: Any) → None[source]#

Update observation properties during execution.

Used to log intermediate states like partial outputs, status, or metadata.

Parameters:

**kwargs – Provider-specific properties (output, status, level, etc).

class kodeagent.tracer.AbstractTracerManager[source]#

Bases: ABC

Abstract interface for tracer management.

Factory for creating hierarchical observations. Handles initialization and backend-specific configuration. Implementations should support at least one tracing backend (Langfuse, LangSmith, etc.) or be a no-op when tracing is disabled.

abstract flush() → None[source]#

Flush any buffered traces to the backend.

Ensures that all recorded traces and spans are sent to the observability platform before the application exits.

abstract start_generation(parent: AbstractObservation, name: str, input_data: Any) → AbstractObservation[source]#

Start a nested LLM generation under a parent observation.

Used specifically for LLM calls within a trace or span.

Parameters:

  • parent – Parent observation (trace or span).

  • name – Identifier for the generation operation.

  • input_data – Input data (e.g., prompt) to log for the generation.

Returns:

An observation object for the generation.

abstract start_span(parent: AbstractObservation, name: str, input_data: Any) → AbstractObservation[source]#

Start a nested span under a parent observation.

Used for logical sub-operations within a trace or parent span.

Parameters:

  • parent – Parent observation (trace or span).

  • name – Identifier for the span operation.

  • input_data – Input data to log for the span.

Returns:

An observation object for the span.

abstract start_trace(name: str, input_data: Any) → AbstractObservation[source]#

Start a new top-level trace.

Parameters:

  • name – Identifier for the trace operation.

  • input_data – Input data to log for the trace.

Returns:

An observation object for the trace root.

class kodeagent.tracer.LangSmithObservation(run_tree: Any)[source]#

Bases: AbstractObservation

LangSmith implementation of observation.

Wraps a LangSmith RunTree object to manage hierarchical runs.

Initialize LangSmith observation.

Parameters:

run_tree – The LangSmith RunTree object.

end(**kwargs: Any) → None[source]#

End the observation and send final data to LangSmith.

Parameters:

**kwargs – Final output/result data.

update(**kwargs: Any) → None[source]#

Update observation outputs during execution.

Parameters:

**kwargs – Output data to accumulate.

class kodeagent.tracer.LangSmithTracerManager[source]#

Bases: AbstractTracerManager

LangSmith implementation of TracerManager.

Uses LangSmith RunTree to manage hierarchical runs.

Initialize the LangSmith client.

flush() → None[source]#

Flush LangSmith runs.

start_generation(parent: AbstractObservation, name: str, input_data: Any) → AbstractObservation[source]#

Start a nested LLM generation.

Parameters:

  • parent – Parent observation (assumes LangSmithObservation).

  • name – Identifier for the generation operation.

  • input_data – Input data (e.g., prompt) to log for the generation.

Returns:

A LangSmith generation object wrapped as AbstractObservation.

start_span(parent: AbstractObservation, name: str, input_data: Any) → AbstractObservation[source]#

Start a nested span under a parent observation.

Parameters:

  • parent – Parent observation (assumes LangSmithObservation).

  • name – Identifier for the span operation.

  • input_data – Input data to log for the span.

Returns:

A LangSmith span object wrapped as AbstractObservation.

start_trace(name: str, input_data: Any) → AbstractObservation[source]#

Start a new trace with LangSmith.

class kodeagent.tracer.LangfuseObservation(obj: Any)[source]#

Bases: AbstractObservation

Langfuse implementation of observation.

Wraps Langfuse Trace, Span, or Generation objects to provide a consistent interface.

Initialize Langfuse observation.

Parameters:

obj – The Langfuse Trace, Span, or Generation object.

end(**kwargs: Any) → None[source]#

End the observation.

Maps ‘result’ to ‘output’ for compatibility with Langfuse. Calls end() if available (Spans/Generations), else update() (Traces).

Parameters:

**kwargs – Final state data.

update(**kwargs: Any) → None[source]#

Update observation properties.

Parameters:

**kwargs – Properties to update.

class kodeagent.tracer.LangfuseTracerManager[source]#

Bases: AbstractTracerManager

Langfuse implementation of TracerManager.

Integrates with Langfuse observability platform to create and manage hierarchical traces, spans, and generations. Tracing is disabled if the Langfuse package is not installed.

Initialize the Langfuse client.

flush() → None[source]#

Flush Langfuse traces.

start_generation(parent: AbstractObservation, name: str, input_data: Any) → AbstractObservation[source]#

Start a nested LLM generation under a parent observation with Langfuse.

Parameters:

  • parent – Parent observation (assumes LangfuseObservation).

  • name – Identifier for the generation operation.

  • input_data – Input data (e.g., prompt) to log for the generation.

Returns:

A Langfuse generation object wrapped as AbstractObservation.

start_span(parent: AbstractObservation, name: str, input_data: Any) → AbstractObservation[source]#

Start a nested span under a parent observation with Langfuse.

Parameters:

  • parent – Parent observation (assumes LangfuseObservation).

  • name – Identifier for the span operation.

  • input_data – Input data to log for the span.

Returns:

A Langfuse span object wrapped as AbstractObservation.

start_trace(name: str, input_data: Any) → AbstractObservation[source]#

Start a new trace with Langfuse.

Parameters:

  • name – Identifier for the trace operation.

  • input_data – Input data to log for the trace.

Returns:

A Langfuse trace object wrapped as AbstractObservation.

class kodeagent.tracer.NoOpObservation[source]#

Bases: AbstractObservation

No-op observation implementation.

Used when tracing is disabled. All methods are no-ops and return self to support hierarchical nesting without side effects.

end(**kwargs: Any) → None[source]#

No-op: ignore end signal.

update(**kwargs: Any) → None[source]#

No-op: ignore all property updates.

class kodeagent.tracer.NoOpTracerManager[source]#

Bases: AbstractTracerManager

No-op tracer manager implementation.

Used when no observability backend is enabled. Provides a complete no-op implementation of the TracerManager interface that satisfies the contract while performing no actual tracing operations.

flush() → None[source]#

No-op: do nothing.

start_generation(parent: AbstractObservation, name: str, input_data: Any) → AbstractObservation[source]#

No-op: return a no-op observation.

start_span(parent: AbstractObservation, name: str, input_data: Any) → AbstractObservation[source]#

No-op: return a no-op observation.

start_trace(name: str, input_data: Any) → AbstractObservation[source]#

No-op: return a no-op observation.

kodeagent.tracer.create_tracer_manager(tracing_type: Literal['langfuse', 'langsmith'] | None = None) → AbstractTracerManager[source]#

Factory function to create a tracer manager based on the specified type.

Parameters:

tracing_type – The type of tracing backend to use. Defaults to None for no-op tracing.

Returns:

An instance of LangfuseTracerManager, LangSmithTracerManager, or NoOpTracerManager for the specified tracing backend.