Skip to content

perf: skip discarded trace payload construction #7155

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chenghao-mou merged 4 commits into from
Sep 9, 2026
+251 −45
Merged

perf: skip discarded trace payload construction #7155

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0d27ae9
perf: skip discarded trace payload construction
chenghao-mou Sep 7, 2026
784810d
perf: skip discarded response accumulation
chenghao-mou Sep 8, 2026
a2773ef
fix: prevent partial trace responses
chenghao-mou Sep 8, 2026
bc636d7
refactor: make stream content guard best effort
chenghao-mou Sep 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 23 additions & 17 deletions livekit-agents/livekit/agents/llm/llm.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -245,6 +245,8 @@ def __init__(
self._event_aiter, monitor_aiter = self._tee_aiter
self._current_attempt_has_error = False
self._provider_request_ids: list[str] = []
self._llm_request_span: trace.Span | None = None
self._record_content = False
self._metrics_task = asyncio.create_task(
self._metrics_monitor_task(monitor_aiter), name="LLM._metrics_task"
)
Expand All @@ -257,14 +259,16 @@ async def _traceable_main_task() -> None:
with tracer.start_as_current_span(
self._llm_request_span_name, end_on_exit=False
) as span:
# Enabling capture later must not emit a partial response.
self._record_content = (
span.is_recording() and gen_ai_telemetry.capture_content_enabled()
)
self._record_genai_request(span)
await self._main_task()

self._task = asyncio.create_task(_traceable_main_task(), name="LLM._main_task")
self._task.add_done_callback(lambda _: self._event_ch.close())

self._llm_request_span: trace.Span | None = None

@abstractmethod
async def _run(self) -> None: ...

Expand All @@ -278,12 +282,13 @@ def _record_genai_request(self, span: trace.Span) -> None:
stream=True,
output_type=trace_types.GenAIOutputType.TEXT,
)
gen_ai_telemetry.set_content_attributes(
span,
system_instructions=gen_ai_telemetry.to_system_instructions(self._chat_ctx),
input_messages=gen_ai_telemetry.to_input_messages(self._chat_ctx),
tool_definitions=gen_ai_telemetry.to_tool_definitions(self._tools),
)
if self._record_content:
gen_ai_telemetry.set_content_attributes(
span,
system_instructions=gen_ai_telemetry.to_system_instructions(self._chat_ctx),
input_messages=gen_ai_telemetry.to_input_messages(self._chat_ctx),
tool_definitions=gen_ai_telemetry.to_tool_definitions(self._tools),
)

async def _main_task(self) -> None:
self._llm_request_span = trace.get_current_span()
Expand Down Expand Up @@ -378,7 +383,7 @@ async def _metrics_monitor_task(self, event_aiter: AsyncIterable[ChatChunk]) ->
completion_start_time = datetime.now(timezone.utc).isoformat()

if ev.delta:
if ev.delta.content:
if ev.delta.content and self._record_content:
response_content += ev.delta.content
if ev.delta.tool_calls:
tool_calls.extend(ev.delta.tool_calls)
Expand Down Expand Up @@ -430,14 +435,15 @@ async def _metrics_monitor_task(self, event_aiter: AsyncIterable[ChatChunk]) ->
finish_reasons=[finish_reason],
time_to_first_chunk=ttft if ttft >= 0 else None,
)
gen_ai_telemetry.set_content_attributes(
self._llm_request_span,
output_messages=gen_ai_telemetry.to_output_messages(
text=response_content,
function_calls=tool_calls,
finish_reason=finish_reason,
),
)
if self._record_content:
Comment thread
chenghao-mou marked this conversation as resolved.
gen_ai_telemetry.set_content_attributes(
self._llm_request_span,
output_messages=gen_ai_telemetry.to_output_messages(
text=response_content,
function_calls=tool_calls,
finish_reason=finish_reason,
),
)
if completion_start_time:
self._llm_request_span.set_attribute(
trace_types.ATTR_LANGFUSE_COMPLETION_START_TIME, f'"{completion_start_time}"'
Expand Down
54 changes: 29 additions & 25 deletions livekit-agents/livekit/agents/voice/generation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -192,20 +192,23 @@ async def _llm_inference_task(
text_ch, function_ch = data.text_ch, data.function_ch
tools = tool_ctx.flatten()

attrs: dict[str, Any] = {
trace_types.ATTR_CHAT_CTX: json.dumps(
chat_ctx.to_dict(
exclude_audio=True,
exclude_image=True,
exclude_timestamp=True,
exclude_metrics=True,
)
),
trace_types.ATTR_FUNCTION_TOOLS: list(tool_ctx.function_tools.keys()),
trace_types.ATTR_PROVIDER_TOOLS: [type(tool).__name__ for tool in tool_ctx.provider_tools],
trace_types.ATTR_TOOL_SETS: [type(tool_set).__name__ for tool_set in tool_ctx.toolsets],
}
current_span.set_attributes(attrs)
if current_span.is_recording():
attrs: dict[str, Any] = {
trace_types.ATTR_CHAT_CTX: json.dumps(
chat_ctx.to_dict(
exclude_audio=True,
exclude_image=True,
exclude_timestamp=True,
exclude_metrics=True,
)
),
trace_types.ATTR_FUNCTION_TOOLS: list(tool_ctx.function_tools.keys()),
trace_types.ATTR_PROVIDER_TOOLS: [
type(tool).__name__ for tool in tool_ctx.provider_tools
],
trace_types.ATTR_TOOL_SETS: [type(tool_set).__name__ for tool_set in tool_ctx.toolsets],
}
current_span.set_attributes(attrs)

# the GenAI inference attributes belong to the nested `llm_request` span, which is the
# provider call the convention describes — setting them here as well would make a
Expand Down Expand Up @@ -391,17 +394,18 @@ def _record_uninstrumented_inference(
gen_ai_telemetry.set_response_attributes(
span, finish_reasons=[finish_reason], time_to_first_chunk=data.ttft
)
gen_ai_telemetry.set_content_attributes(
span,
system_instructions=gen_ai_telemetry.to_system_instructions(chat_ctx),
input_messages=gen_ai_telemetry.to_input_messages(chat_ctx),
tool_definitions=gen_ai_telemetry.to_tool_definitions(tools),
output_messages=gen_ai_telemetry.to_output_messages(
text=data.generated_text,
function_calls=data.generated_functions,
finish_reason=finish_reason,
),
)
if span.is_recording() and gen_ai_telemetry.capture_content_enabled():
gen_ai_telemetry.set_content_attributes(
span,
system_instructions=gen_ai_telemetry.to_system_instructions(chat_ctx),
input_messages=gen_ai_telemetry.to_input_messages(chat_ctx),
tool_definitions=gen_ai_telemetry.to_tool_definitions(tools),
output_messages=gen_ai_telemetry.to_output_messages(
text=data.generated_text,
function_calls=data.generated_functions,
finish_reason=finish_reason,
),
)
if usage is not None:
gen_ai_telemetry.set_usage_attributes(span, usage)

Expand Down
Loading