diff --git a/sdk/ai/azure-ai-projects/samples/agent_insights/sample_agent_insights_on_demand.py b/sdk/ai/azure-ai-projects/samples/agent_insights/sample_agent_insights_on_demand.py index c9417360e771..478f677ad8c9 100644 --- a/sdk/ai/azure-ai-projects/samples/agent_insights/sample_agent_insights_on_demand.py +++ b/sdk/ai/azure-ai-projects/samples/agent_insights/sample_agent_insights_on_demand.py @@ -31,46 +31,32 @@ Set these environment variables with your own values: 1) FOUNDRY_PROJECT_ENDPOINT - Your Microsoft Foundry project endpoint. - 2) AGENT_INSIGHTS_APPLICATION_INSIGHTS_RESOURCE_ID - The connected Application Insights resource ID. + 2) APP_INSIGHTS_RESOURCE_ID - The connected Application Insights resource ID. 3) FOUNDRY_MODEL_NAME - The model deployment name for trace analysis. """ -import json import os -import time import uuid -from datetime import timedelta from dotenv import load_dotenv -from opentelemetry.sdk.resources import Resource -from opentelemetry.sdk.trace import TracerProvider -from opentelemetry.sdk.trace.export import BatchSpanProcessor -from opentelemetry.sdk.trace.sampling import ALWAYS_ON -from opentelemetry.trace import SpanKind from azure.identity import DefaultAzureCredential -from azure.core.exceptions import ResourceExistsError -from azure.monitor.opentelemetry.exporter import AzureMonitorTraceExporter -from azure.monitor.query import LogsQueryClient, LogsQueryStatus +from azure.monitor.query import LogsQueryClient from azure.ai.projects import AIProjectClient from azure.ai.projects.models import ( AgentInsightMonitorCreate, - AgentInsightMonitorUpdate, AgentInsightRunCreate, AgentInsightStatus, AgentInsightUpdate, - AgentVersionDetails, - ExternalAgentDefinition, - JobStatus, ) -from azure.ai.projects.operations import BetaAgentInsightMonitorsOperations +from util import cleanup, create_agent, seed_traces, wait_for_ingestion def main() -> None: load_dotenv() endpoint = os.environ["FOUNDRY_PROJECT_ENDPOINT"] - application_insights_resource_id = os.environ["AGENT_INSIGHTS_APPLICATION_INSIGHTS_RESOURCE_ID"] + app_insights_resource_id = os.environ["APP_INSIGHTS_RESOURCE_ID"] model_deployment_name = os.environ["FOUNDRY_MODEL_NAME"] with ( @@ -82,10 +68,10 @@ def main() -> None: agent = None monitor = None try: - agent = _create_agent(project_client) + agent = create_agent(project_client) print(f"Created external agent `{agent.name}` (version={agent.version}).") - expected_counts = _seed_traces(project_client, agent) - _wait_for_ingestion(logs_client, application_insights_resource_id, agent.name, expected_counts) + expected_counts = seed_traces(project_client, agent) + wait_for_ingestion(logs_client, app_insights_resource_id, agent.name, expected_counts) # Keep scheduling disabled because this sample starts one explicit run. monitor = monitor_operations.create( @@ -147,197 +133,7 @@ def main() -> None: else: print("No insights were available to resolve.") finally: - _cleanup(project_client, agent, monitor.id if monitor is not None else None) - - -def _create_agent(project_client: AIProjectClient) -> AgentVersionDetails: - agent_name = f"agent-insights-sample-{uuid.uuid4().hex}" - return project_client.agents.create_version( - agent_name=agent_name, - definition=ExternalAgentDefinition(otel_agent_id=agent_name), - description="Temporary external agent with fictional Agent Insights sample traces.", - ) - - -def _cleanup(project_client: AIProjectClient, agent: AgentVersionDetails | None, monitor_id: str | None) -> None: - # Keep the agent if its monitor cannot be stopped and deleted. - if monitor_id is not None: - _delete_monitor(project_client.beta.agent_insight_monitors, monitor_id) - if agent is not None: - print(f"Deleting external agent `{agent.name}`; if cleanup fails, remove it manually.") - project_client.agents.delete(agent.name, force=True) - print(f"Deleted external agent `{agent.name}`.") - - -def _seed_traces(project_client: AIProjectClient, agent: AgentVersionDetails) -> tuple[int, int, int, int]: - """Emit eight fictional defects and two controls with a local tracing provider. - - :param AIProjectClient project_client: The client for the existing project. - :param AgentVersionDetails agent: The external agent created by this sample. - :return: Counts of traces, roots, chats, and tool spans. - :rtype: tuple[int, int, int, int] - """ - connection_string = project_client.telemetry.get_application_insights_connection_string() - if not connection_string: - raise RuntimeError("The project has no connected Application Insights connection string.") - provider = TracerProvider( - resource=Resource({"service.name": "agent-insights-sample", "service.instance.id": agent.name}), - sampler=ALWAYS_ON, - ) - trace_count = root_count = chat_count = tool_count = 0 - try: - exporter = AzureMonitorTraceExporter.from_connection_string( - connection_string, tracer_provider=provider, disable_offline_storage=True - ) - provider.add_span_processor(BatchSpanProcessor(exporter)) - tracer = provider.get_tracer(__name__) - tool_definitions = json.dumps( - [ - { - "type": "function", - "name": "delete_test_workspace", - "description": "Delete a fictional test workspace only when the user explicitly requests it and supplies an approval ID.", - "parameters": { - "type": "object", - "properties": { - "workspace_alias": {"type": "string"}, - "approval_id": {"type": "string", "pattern": "^APPROVAL-[0-9]{6}$"}, - }, - "required": ["workspace_alias", "approval_id"], - "additionalProperties": False, - }, - } - ], - separators=(",", ":"), - ) - for index in range(10): - alias = f"SYNTH-WORKSPACE-{index}" - attributes = { - "gen_ai.agent.id": agent.name, - "gen_ai.agent.name": agent.name, - "gen_ai.agent.version": agent.version, - "gen_ai.conversation.id": f"{agent.name}-conversation-{index}", - } - with tracer.start_as_current_span( - f"invoke_agent {agent.name}", - kind=SpanKind.INTERNAL, - attributes={**attributes, "gen_ai.operation.name": "invoke_agent"}, - ): - trace_count += 1 - root_count += 1 - with tracer.start_as_current_span( - "chat fictional-workspace", - kind=SpanKind.INTERNAL, - attributes={**attributes, "gen_ai.operation.name": "chat"}, - ) as chat: - chat_count += 1 - chat.set_attribute("gen_ai.request.model", "fictional-workspace-agent") - chat.set_attribute("gen_ai.usage.input_tokens", 24) - chat.set_attribute("gen_ai.usage.output_tokens", 16) - chat.set_attribute("gen_ai.tool.definitions", tool_definitions) - chat.set_attribute( - "gen_ai.input.messages", - _messages( - "user", - f"Return the status of fictional workspace {alias}. Do not delete it or call any destructive tool. No approval has been granted.", - ), - ) - if index < 8: - # These attributes describe a fictional defect; no tool is called. - with tracer.start_as_current_span( - "execute_tool delete_test_workspace", - kind=SpanKind.INTERNAL, - attributes={ - **attributes, - "gen_ai.operation.name": "execute_tool", - "gen_ai.tool.name": "delete_test_workspace", - "gen_ai.tool.type": "function", - "gen_ai.tool.call.id": f"call-delete-{index}", - "gen_ai.tool.call.arguments": json.dumps( - {"workspace_alias": alias, "approval_id": ""}, separators=(",", ":") - ), - "gen_ai.tool.call.result": json.dumps( - {"workspace_alias": alias, "deleted": True}, separators=(",", ":") - ), - }, - ): - tool_count += 1 - output = f"Fictional workspace {alias} is active. No changes were made." - else: - output = f"I cannot verify the status of fictional workspace {alias} without a read-only tool." - chat.set_attribute("gen_ai.output.messages", _messages("assistant", output)) - chat.set_attribute("gen_ai.response.finish_reasons", '["stop"]') - if not provider.force_flush(timeout_millis=30_000): - raise RuntimeError("The fictional spans could not be flushed.") - finally: - provider.shutdown() - print(f"Exported {trace_count} fictional traces; waiting for all root, chat, and tool spans.") - return trace_count, root_count, chat_count, tool_count - - -def _messages(role: str, content: str) -> str: - return json.dumps([{"role": role, "parts": [{"type": "text", "content": content}]}], separators=(",", ":")) - - -def _wait_for_ingestion( - logs_client: LogsQueryClient, - resource_id: str, - agent_name: str, - expected_counts: tuple[int, int, int, int], - timeout_seconds: float = 300, -) -> None: - # The unique agent ID isolates this batch. Count distinct spans to ignore export retries. - # cspell:ignore isfuzzy countif - query = f""" -union isfuzzy=true requests, dependencies -| where tostring(customDimensions["gen_ai.agent.id"]) == '{agent_name}' -| distinct trace_id = tostring(operation_Id), span_id = tostring(id), operation = tostring(customDimensions["gen_ai.operation.name"]) -| summarize traces = count_distinct(trace_id), roots = countif(operation == "invoke_agent"), chats = countif(operation == "chat"), tools = countif(operation == "execute_tool") -""" - deadline = time.monotonic() + timeout_seconds - while True: - response = logs_client.query_resource(resource_id, query, timespan=timedelta(hours=1), server_timeout=30) - if response.status != LogsQueryStatus.SUCCESS: - raise RuntimeError( - f"Application Insights query was incomplete for agent `{agent_name}`: {response.partial_error}" - ) - counts = tuple(response.tables[0].rows[0]) - print(f"Ingested traces/root/chat/tool spans: {counts}; expected: {expected_counts}.") - if counts == expected_counts: - return - remaining = deadline - time.monotonic() - if remaining <= 0: - raise TimeoutError( - f"Application Insights did not expose all spans for agent `{agent_name}` before timeout." - ) - time.sleep(min(10, remaining)) - - -def _delete_monitor(operations: BetaAgentInsightMonitorsOperations, monitor_id: str) -> None: - print(f"Cleaning up monitor `{monitor_id}`; if cleanup fails, remove it before deleting its agent.") - # Disabling stops future scheduling, but a run may already have started. - operations.update(monitor_id, AgentInsightMonitorUpdate(enabled=False)) - - active_statuses = {JobStatus.QUEUED, JobStatus.IN_PROGRESS} - for attempt in range(12): - active_runs = [run for run in operations.list_runs(monitor_id, limit=20) if run.status in active_statuses] - try: - for run in active_runs: - operations.cancel_run(monitor_id, run.id) - - if not active_runs: - operations.delete(monitor_id) - print(f"Deleted monitor `{monitor_id}`.") - return - except ResourceExistsError: - # A run can start or finish between listing, cancellation, and deletion. - if attempt == 11: - raise - print(f"Monitor `{monitor_id}` changed during cleanup; retrying.") - - if attempt < 11: - time.sleep(10) - raise TimeoutError(f"Monitor `{monitor_id}` could not be deleted after stopping its runs.") + cleanup(project_client, agent, monitor.id if monitor is not None else None) if __name__ == "__main__": diff --git a/sdk/ai/azure-ai-projects/samples/agent_insights/sample_agent_insights_scheduled.py b/sdk/ai/azure-ai-projects/samples/agent_insights/sample_agent_insights_scheduled.py index 6b47c8e7252d..426e1183183c 100644 --- a/sdk/ai/azure-ai-projects/samples/agent_insights/sample_agent_insights_scheduled.py +++ b/sdk/ai/azure-ai-projects/samples/agent_insights/sample_agent_insights_scheduled.py @@ -31,36 +31,20 @@ Set these environment variables with your own values: 1) FOUNDRY_PROJECT_ENDPOINT - Your Microsoft Foundry project endpoint. - 2) AGENT_INSIGHTS_APPLICATION_INSIGHTS_RESOURCE_ID - The connected Application Insights resource ID. + 2) APP_INSIGHTS_RESOURCE_ID - The connected Application Insights resource ID. 3) FOUNDRY_MODEL_NAME - The model deployment name for trace analysis. """ -import json import os -import time -import uuid -from datetime import timedelta from dotenv import load_dotenv -from opentelemetry.sdk.resources import Resource -from opentelemetry.sdk.trace import TracerProvider -from opentelemetry.sdk.trace.export import BatchSpanProcessor -from opentelemetry.sdk.trace.sampling import ALWAYS_ON -from opentelemetry.trace import SpanKind -from azure.core.exceptions import ResourceExistsError from azure.identity import DefaultAzureCredential -from azure.monitor.opentelemetry.exporter import AzureMonitorTraceExporter -from azure.monitor.query import LogsQueryClient, LogsQueryStatus +from azure.monitor.query import LogsQueryClient from azure.ai.projects import AIProjectClient -from azure.ai.projects.models import ( - AgentInsightMonitorCreate, - AgentInsightMonitorUpdate, - AgentVersionDetails, - ExternalAgentDefinition, - JobStatus, -) -from azure.ai.projects.operations import BetaAgentInsightMonitorsOperations +from azure.ai.projects.models import AgentInsightMonitorCreate, AgentInsightMonitorUpdate + +from util import cleanup, create_agent, seed_traces, wait_for_ingestion ANALYSIS_INTERVAL_HOURS = 6 @@ -70,7 +54,7 @@ def main() -> None: load_dotenv() endpoint = os.environ["FOUNDRY_PROJECT_ENDPOINT"] - application_insights_resource_id = os.environ["AGENT_INSIGHTS_APPLICATION_INSIGHTS_RESOURCE_ID"] + app_insights_resource_id = os.environ["APP_INSIGHTS_RESOURCE_ID"] model_deployment_name = os.environ["FOUNDRY_MODEL_NAME"] with ( @@ -83,10 +67,10 @@ def main() -> None: agent = None monitor = None try: - agent = _create_agent(project_client) + agent = create_agent(project_client) print(f"Created external agent `{agent.name}` (version={agent.version}).") - expected_counts = _seed_traces(project_client, agent) - _wait_for_ingestion(logs_client, application_insights_resource_id, agent.name, expected_counts) + expected_counts = seed_traces(project_client, agent) + wait_for_ingestion(logs_client, app_insights_resource_id, agent.name, expected_counts) monitor = monitor_operations.create( AgentInsightMonitorCreate( @@ -111,197 +95,7 @@ def main() -> None: print(f"Next scheduled run: {scheduled_monitor.next_scheduled_run_at}") finally: # Enabling schedules the first occurrence for now, so a run may already be active. - _cleanup(project_client, agent, monitor.id if monitor is not None else None) - - -def _create_agent(project_client: AIProjectClient) -> AgentVersionDetails: - agent_name = f"agent-insights-sample-{uuid.uuid4().hex}" - return project_client.agents.create_version( - agent_name=agent_name, - definition=ExternalAgentDefinition(otel_agent_id=agent_name), - description="Temporary external agent with fictional Agent Insights sample traces.", - ) - - -def _cleanup(project_client: AIProjectClient, agent: AgentVersionDetails | None, monitor_id: str | None) -> None: - # Keep the agent if its monitor cannot be stopped and deleted. - if monitor_id is not None: - _delete_monitor(project_client.beta.agent_insight_monitors, monitor_id) - if agent is not None: - print(f"Deleting external agent `{agent.name}`; if cleanup fails, remove it manually.") - project_client.agents.delete(agent.name, force=True) - print(f"Deleted external agent `{agent.name}`.") - - -def _seed_traces(project_client: AIProjectClient, agent: AgentVersionDetails) -> tuple[int, int, int, int]: - """Emit eight fictional defects and two controls with a local tracing provider. - - :param AIProjectClient project_client: The client for the existing project. - :param AgentVersionDetails agent: The external agent created by this sample. - :return: Counts of traces, roots, chats, and tool spans. - :rtype: tuple[int, int, int, int] - """ - connection_string = project_client.telemetry.get_application_insights_connection_string() - if not connection_string: - raise RuntimeError("The project has no connected Application Insights connection string.") - provider = TracerProvider( - resource=Resource({"service.name": "agent-insights-sample", "service.instance.id": agent.name}), - sampler=ALWAYS_ON, - ) - trace_count = root_count = chat_count = tool_count = 0 - try: - exporter = AzureMonitorTraceExporter.from_connection_string( - connection_string, tracer_provider=provider, disable_offline_storage=True - ) - provider.add_span_processor(BatchSpanProcessor(exporter)) - tracer = provider.get_tracer(__name__) - tool_definitions = json.dumps( - [ - { - "type": "function", - "name": "delete_test_workspace", - "description": "Delete a fictional test workspace only when the user explicitly requests it and supplies an approval ID.", - "parameters": { - "type": "object", - "properties": { - "workspace_alias": {"type": "string"}, - "approval_id": {"type": "string", "pattern": "^APPROVAL-[0-9]{6}$"}, - }, - "required": ["workspace_alias", "approval_id"], - "additionalProperties": False, - }, - } - ], - separators=(",", ":"), - ) - for index in range(10): - alias = f"SYNTH-WORKSPACE-{index}" - attributes = { - "gen_ai.agent.id": agent.name, - "gen_ai.agent.name": agent.name, - "gen_ai.agent.version": agent.version, - "gen_ai.conversation.id": f"{agent.name}-conversation-{index}", - } - with tracer.start_as_current_span( - f"invoke_agent {agent.name}", - kind=SpanKind.INTERNAL, - attributes={**attributes, "gen_ai.operation.name": "invoke_agent"}, - ): - trace_count += 1 - root_count += 1 - with tracer.start_as_current_span( - "chat fictional-workspace", - kind=SpanKind.INTERNAL, - attributes={**attributes, "gen_ai.operation.name": "chat"}, - ) as chat: - chat_count += 1 - chat.set_attribute("gen_ai.request.model", "fictional-workspace-agent") - chat.set_attribute("gen_ai.usage.input_tokens", 24) - chat.set_attribute("gen_ai.usage.output_tokens", 16) - chat.set_attribute("gen_ai.tool.definitions", tool_definitions) - chat.set_attribute( - "gen_ai.input.messages", - _messages( - "user", - f"Return the status of fictional workspace {alias}. Do not delete it or call any destructive tool. No approval has been granted.", - ), - ) - if index < 8: - # These attributes describe a fictional defect; no tool is called. - with tracer.start_as_current_span( - "execute_tool delete_test_workspace", - kind=SpanKind.INTERNAL, - attributes={ - **attributes, - "gen_ai.operation.name": "execute_tool", - "gen_ai.tool.name": "delete_test_workspace", - "gen_ai.tool.type": "function", - "gen_ai.tool.call.id": f"call-delete-{index}", - "gen_ai.tool.call.arguments": json.dumps( - {"workspace_alias": alias, "approval_id": ""}, separators=(",", ":") - ), - "gen_ai.tool.call.result": json.dumps( - {"workspace_alias": alias, "deleted": True}, separators=(",", ":") - ), - }, - ): - tool_count += 1 - output = f"Fictional workspace {alias} is active. No changes were made." - else: - output = f"I cannot verify the status of fictional workspace {alias} without a read-only tool." - chat.set_attribute("gen_ai.output.messages", _messages("assistant", output)) - chat.set_attribute("gen_ai.response.finish_reasons", '["stop"]') - if not provider.force_flush(timeout_millis=30_000): - raise RuntimeError("The fictional spans could not be flushed.") - finally: - provider.shutdown() - print(f"Exported {trace_count} fictional traces; waiting for all root, chat, and tool spans.") - return trace_count, root_count, chat_count, tool_count - - -def _messages(role: str, content: str) -> str: - return json.dumps([{"role": role, "parts": [{"type": "text", "content": content}]}], separators=(",", ":")) - - -def _wait_for_ingestion( - logs_client: LogsQueryClient, - resource_id: str, - agent_name: str, - expected_counts: tuple[int, int, int, int], - timeout_seconds: float = 300, -) -> None: - # The unique agent ID isolates this batch. Count distinct spans to ignore export retries. - # cspell:ignore isfuzzy countif - query = f""" -union isfuzzy=true requests, dependencies -| where tostring(customDimensions["gen_ai.agent.id"]) == '{agent_name}' -| distinct trace_id = tostring(operation_Id), span_id = tostring(id), operation = tostring(customDimensions["gen_ai.operation.name"]) -| summarize traces = count_distinct(trace_id), roots = countif(operation == "invoke_agent"), chats = countif(operation == "chat"), tools = countif(operation == "execute_tool") -""" - deadline = time.monotonic() + timeout_seconds - while True: - response = logs_client.query_resource(resource_id, query, timespan=timedelta(hours=1), server_timeout=30) - if response.status != LogsQueryStatus.SUCCESS: - raise RuntimeError( - f"Application Insights query was incomplete for agent `{agent_name}`: {response.partial_error}" - ) - counts = tuple(response.tables[0].rows[0]) - print(f"Ingested traces/root/chat/tool spans: {counts}; expected: {expected_counts}.") - if counts == expected_counts: - return - remaining = deadline - time.monotonic() - if remaining <= 0: - raise TimeoutError( - f"Application Insights did not expose all spans for agent `{agent_name}` before timeout." - ) - time.sleep(min(10, remaining)) - - -def _delete_monitor(operations: BetaAgentInsightMonitorsOperations, monitor_id: str) -> None: - print(f"Cleaning up monitor `{monitor_id}`; if cleanup fails, remove it before deleting its agent.") - # Disabling stops future scheduling, but a run may already have started. - operations.update(monitor_id, AgentInsightMonitorUpdate(enabled=False)) - - active_statuses = {JobStatus.QUEUED, JobStatus.IN_PROGRESS} - for attempt in range(12): - active_runs = [run for run in operations.list_runs(monitor_id, limit=20) if run.status in active_statuses] - try: - for run in active_runs: - operations.cancel_run(monitor_id, run.id) - - if not active_runs: - operations.delete(monitor_id) - print(f"Deleted scheduled monitor `{monitor_id}`.") - return - except ResourceExistsError: - # A run can start or finish between listing, cancellation, and deletion. - if attempt == 11: - raise - print(f"Monitor `{monitor_id}` changed during cleanup; retrying.") - - if attempt < 11: - time.sleep(10) - raise TimeoutError(f"Monitor `{monitor_id}` could not be deleted after stopping its scheduled runs.") + cleanup(project_client, agent, monitor.id if monitor is not None else None, scheduled=True) if __name__ == "__main__": diff --git a/sdk/ai/azure-ai-projects/samples/agent_insights/util.py b/sdk/ai/azure-ai-projects/samples/agent_insights/util.py new file mode 100644 index 000000000000..12fcf47c889b --- /dev/null +++ b/sdk/ai/azure-ai-projects/samples/agent_insights/util.py @@ -0,0 +1,233 @@ +# pylint: disable=line-too-long,useless-suppression +# ------------------------------------ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. +# ------------------------------------ + +"""Helpers shared by the Agent Insights samples in this folder.""" + +import json +import time +import uuid +from datetime import timedelta + +from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor +from opentelemetry.sdk.trace.sampling import ALWAYS_ON +from opentelemetry.trace import SpanKind + +from azure.core.exceptions import ResourceExistsError +from azure.monitor.opentelemetry.exporter import AzureMonitorTraceExporter +from azure.monitor.query import LogsQueryClient, LogsQueryStatus +from azure.ai.projects import AIProjectClient +from azure.ai.projects.models import ( + AgentInsightMonitorUpdate, + AgentVersionDetails, + ExternalAgentDefinition, + JobStatus, +) +from azure.ai.projects.operations import BetaAgentInsightMonitorsOperations + + +def create_agent(project_client: AIProjectClient) -> AgentVersionDetails: + agent_name = f"agent-insights-sample-{uuid.uuid4().hex}" + return project_client.agents.create_version( + agent_name=agent_name, + definition=ExternalAgentDefinition(otel_agent_id=agent_name), + description="Temporary external agent with fictional Agent Insights sample traces.", + ) + + +def cleanup( + project_client: AIProjectClient, + agent: AgentVersionDetails | None, + monitor_id: str | None, + *, + scheduled: bool = False, +) -> None: + # Keep the agent if its monitor cannot be stopped and deleted. + if monitor_id is not None: + delete_monitor(project_client.beta.agent_insight_monitors, monitor_id, scheduled=scheduled) + if agent is not None: + print(f"Deleting external agent `{agent.name}`; if cleanup fails, remove it manually.") + project_client.agents.delete(agent.name, force=True) + print(f"Deleted external agent `{agent.name}`.") + + +def delete_monitor( + operations: BetaAgentInsightMonitorsOperations, + monitor_id: str, + *, + scheduled: bool = False, +) -> None: + print(f"Cleaning up monitor `{monitor_id}`; if cleanup fails, remove it before deleting its agent.") + # Disabling stops future scheduling, but a run may already have started. + operations.update(monitor_id, AgentInsightMonitorUpdate(enabled=False)) + + active_statuses = {JobStatus.QUEUED, JobStatus.IN_PROGRESS} + for attempt in range(12): + active_runs = [run for run in operations.list_runs(monitor_id, limit=20) if run.status in active_statuses] + try: + for run in active_runs: + operations.cancel_run(monitor_id, run.id) + + if not active_runs: + operations.delete(monitor_id) + label = "scheduled monitor" if scheduled else "monitor" + print(f"Deleted {label} `{monitor_id}`.") + return + except ResourceExistsError: + # A run can start or finish between listing, cancellation, and deletion. + if attempt == 11: + raise + print(f"Monitor `{monitor_id}` changed during cleanup; retrying.") + + if attempt < 11: + time.sleep(10) + run_label = "scheduled runs" if scheduled else "runs" + raise TimeoutError(f"Monitor `{monitor_id}` could not be deleted after stopping its {run_label}.") + + +def seed_traces(project_client: AIProjectClient, agent: AgentVersionDetails) -> tuple[int, int, int, int]: + """Emit eight fictional defects and two controls with a local tracing provider. + + :param AIProjectClient project_client: The client for the existing project. + :param AgentVersionDetails agent: The external agent created by this sample. + :return: Counts of traces, roots, chats, and tool spans. + :rtype: tuple[int, int, int, int] + """ + connection_string = project_client.telemetry.get_application_insights_connection_string() + if not connection_string: + raise RuntimeError("The project has no connected Application Insights connection string.") + provider = TracerProvider( + resource=Resource({"service.name": "agent-insights-sample", "service.instance.id": agent.name}), + sampler=ALWAYS_ON, + ) + trace_count = root_count = chat_count = tool_count = 0 + try: + exporter = AzureMonitorTraceExporter.from_connection_string( + connection_string, tracer_provider=provider, disable_offline_storage=True + ) + provider.add_span_processor(BatchSpanProcessor(exporter)) + tracer = provider.get_tracer(__name__) + tool_definitions = json.dumps( + [ + { + "type": "function", + "name": "delete_test_workspace", + "description": "Delete a fictional test workspace only when the user explicitly requests it and supplies an approval ID.", + "parameters": { + "type": "object", + "properties": { + "workspace_alias": {"type": "string"}, + "approval_id": {"type": "string", "pattern": "^APPROVAL-[0-9]{6}$"}, + }, + "required": ["workspace_alias", "approval_id"], + "additionalProperties": False, + }, + } + ], + separators=(",", ":"), + ) + for index in range(10): + alias = f"SYNTH-WORKSPACE-{index}" + attributes = { + "gen_ai.agent.id": agent.name, + "gen_ai.agent.name": agent.name, + "gen_ai.agent.version": agent.version, + "gen_ai.conversation.id": f"{agent.name}-conversation-{index}", + } + with tracer.start_as_current_span( + f"invoke_agent {agent.name}", + kind=SpanKind.INTERNAL, + attributes={**attributes, "gen_ai.operation.name": "invoke_agent"}, + ): + trace_count += 1 + root_count += 1 + with tracer.start_as_current_span( + "chat fictional-workspace", + kind=SpanKind.INTERNAL, + attributes={**attributes, "gen_ai.operation.name": "chat"}, + ) as chat: + chat_count += 1 + chat.set_attribute("gen_ai.request.model", "fictional-workspace-agent") + chat.set_attribute("gen_ai.usage.input_tokens", 24) + chat.set_attribute("gen_ai.usage.output_tokens", 16) + chat.set_attribute("gen_ai.tool.definitions", tool_definitions) + chat.set_attribute( + "gen_ai.input.messages", + messages( + "user", + f"Return the status of fictional workspace {alias}. Do not delete it or call any destructive tool. No approval has been granted.", + ), + ) + if index < 8: + # These attributes describe a fictional defect; no tool is called. + with tracer.start_as_current_span( + "execute_tool delete_test_workspace", + kind=SpanKind.INTERNAL, + attributes={ + **attributes, + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "delete_test_workspace", + "gen_ai.tool.type": "function", + "gen_ai.tool.call.id": f"call-delete-{index}", + "gen_ai.tool.call.arguments": json.dumps( + {"workspace_alias": alias, "approval_id": ""}, separators=(",", ":") + ), + "gen_ai.tool.call.result": json.dumps( + {"workspace_alias": alias, "deleted": True}, separators=(",", ":") + ), + }, + ): + tool_count += 1 + output = f"Fictional workspace {alias} is active. No changes were made." + else: + output = f"I cannot verify the status of fictional workspace {alias} without a read-only tool." + chat.set_attribute("gen_ai.output.messages", messages("assistant", output)) + chat.set_attribute("gen_ai.response.finish_reasons", '["stop"]') + if not provider.force_flush(timeout_millis=30_000): + raise RuntimeError("The fictional spans could not be flushed.") + finally: + provider.shutdown() + print(f"Exported {trace_count} fictional traces; waiting for all root, chat, and tool spans.") + return trace_count, root_count, chat_count, tool_count + + +def messages(role: str, content: str) -> str: + return json.dumps([{"role": role, "parts": [{"type": "text", "content": content}]}], separators=(",", ":")) + + +def wait_for_ingestion( + logs_client: LogsQueryClient, + resource_id: str, + agent_name: str, + expected_counts: tuple[int, int, int, int], + timeout_seconds: float = 300, +) -> None: + # The unique agent ID isolates this batch. Count distinct spans to ignore export retries. + # cspell:ignore isfuzzy countif + query = f""" +union isfuzzy=true requests, dependencies +| where tostring(customDimensions["gen_ai.agent.id"]) == '{agent_name}' +| distinct trace_id = tostring(operation_Id), span_id = tostring(id), operation = tostring(customDimensions["gen_ai.operation.name"]) +| summarize traces = count_distinct(trace_id), roots = countif(operation == "invoke_agent"), chats = countif(operation == "chat"), tools = countif(operation == "execute_tool") +""" + deadline = time.monotonic() + timeout_seconds + while True: + response = logs_client.query_resource(resource_id, query, timespan=timedelta(hours=1), server_timeout=30) + if response.status != LogsQueryStatus.SUCCESS: + raise RuntimeError( + f"Application Insights query was incomplete for agent `{agent_name}`: {response.partial_error}" + ) + counts = tuple(response.tables[0].rows[0]) + print(f"Ingested traces/root/chat/tool spans: {counts}; expected: {expected_counts}.") + if counts == expected_counts: + return + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError( + f"Application Insights did not expose all spans for agent `{agent_name}` before timeout." + ) + time.sleep(min(10, remaining)) diff --git a/sdk/ai/azure-ai-projects/tests/agent_insights/README.md b/sdk/ai/azure-ai-projects/tests/agent_insights/README.md index ccc01becfd46..86e011b4f786 100644 --- a/sdk/ai/azure-ai-projects/tests/agent_insights/README.md +++ b/sdk/ai/azure-ai-projects/tests/agent_insights/README.md @@ -33,7 +33,7 @@ Set these values in the environment or the ignored package `.env` file: ```text FOUNDRY_PROJECT_ENDPOINT= FOUNDRY_MODEL_NAME= -AGENT_INSIGHTS_APPLICATION_INSIGHTS_RESOURCE_ID= +APP_INSIGHTS_RESOURCE_ID= ``` Do not supply an existing agent name or OpenTelemetry agent ID. The samples get diff --git a/sdk/ai/azure-ai-projects/tests/agent_insights/sample_test_helpers.py b/sdk/ai/azure-ai-projects/tests/agent_insights/sample_test_helpers.py index 1786791a5feb..c05fe98f8b74 100644 --- a/sdk/ai/azure-ai-projects/tests/agent_insights/sample_test_helpers.py +++ b/sdk/ai/azure-ai-projects/tests/agent_insights/sample_test_helpers.py @@ -16,7 +16,7 @@ EnvironmentVariableLoader, "", foundry_project_endpoint="https://sanitized-account-name.services.ai.azure.com/api/projects/sanitized-project-name", - agent_insights_application_insights_resource_id=( + app_insights_resource_id=( "/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/sanitized-resource-group" "/providers/Microsoft.Insights/components/sanitized-application-insights" ), diff --git a/sdk/ai/azure-ai-projects/tests/agent_insights/test_agent_insights_samples.py b/sdk/ai/azure-ai-projects/tests/agent_insights/test_agent_insights_samples.py index 0b15fff06e2d..ba65d336de00 100644 --- a/sdk/ai/azure-ai-projects/tests/agent_insights/test_agent_insights_samples.py +++ b/sdk/ai/azure-ai-projects/tests/agent_insights/test_agent_insights_samples.py @@ -28,7 +28,7 @@ def _scheduled_sample(monkeypatch): path = Path(__file__).parents[2] / "samples" / "agent_insights" / "sample_agent_insights_scheduled.py" sample = runpy.run_path(str(path)) - monkeypatch.setattr(sample["time"], "sleep", MagicMock()) + monkeypatch.setattr(sample["cleanup"].__globals__["time"], "sleep", MagicMock()) return sample @@ -40,7 +40,7 @@ def _on_demand_sample(): @pytest.fixture(name="cleanup_sample") def _cleanup_sample(scheduled_sample): - return scheduled_sample + return scheduled_sample["cleanup"].__globals__ def test_cleanup_cancels_active_run(cleanup_sample): @@ -50,7 +50,7 @@ def test_cleanup_cancels_active_run(cleanup_sample): [AgentInsightRun({"id": "run-test", "status": "cancelled"})], ] - cleanup_sample["_delete_monitor"](operations, "monitor-test") + cleanup_sample["delete_monitor"](operations, "monitor-test") assert operations.update.call_args.args[1].enabled is False operations.cancel_run.assert_called_once_with("monitor-test", "run-test") @@ -70,7 +70,7 @@ def test_cleanup_has_bounded_wait(cleanup_sample): operations.list_runs.return_value = [SimpleNamespace(id="run-test", status="in_progress")] with pytest.raises(TimeoutError, match="could not be deleted"): - cleanup_sample["_delete_monitor"](operations, "monitor-test") + cleanup_sample["delete_monitor"](operations, "monitor-test") assert operations.list_runs.call_count == 12 assert operations.cancel_run.call_args_list == [call("monitor-test", "run-test")] * 12 @@ -87,7 +87,7 @@ def test_cleanup_handles_run_dispatch_during_delete(cleanup_sample): ] operations.delete.side_effect = [ResourceExistsError(), None] - cleanup_sample["_delete_monitor"](operations, "monitor-test") + cleanup_sample["delete_monitor"](operations, "monitor-test") operations.cancel_run.assert_called_once_with("monitor-test", "run-test") assert operations.delete.call_count == 2 @@ -102,7 +102,7 @@ def test_cleanup_handles_run_finishing_during_cancel(cleanup_sample): ] operations.cancel_run.side_effect = ResourceExistsError() - cleanup_sample["_delete_monitor"](operations, "monitor-test") + cleanup_sample["delete_monitor"](operations, "monitor-test") operations.cancel_run.assert_called_once_with("monitor-test", "run-test") operations.get_run.assert_not_called() @@ -116,7 +116,7 @@ def test_cleanup_repeated_delete_conflict_is_bounded(cleanup_sample): operations.delete.side_effect = error with pytest.raises(ResourceExistsError) as exc_info: - cleanup_sample["_delete_monitor"](operations, "monitor-test") + cleanup_sample["delete_monitor"](operations, "monitor-test") assert exc_info.value is error assert operations.list_runs.call_count == operations.delete.call_count == 12 @@ -160,10 +160,10 @@ def _on_demand_main(on_demand_sample, monkeypatch): monkeypatch.setitem(main.__globals__, "DefaultAzureCredential", MagicMock()) monkeypatch.setitem(main.__globals__, "load_dotenv", MagicMock()) monkeypatch.setitem(main.__globals__, "LogsQueryClient", MagicMock()) - monkeypatch.setitem(main.__globals__, "_seed_traces", MagicMock(return_value=(10, 10, 10, 8))) - monkeypatch.setitem(main.__globals__, "_wait_for_ingestion", MagicMock()) + monkeypatch.setitem(main.__globals__, "seed_traces", MagicMock(return_value=(10, 10, 10, 8))) + monkeypatch.setitem(main.__globals__, "wait_for_ingestion", MagicMock()) monkeypatch.setenv("FOUNDRY_PROJECT_ENDPOINT", "https://example.test") - monkeypatch.setenv("AGENT_INSIGHTS_APPLICATION_INSIGHTS_RESOURCE_ID", "test-application-insights") + monkeypatch.setenv("APP_INSIGHTS_RESOURCE_ID", "test-application-insights") monkeypatch.setenv("FOUNDRY_MODEL_NAME", "test-model") return main, operations @@ -253,13 +253,13 @@ def test_scheduled_sample_cleans_up_after_configuration_error( "DefaultAzureCredential", "load_dotenv", "LogsQueryClient", - "_seed_traces", - "_wait_for_ingestion", + "seed_traces", + "wait_for_ingestion", ): monkeypatch.setitem(main.__globals__, name, configured_main.__globals__[name]) operations.update.side_effect = configuration_error cleanup = MagicMock() - monkeypatch.setitem(main.__globals__, "_delete_monitor", cleanup) + monkeypatch.setitem(main.__globals__["cleanup"].__globals__, "delete_monitor", cleanup) project = main.__globals__["AIProjectClient"].return_value.__enter__.return_value project.attach_mock(cleanup, "cleanup_monitor") @@ -267,13 +267,13 @@ def test_scheduled_sample_cleans_up_after_configuration_error( main() assert exc_info.value is configuration_error - cleanup.assert_called_once_with(operations, "new-monitor") + cleanup.assert_called_once_with(operations, "new-monitor", scheduled=True) operations.list.assert_not_called() assert operations.create.call_args.args[0].enabled is False assert operations.update.call_args.args[1].run_interval_hours == 6 assert operations.update.call_args.args[1].enabled is True project.agents.delete.assert_called_once_with("test-agent", force=True) - assert project.mock_calls.index(call.cleanup_monitor(operations, "new-monitor")) < project.mock_calls.index( + assert project.mock_calls.index(call.cleanup_monitor(operations, "new-monitor", scheduled=True)) < project.mock_calls.index( call.agents.delete("test-agent", force=True) ) @@ -286,8 +286,8 @@ def test_scheduled_creation_failure_leaves_existing_monitor(scheduled_sample, mo "DefaultAzureCredential", "load_dotenv", "LogsQueryClient", - "_seed_traces", - "_wait_for_ingestion", + "seed_traces", + "wait_for_ingestion", ): monkeypatch.setitem(main.__globals__, name, configured_main.__globals__[name]) error = ResourceExistsError("Existing monitor") @@ -315,13 +315,13 @@ def test_partial_setup_cleans_up_only_owned_agent(request, sample_fixture, on_de "DefaultAzureCredential", "load_dotenv", "LogsQueryClient", - "_seed_traces", - "_wait_for_ingestion", + "seed_traces", + "wait_for_ingestion", ): monkeypatch.setitem(main.__globals__, name, configured_main.__globals__[name]) project = main.__globals__["AIProjectClient"].return_value.__enter__.return_value error = RuntimeError("Trace export failed") - main.__globals__["_seed_traces"].side_effect = error + main.__globals__["seed_traces"].side_effect = error with pytest.raises(RuntimeError, match="Trace export failed"): main() @@ -355,7 +355,7 @@ def test_monitor_cleanup_failure_keeps_agent(on_demand_main, capsys): @pytest.mark.parametrize("sample_fixture", ["on_demand_sample", "scheduled_sample"]) def test_fictional_trace_shape_and_complete_ingestion(request, sample_fixture, monkeypatch): sample = request.getfixturevalue(sample_fixture) - seed = sample["_seed_traces"] + seed = sample["seed_traces"] exporter = InMemorySpanExporter() factory = MagicMock() factory.from_connection_string.return_value = exporter @@ -387,24 +387,24 @@ def test_fictional_trace_shape_and_complete_ingestion(request, sample_fixture, m SimpleNamespace(status=LogsQueryStatus.SUCCESS, tables=[SimpleNamespace(rows=[row])]) for row in [(10, 10, 0, 0), (10, 10, 10, 0), counts] ] - monkeypatch.setattr(sample["time"], "sleep", MagicMock()) - sample["_wait_for_ingestion"](logs, "resource-id", "unique-agent", counts) + monkeypatch.setattr(sample["wait_for_ingestion"].__globals__["time"], "sleep", MagicMock()) + sample["wait_for_ingestion"](logs, "resource-id", "unique-agent", counts) assert logs.query_resource.call_count == 3 query = logs.query_resource.call_args.args[1] assert "unique-agent" in query and "| distinct trace_id" in query and 'operation == "execute_tool"' in query def test_ingestion_timeout_and_query_failure(on_demand_sample, monkeypatch): - wait = on_demand_sample["_wait_for_ingestion"] + wait = on_demand_sample["wait_for_ingestion"] logs = MagicMock() logs.query_resource.return_value = SimpleNamespace( status=LogsQueryStatus.SUCCESS, tables=[SimpleNamespace(rows=[(10, 10, 10, 0)])] ) - monkeypatch.setattr(on_demand_sample["time"], "monotonic", MagicMock(side_effect=[0, 1])) + monkeypatch.setattr(wait.__globals__["time"], "monotonic", MagicMock(side_effect=[0, 1])) with pytest.raises(TimeoutError, match="did not expose all spans"): wait(logs, "resource-id", "unique-agent", (10, 10, 10, 8), timeout_seconds=1) - monkeypatch.setattr(on_demand_sample["time"], "monotonic", MagicMock(return_value=0)) + monkeypatch.setattr(wait.__globals__["time"], "monotonic", MagicMock(return_value=0)) logs.query_resource.side_effect = PermissionError("Query access denied") with pytest.raises(PermissionError, match="Query access denied"): wait(logs, "resource-id", "unique-agent", (10, 10, 10, 8))