Function start_as_current_node_span

src/google/adk/telemetry/node_tracing.py:59–109

Creates a scope-based OpenTelemetry span, representing a node invocation. Implements emitting of the following spans: - `invoke_agent {agent.name}` - `invoke_workflow {workflow.name}` - `invoke_node {node.name}` invoke_agent spans align with OpenTelemetry Semantic Conventions (semconv) v

(
    context: Context, node: BaseNode
)

@asynccontextmanager
async def start_as_current_node_span(
    context: Context, node: BaseNode
) -> AsyncIterator[TelemetryContext]:
  """Creates a scope-based OpenTelemetry span, representing a node invocation.

  Implements emitting of the following spans:
  - `invoke_agent {agent.name}`
  - `invoke_workflow {workflow.name}`
  - `invoke_node {node.name}`

  invoke_agent spans align with OpenTelemetry Semantic Conventions (semconv) version 1.36 spans for backwards compatibility.
  https://github.com/open-telemetry/semantic-conventions/blob/v1.36.0/docs/gen-ai/README.md

  invoke_workflow spans align with semconv version 1.41, because these were not included in any prior releases.
  https://github.com/open-telemetry/semantic-conventions/blob/main/docs/gen-ai/README.md

  invoke_node spans are not present in any semconv release.
  We will create a proposal to standardize them.

  Args:
    context: Context in which the span is created.
    node: The node to be invoked inside the created span.

  Yields:
    Context with the started span.
  """

  span_metadata = _span_metadata(context, node)
  if span_metadata is None:
    token = context_api.attach(context.telemetry_context.otel_context)
    try:
      yield TelemetryContext(
          otel_context=context.telemetry_context.otel_context
      )
    finally:
      context_api.detach(token)
    return

  with tracer.start_as_current_span(
      span_metadata.name,
      attributes=span_metadata.attributes,
      context=context.telemetry_context.otel_context,
  ) as span:
    telemetry_context = TelemetryContext(otel_context=context_api.get_current())
    yield telemetry_context

    if span.is_recording() and len(telemetry_context._associated_event_ids) > 0:
      span.set_attribute(
          "gcp.vertex.agent.associated_event_ids",
          telemetry_context._associated_event_ids,
      )


def _span_metadata(context: Context, node: BaseNode) -> _SpanMetadata | None: