Function use_inference_span

src/google/adk/telemetry/tracing.py:581–634

Context manager encompassing `generate_content {model.name}` span. When an external library for inference instrumentation is installed (e.g. opentelemetry-instrumentation-google-genai), span creation is delegated to said library.

(
    llm_request: LlmRequest,
    invocation_context: InvocationContext,
    model_response_event: Event,
)

@asynccontextmanager
async def use_inference_span(
    llm_request: LlmRequest,
    invocation_context: InvocationContext,
    model_response_event: Event,
) -> AsyncIterator[GenerateContentSpan | None]:
  """Context manager encompassing `generate_content {model.name}` span.

  When an external library for inference instrumentation is installed (e.g.
  opentelemetry-instrumentation-google-genai),
  span creation is delegated to said library.
  """

  telemetry_config = _telemetry_config_from_invocation_context(
      invocation_context
  )
  common_attributes = {
      GEN_AI_AGENT_NAME: invocation_context.agent.name,
      GEN_AI_CONVERSATION_ID: invocation_context.session.id,
      "gcp.vertex.agent.event_id": model_response_event.id,
      "gcp.vertex.agent.invocation_id": invocation_context.invocation_id,
  }
  log_only_common_attributes = {}
  if invocation_context.session.user_id is not None:
    log_only_common_attributes[USER_ID] = invocation_context.session.user_id
  if _should_emit_native_telemetry(invocation_context.agent):
    async with _use_native_generate_content_span(
        llm_request=llm_request,
        common_attributes=common_attributes,
        log_only_common_attributes=log_only_common_attributes,
        telemetry_config=telemetry_config,
    ) as gc_span:
      if telemetry_config.should_use_experimental_genai_semconv:
        set_operation_details_common_attributes(
            gc_span.operation_details_common_attributes,
            telemetry_config,
            common_attributes,
            log_only_attributes=log_only_common_attributes,
        )
      try:
        yield gc_span
      finally:
        maybe_log_completion_details(
            gc_span.span,
            otel_logger,
            gc_span.operation_details_attributes,
            gc_span.operation_details_common_attributes,
            telemetry_config,
        )
  else:
    with _use_extra_generate_content_attributes(
        common_attributes,
        log_only_extra_attributes=log_only_common_attributes,
    ):
      yield


def _instrumented_with_opentelemetry_instrumentation_google_genai() -> bool: