trulens.core.utils.trace_provider¶

trulens.core.utils.trace_provider ¶

Provider interface for trace-specific parsing and plan extraction.

This allows different app integrations (LangGraph, LangChain, etc.) to provide their own trace parsing logic while keeping the core compression generic.

Classes¶

TraceProvider ¶

Bases: ABC

Abstract base class for trace provider-specific parsing.

Functions¶

can_handle `abstractmethod` ¶

can_handle(trace_data: Dict[str, Any]) -> bool

Check if this provider can handle the given trace data.

PARAMETER	DESCRIPTION
`trace_data`	Raw trace data to check TYPE: `Dict[str, Any]`

RETURNS	DESCRIPTION
`bool`	True if this provider can parse this trace format

extract_plan `abstractmethod` ¶

extract_plan(trace_data: Dict[str, Any]) -> Optional[Any]

Extract plan information from trace data.

PARAMETER	DESCRIPTION
`trace_data`	Raw trace data TYPE: `Dict[str, Any]`

RETURNS	DESCRIPTION
`Optional[Any]`	Plan data if found, None otherwise

extract_execution_flow `abstractmethod` ¶

extract_execution_flow(
    trace_data: Dict[str, Any]
) -> List[Dict[str, Any]]

Extract execution flow from trace data.

PARAMETER	DESCRIPTION
`trace_data`	Raw trace data TYPE: `Dict[str, Any]`

RETURNS	DESCRIPTION
`List[Dict[str, Any]]`	List of execution steps

extract_agent_interactions `abstractmethod` ¶

extract_agent_interactions(
    trace_data: Dict[str, Any]
) -> List[Dict[str, Any]]

Extract agent interactions from trace data.

PARAMETER	DESCRIPTION
`trace_data`	Raw trace data TYPE: `Dict[str, Any]`

RETURNS	DESCRIPTION
`List[Dict[str, Any]]`	List of agent interactions

compress_trace ¶

compress_trace(
    trace_data: Dict[str, Any]
) -> Dict[str, Any]

Compress trace data while preserving essential information.

PARAMETER	DESCRIPTION
`trace_data`	Raw trace data to compress TYPE: `Dict[str, Any]`

RETURNS	DESCRIPTION
`Dict[str, Any]`	Compressed trace data

compress_with_plan_priority ¶

compress_with_plan_priority(
    trace_data: Dict[str, Any],
    target_token_limit: int = DEFAULT_TOKEN_LIMIT,
) -> Dict[str, Any]

Compress trace with plan preservation as highest priority. If context window is exceeded, compress other data more aggressively.

PARAMETER	DESCRIPTION
`trace_data`	Raw trace data to compress TYPE: `Dict[str, Any]`
`target_token_limit`	Target token limit for context window management TYPE: `int` DEFAULT: `DEFAULT_TOKEN_LIMIT`

RETURNS	DESCRIPTION
`Dict[str, Any]`	Compressed trace data with plan preservation prioritized

GenericTraceProvider ¶

Bases: TraceProvider

Generic trace provider for standard trace formats.

Functions¶

can_handle ¶

can_handle(trace_data: Dict[str, Any]) -> bool

Generic provider handles any trace data as fallback.

extract_plan ¶

extract_plan(trace_data: Dict[str, Any]) -> Optional[Any]

Extract plan using generic field names.

extract_execution_flow ¶

extract_execution_flow(
    trace_data: Dict[str, Any]
) -> List[Dict[str, Any]]

Extract execution flow using generic span structure.

extract_agent_interactions ¶

extract_agent_interactions(
    trace_data: Dict[str, Any]
) -> List[Dict[str, Any]]

Extract agent interactions using generic message structure.

compress_trace ¶

compress_trace(
    trace_data: Dict[str, Any]
) -> Dict[str, Any]

Compress trace data while preserving essential information.

PARAMETER	DESCRIPTION
`trace_data`	Raw trace data to compress TYPE: `Dict[str, Any]`

RETURNS	DESCRIPTION
`Dict[str, Any]`	Compressed trace data

compress_with_plan_priority ¶

compress_with_plan_priority(
    trace_data: Dict[str, Any],
    target_token_limit: int = DEFAULT_TOKEN_LIMIT,
) -> Dict[str, Any]

Compress trace with plan preservation as highest priority. If context window is exceeded, compress other data more aggressively.

PARAMETER	DESCRIPTION
`trace_data`	Raw trace data to compress TYPE: `Dict[str, Any]`
`target_token_limit`	Target token limit for context window management TYPE: `int` DEFAULT: `DEFAULT_TOKEN_LIMIT`

RETURNS	DESCRIPTION
`Dict[str, Any]`	Compressed trace data with plan preservation prioritized

TraceProviderRegistry ¶

Registry for trace providers with priority ordering.

Functions¶

register_provider ¶

register_provider(provider: TraceProvider)

Register a trace provider. Last registered has highest priority.

get_provider ¶

get_provider(trace_data: Dict[str, Any]) -> TraceProvider

Get the first provider that can handle the trace data.

Functions¶

register_trace_provider ¶

register_trace_provider(provider: TraceProvider)

Register a trace provider globally.

get_trace_provider ¶

get_trace_provider(
    trace_data: Dict[str, Any]
) -> TraceProvider

Get appropriate trace provider for the given data.

trulens.core.utils.trace_provider¶