-
Notifications
You must be signed in to change notification settings - Fork 2.3k
Python: Show FoundryAgent client spans in Foundry traces #7981
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.
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
base: main
Are you sure you want to change the base?
Changes from all commits
c133e29
2ac2114
c46ff52
7180c58
79db809
7695b7f
a3eeaee
807a12a
4f3b8b1
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -124,6 +124,9 @@ | |
| ) | ||
| INNER_RESPONSE_ID_CAPTURED_FIELD: Final[str] = "response_id" | ||
| INNER_USAGE_CAPTURED_FIELD: Final[str] = "usage" | ||
| INNER_CAPTURED_RESPONSE_ID: Final[contextvars.ContextVar[str | None]] = contextvars.ContextVar( | ||
| "inner_captured_response_id", default=None | ||
| ) | ||
|
|
||
| # Tracks accumulated token usage from all inner chat completion spans within an agent invoke. | ||
| INNER_ACCUMULATED_USAGE: Final[contextvars.ContextVar[UsageDetails | None]] = contextvars.ContextVar( | ||
|
|
@@ -2290,6 +2293,14 @@ def __init__( | |
| self.token_usage_histogram = _get_token_usage_histogram() | ||
| self.duration_histogram = _get_duration_histogram() | ||
|
|
||
| def _get_additional_otel_agent_attributes(self) -> Mapping[str, Any]: | ||
| """Return provider-specific attributes emitted on agent spans.""" | ||
| return {} | ||
|
|
||
| def _should_capture_agent_response_id(self) -> bool: | ||
| """Return whether the agent span must retain an inner response ID.""" | ||
| return False | ||
|
|
||
| def _trace_agent_invocation( | ||
| self, | ||
| *, | ||
|
|
@@ -2333,6 +2344,7 @@ def _trace_agent_invocation( | |
| all_options=dict(merged_options), | ||
| **merged_client_kwargs, | ||
| ) | ||
| attributes.update(self._get_additional_otel_agent_attributes()) | ||
|
|
||
| if stream: | ||
| # Do NOT set the inner-telemetry context vars here: this synchronous run() body executes | ||
|
|
@@ -2344,6 +2356,7 @@ def _trace_agent_invocation( | |
| # below), so set and reset both happen in the consumer's context. | ||
| inner_response_telemetry_captured_fields: set[str] = set() | ||
| inner_response_telemetry_captured_fields_token: contextvars.Token[set[str] | None] | None = None | ||
| inner_captured_response_id_token: contextvars.Token[str | None] | None = None | ||
| inner_accumulated_usage_token: contextvars.Token[UsageDetails | None] | None = None | ||
| # Agent Framework's agents run in-process (the actual network call happens on a nested | ||
| # chat span), so invoke_agent spans use the default INTERNAL kind. | ||
|
|
@@ -2411,10 +2424,14 @@ async def _finalize_stream() -> None: | |
| response_attributes = _get_response_attributes( | ||
| attributes, | ||
| response, | ||
| capture_response_id=INNER_RESPONSE_ID_CAPTURED_FIELD | ||
| not in inner_response_telemetry_captured_fields, | ||
| capture_response_id=( | ||
| self._should_capture_agent_response_id() | ||
| or INNER_RESPONSE_ID_CAPTURED_FIELD not in inner_response_telemetry_captured_fields | ||
| ), | ||
| capture_usage=INNER_USAGE_CAPTURED_FIELD not in inner_response_telemetry_captured_fields, | ||
| ) | ||
| if self._should_capture_agent_response_id(): | ||
| _apply_captured_response_id(response_attributes) | ||
| _apply_accumulated_usage(response_attributes, inner_response_telemetry_captured_fields) | ||
| _capture_response(span=span, attributes=response_attributes, duration=duration) | ||
| if ( | ||
|
|
@@ -2436,6 +2453,8 @@ async def _finalize_stream() -> None: | |
| # pull-context factory below set the tokens in — so the reset is cross-context safe. | ||
| if inner_response_telemetry_captured_fields_token is not None: | ||
| INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(inner_response_telemetry_captured_fields_token) | ||
| if inner_captured_response_id_token is not None: | ||
| INNER_CAPTURED_RESPONSE_ID.reset(inner_captured_response_id_token) | ||
| if inner_accumulated_usage_token is not None: | ||
| INNER_ACCUMULATED_USAGE.reset(inner_accumulated_usage_token) | ||
| _close_span() | ||
|
|
@@ -2447,11 +2466,14 @@ def _inner_telemetry_pull_context() -> contextlib.AbstractContextManager[Any]: | |
| # avoiding the cross-context Token reset failure. Setting happens before the | ||
| # underlying iterator is pulled, so inner chat completion spans created during the | ||
| # pull can still accumulate usage / mark captured fields. | ||
| nonlocal inner_response_telemetry_captured_fields_token, inner_accumulated_usage_token | ||
| nonlocal inner_response_telemetry_captured_fields_token | ||
| nonlocal inner_captured_response_id_token | ||
| nonlocal inner_accumulated_usage_token | ||
| if inner_response_telemetry_captured_fields_token is None: | ||
| inner_response_telemetry_captured_fields_token = INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.set( | ||
| inner_response_telemetry_captured_fields | ||
| ) | ||
| inner_captured_response_id_token = INNER_CAPTURED_RESPONSE_ID.set(None) | ||
| inner_accumulated_usage_token = INNER_ACCUMULATED_USAGE.set({}) | ||
| return _activate_span(span) | ||
|
|
||
|
|
@@ -2483,6 +2505,7 @@ async def _run() -> AgentResponse[Any]: | |
| inner_response_telemetry_captured_fields_token = INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.set( | ||
| inner_response_telemetry_captured_fields | ||
| ) | ||
| inner_captured_response_id_token = INNER_CAPTURED_RESPONSE_ID.set(None) | ||
| inner_accumulated_usage_token = INNER_ACCUMULATED_USAGE.set({}) | ||
| try: | ||
| with _get_span(attributes=attributes, span_name_attribute=OtelAttr.AGENT_NAME) as span: | ||
|
|
@@ -2500,12 +2523,16 @@ async def _run() -> AgentResponse[Any]: | |
| response_attributes = _get_response_attributes( | ||
| attributes, | ||
| response, | ||
| capture_response_id=INNER_RESPONSE_ID_CAPTURED_FIELD | ||
| not in inner_response_telemetry_captured_fields, | ||
| capture_response_id=( | ||
| self._should_capture_agent_response_id() | ||
| or INNER_RESPONSE_ID_CAPTURED_FIELD not in inner_response_telemetry_captured_fields | ||
| ), | ||
| capture_usage=( | ||
| INNER_USAGE_CAPTURED_FIELD not in inner_response_telemetry_captured_fields | ||
| ), | ||
| ) | ||
| if self._should_capture_agent_response_id(): | ||
| _apply_captured_response_id(response_attributes) | ||
| _apply_accumulated_usage( | ||
| response_attributes, | ||
| inner_response_telemetry_captured_fields, | ||
|
|
@@ -2531,6 +2558,7 @@ async def _run() -> AgentResponse[Any]: | |
| raise | ||
| finally: | ||
| INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(inner_response_telemetry_captured_fields_token) | ||
| INNER_CAPTURED_RESPONSE_ID.reset(inner_captured_response_id_token) | ||
| INNER_ACCUMULATED_USAGE.reset(inner_accumulated_usage_token) | ||
|
|
||
| return _run() | ||
|
|
@@ -3439,6 +3467,7 @@ def _mark_inner_response_telemetry_captured( | |
| return | ||
| if response.response_id: | ||
| captured_fields.add(INNER_RESPONSE_ID_CAPTURED_FIELD) | ||
| INNER_CAPTURED_RESPONSE_ID.set(response.response_id) | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. What happens when an |
||
| if response.usage_details: | ||
| captured_fields.add(INNER_USAGE_CAPTURED_FIELD) | ||
| accumulated = INNER_ACCUMULATED_USAGE.get() | ||
|
|
@@ -3448,6 +3477,12 @@ def _mark_inner_response_telemetry_captured( | |
| INNER_ACCUMULATED_USAGE.set(add_usage_details(accumulated, response.usage_details)) | ||
|
|
||
|
|
||
| def _apply_captured_response_id(attributes: dict[str, Any]) -> None: | ||
| """Apply the inner chat response ID to an agent span when the provider requires it.""" | ||
| if response_id := INNER_CAPTURED_RESPONSE_ID.get(): | ||
| attributes.setdefault(OtelAttr.RESPONSE_ID, response_id) | ||
|
|
||
|
|
||
| def _apply_accumulated_usage(attributes: dict[str, Any], captured_fields: set[str]) -> None: | ||
| """Apply accumulated usage from inner chat spans to the invoke_agent span attributes.""" | ||
| if INNER_USAGE_CAPTURED_FIELD not in captured_fields: | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -36,6 +36,7 @@ | |
| from agent_framework.observability import AgentTelemetryLayer, ChatTelemetryLayer | ||
| from agent_framework_openai._chat_client import OpenAIChatOptions, RawOpenAIChatClient | ||
| from azure.ai.projects.aio import AIProjectClient | ||
| from azure.ai.projects.models import ConnectionType | ||
| from azure.core.credentials import TokenCredential | ||
| from azure.core.credentials_async import AsyncTokenCredential | ||
|
|
||
|
|
@@ -95,6 +96,7 @@ class FoundryAgentSettings(TypedDict, total=False): | |
|
|
||
|
|
||
| FOUNDRY_HOSTED_AGENT_SESSION_ID_KEY = "foundry_hosted_agent_session_id" | ||
| _FOUNDRY_PROJECT_ARM_ID_ATTRIBUTE = "microsoft.foundry.project.id" | ||
|
|
||
|
|
||
| class FoundryAgentOptions(OpenAIChatOptions, total=False): | ||
|
|
@@ -750,6 +752,7 @@ def __init__( | |
| client_kwargs["function_invocation_configuration"] = function_invocation_configuration | ||
|
|
||
| client = actual_client_type(**client_kwargs) | ||
| self._foundry_project_arm_id: str | None = None | ||
|
|
||
| super().__init__( | ||
| client=client, # type: ignore[arg-type] | ||
|
|
@@ -841,6 +844,22 @@ def _update_session_from_chat_response_update( | |
| if session is not None and isinstance(agent_session_id, str) and agent_session_id: | ||
| session.state[FOUNDRY_HOSTED_AGENT_SESSION_ID_KEY] = agent_session_id | ||
|
|
||
| async def _get_foundry_project_arm_id(self) -> str: | ||
| """Get the Foundry project ARM ID from its Application Insights connection.""" | ||
| client = cast(RawFoundryAgentChatClient, self.client) | ||
| # AIProjectClient does not expose the project ARM ID directly. Derive it from the | ||
| # project-scoped connection until https://github.com/Azure/azure-sdk-for-python/issues/48825 is addressed. | ||
| connections = client.project_client.connections.list(connection_type=ConnectionType.APPLICATION_INSIGHTS) | ||
| async for connection in connections: | ||
| connection_suffix = f"/connections/{connection.name}" | ||
| if not connection.id.lower().endswith(connection_suffix.lower()): | ||
| raise ValueError( | ||
| f"The Foundry Application Insights connection ID has an unexpected format: {connection.id!r}." | ||
| ) | ||
| return connection.id[: -len(connection_suffix)] | ||
|
|
||
| raise ValueError("The Foundry project does not have an Application Insights connection.") | ||
|
|
||
| async def configure_azure_monitor( | ||
| self, | ||
| enable_sensitive_data: bool = False, | ||
|
|
@@ -858,6 +877,8 @@ async def configure_azure_monitor( | |
|
|
||
| Raises: | ||
| ImportError: If azure-monitor-opentelemetry-exporter is not installed. | ||
| ValueError: If the Application Insights connection does not contain the expected | ||
| project-scoped ARM resource ID. | ||
| """ | ||
| from agent_framework.observability import ( | ||
| OBSERVABILITY_SETTINGS, | ||
|
|
@@ -897,6 +918,8 @@ async def configure_azure_monitor( | |
| "Install it with: pip install azure-monitor-opentelemetry" | ||
| ) from exc | ||
|
|
||
| self._foundry_project_arm_id = await self._get_foundry_project_arm_id() | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Could this lookup degrade gracefully instead of aborting setup? The connection string has already been retrieved successfully, but the separate connections API can return no
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. so does calling this here, mean that only if you setup using this helper method this get's applied and things work? that's either something we need to document extra well, including the in the current samples, or something we need to rethink... (to be clear, I'm not opposed perse) |
||
|
|
||
| if "resource" not in kwargs: | ||
| kwargs["resource"] = create_resource() | ||
|
|
||
|
|
@@ -951,6 +974,18 @@ class FoundryAgent( # type: ignore[misc] | |
| ) | ||
| """ | ||
|
|
||
| @override | ||
| def _get_additional_otel_agent_attributes(self) -> Mapping[str, Any]: | ||
| """Return Foundry attributes required to discover the agent trace.""" | ||
| if self._foundry_project_arm_id: | ||
| return {_FOUNDRY_PROJECT_ARM_ID_ATTRIBUTE: self._foundry_project_arm_id} | ||
| return {} | ||
|
|
||
| @override | ||
| def _should_capture_agent_response_id(self) -> bool: | ||
| """Keep the response ID on the client agent span for Foundry trace discovery.""" | ||
| return True | ||
|
|
||
| def __init__( | ||
| self, | ||
| *, | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could the per-invocation response bookkeeping use one
_InnerResponseTelemetryStatecontext variable rather than adding a third independently managed value?INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS,INNER_CAPTURED_RESPONSE_ID, andINNER_ACCUMULATED_USAGEnow have to be initialized and reset together in both execution paths, so adding another captured value or missing one operation can leak stale state into a nested run. Keeping those fields in one object would make their lifecycle atomic without changing the emitted spans.