trulens.core.otel.instrument¶

trulens.core.otel.instrument ¶

Attributes¶

Classes¶

span_group ¶

Context manager that tags every span created inside the block with a group label via SpanAttributes.SPAN_GROUPS.

Uses a contextvars.ContextVar — no OTEL baggage, no cross-process propagation. Span groups are an in-process concept.

Example::

with span_group("hop1"):
    ctx1 = retrieve("query 1")   # span gets SPAN_GROUPS=["hop1"]
with span_group("hop2"):
    ctx2 = retrieve("query 2")   # span gets SPAN_GROUPS=["hop2"]

Nesting merges groups::

with span_group("hop1"):
    with span_group("retry"):
        retrieve("q")            # SPAN_GROUPS=["hop1", "retry"]

prompt_lineage ¶

Context manager that tags the current span with prompt lineage.

The span belongs to the caller. This writes the prompt id, slug, exact version id, requested label, and rendered-content hash onto it, and calls no model.

Example::

resolved = session.get_prompt("support-assistant", label="production")
request = resolved.render(question=question)

with prompt_lineage(request):
    answer = my_generation_call(request.messages)

instrument ¶

Functions¶

init ¶

__init__(
    *,
    name: str | None = None,
    span_type: SpanType = UNKNOWN,
    attributes: Attributes = None,
    **kwargs
) -> None

Decorator for marking functions to be instrumented with OpenTelemetry tracing.

Optional custom span name. If not provided, derives from function's

module.qualname. Use this for clean span names without module prefixes (e.g., name="call_llm" instead of "main.call_llm").

span_type: Span type to be used for the span. attributes: A dictionary or a callable that returns a dictionary of attributes (i.e. a typing.Dict[str, typing.Any]) to be set on the span.

OtelBaseRecordingContext ¶

Attributes¶

run_name `instance-attribute` ¶

run_name: str = run_name

The name of the run that the recording context is currently processing.

input_id `instance-attribute` ¶

input_id: str = input_id

The ID of the input that the recording context is currently processing.

tokens `class-attribute` `instance-attribute` ¶

tokens: list[object] = []

OTEL context tokens for the current context manager. These tokens are how the OTEL context api keeps track of what is changed in the context, and used to undo the changes.

context_keys_added `class-attribute` `instance-attribute` ¶

context_keys_added: list[str] = []

Keys added to the OTEL context.

OtelRecordingContext ¶

Bases: OtelBaseRecordingContext

Attributes¶

record_ids `property` ¶

record_ids: list[str]

Record IDs created inside this recording context.

run_name `instance-attribute` ¶

run_name: str = run_name

The name of the run that the recording context is currently processing.

input_id `instance-attribute` ¶

input_id: str = input_id

The ID of the input that the recording context is currently processing.

tokens `class-attribute` `instance-attribute` ¶

tokens: list[object] = []

OTEL context tokens for the current context manager. These tokens are how the OTEL context api keeps track of what is changed in the context, and used to undo the changes.

context_keys_added `class-attribute` `instance-attribute` ¶

context_keys_added: list[str] = []

Keys added to the OTEL context.

Functions¶

set_prompt_lineage_attributes ¶

set_prompt_lineage_attributes(
    span: Span, rendered: "prompt_schema.RenderedPrompt"
) -> None

Attach prompt lineage to a span the caller owns.

Only identifiers and a hash of the rendered content are written, so this works with GenAI content capture off and never copies a prompt body into the span.

PARAMETER	DESCRIPTION
`span`	The span to write to. Ignored when it is not recording. TYPE: `Span`
`rendered`	The result of rendering one exact prompt version. TYPE: `'prompt_schema.RenderedPrompt'`

extract_input_content ¶

extract_input_content(messages: Sequence[Any]) -> str

Extract the text content from the input messages.

Looks for the last HumanMessage's content, or falls back to the first message's content if no HumanMessage is found.

PARAMETER	DESCRIPTION
`messages`	List of message objects (e.g., LangChain messages) TYPE: `Sequence[Any]`

RETURNS	DESCRIPTION
`str`	The extracted text content as a string

extract_output_content ¶

extract_output_content(ret: Any) -> str

Extract the text content from an LLM response.

PARAMETER	DESCRIPTION
`ret`	The LLM response object TYPE: `Any`

RETURNS	DESCRIPTION
`str`	The extracted text content as a string

extract_tool_calls ¶

extract_tool_calls(ret: Any) -> str | None

Extract and format tool calls from an LLM response.

Formats tool calls as: "tool_name(arg1=val1, arg2=val2), other_tool(...)"

PARAMETER	DESCRIPTION
`ret`	The LLM response object TYPE: `Any`

RETURNS	DESCRIPTION
`str \| None`	Formatted string of tool calls, or None if no tool calls

generation_attributes ¶

generation_attributes() -> Callable

Create an attributes lambda for GENERATION spans.

Extracts input_content, output_content, and tool_calls from the function call and return value.

RETURNS	DESCRIPTION
`Callable`	A callable suitable for the @instrument(attributes=...) parameter

Example

@instrument( name="call_llm", span_type=SpanAttributes.SpanType.GENERATION, attributes=generation_attributes() ) def call_llm(messages): return model.invoke(messages)

instrument_tools ¶

instrument_tools(
    tools_by_name: dict[str, Any],
    *,
    invoke_method: str = "invoke"
) -> None

Instrument a tools dictionary in place for clean tool span names.

Replaces each tool in the dictionary with a wrapper that produces spans named after the tool (e.g., "add", "multiply") when invoke() is called.

This is the least invasive way to get clean tool spans - no changes to app code required beyond this one setup call.

PARAMETER	DESCRIPTION
`tools_by_name`	Dictionary mapping tool names to tool objects TYPE: `dict[str, Any]`
`invoke_method`	Name of the method to wrap (default: "invoke") TYPE: `str` DEFAULT: `'invoke'`

Example

tools_by_name = {"add": add_tool, "multiply": multiply_tool}

One line setup - wraps tools in place¶

instrument_tools(tools_by_name)

App code unchanged - just call tool.invoke() as normal¶

def call_tool(tool_call): tool = tools_by_name[tool_call["name"]] result = tool.invoke(tool_call["args"]) # Now creates "add" span return ToolMessage(content=result, ...)

trulens.core.otel.instrument¶