agent/agent-framework
2
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Python: Telemetry and observability follow-up (#833)

Browse Source 
* updated telemetry work

* updated telemetry

* slight improvement

* updated tests

* fixes for telemetry

* fixes for mypy

* added settings setup to runner to avoid error

* streamline usage

* updated tests

* updated tests

* further refinement

* fix dumped item for otel

* removed enable_workflow_otel

* final fixes

* final fixes

* updated samples

* removed exporters

* fix tests

* fixed last import'

* fixed devui
This commit is contained in:
Eduard van Valkenburg
2025-09-23 08:21:56 +02:00
committed by GitHub
Unverified
parent f93f16a9ad
commit 2576e7a091
52 changed files with 1625 additions and 1586 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/main/agent_framework/__init__.py
+5 -2
View File
@@ -2,11 +2,13 @@
import importlib
import importlib.metadata
from typing import Final
try:
__version__ = importlib.metadata.version(__name__)
_version = importlib.metadata.version(__name__)
except importlib.metadata.PackageNotFoundError:
__version__ = "0.0.0" # Fallback for development mode
_version = "0.0.0" # Fallback for development mode
__version__: Final[str] = _version
from ._agents import * # noqa: F403
from ._clients import * # noqa: F403
@@ -14,6 +16,7 @@ from ._logging import * # noqa: F403
from ._mcp import * # noqa: F403
from ._memory import * # noqa: F403
from ._middleware import * # noqa: F403
from ._telemetry import * # noqa: F403
from ._threads import * # noqa: F403
from ._tools import * # noqa: F403
from ._types import * # noqa: F403
python/packages/main/agent_framework/_agents.py
+2 -2
View File
@@ -28,7 +28,7 @@ from ._types import (
Role,
)
from .exceptions import AgentExecutionException
from .telemetry import use_agent_telemetry
from .observability import use_agent_observability
if sys.version_info >= (3, 11):
from typing import Self # pragma: no cover
@@ -258,7 +258,7 @@ class BaseAgent(AFBaseModel):
@use_agent_middleware
@use_agent_telemetry
@use_agent_observability
class ChatAgent(BaseAgent):
"""A Chat Client Agent."""
python/packages/main/agent_framework/_clients.py
-1
View File
@@ -428,7 +428,6 @@ class BaseChatClient(AFBaseModel, ABC):
tools=self._normalize_tools(tools), # type: ignore
user=user,
additional_properties=additional_properties or {},
**kwargs,
)
prepped_messages = self.prepare_messages(messages)
self._prepare_tool_choice(chat_options=chat_options)
python/packages/main/agent_framework/_telemetry.py
+59
View File
@@ -0,0 +1,59 @@
# Copyright (c) Microsoft. All rights reserved.
import os
from typing import Any, Final
from . import __version__ as version_info
from ._logging import get_logger
logger = get_logger()
__all__ = [
"AGENT_FRAMEWORK_USER_AGENT",
"APP_INFO",
"USER_AGENT_KEY",
"USER_AGENT_TELEMETRY_DISABLED_ENV_VAR",
"prepend_agent_framework_to_user_agent",
]
# Note that if this environment variable does not exist, user agent telemetry is enabled.
USER_AGENT_TELEMETRY_DISABLED_ENV_VAR = "AGENT_FRAMEWORK_USER_AGENT_DISABLED"
IS_TELEMETRY_ENABLED = os.environ.get(USER_AGENT_TELEMETRY_DISABLED_ENV_VAR, "false").lower() not in ["true", "1"]
APP_INFO = (
{
"agent-framework-version": f"python/{version_info}", # type: ignore[has-type]
}
if IS_TELEMETRY_ENABLED
else None
)
USER_AGENT_KEY: Final[str] = "User-Agent"
HTTP_USER_AGENT: Final[str] = "agent-framework-python"
AGENT_FRAMEWORK_USER_AGENT = f"{HTTP_USER_AGENT}/{version_info}" # type: ignore[has-type]
def prepend_agent_framework_to_user_agent(headers: dict[str, Any] | None = None) -> dict[str, Any]:
"""Prepend "agent-framework" to the User-Agent in the headers.
When user agent telemetry is disabled, through the AZURE_TELEMETRY_DISABLED environment variable,
the User-Agent header will not include the agent-framework information, it will be sent back as is,
or as a empty dict when None is passed.
Args:
headers: The existing headers dictionary.
Returns:
A new dict with "User-Agent" set to "agent-framework-python/{version}" if headers is None.
The modified headers dictionary with "agent-framework-python/{version}" prepended to the User-Agent.
"""
if not IS_TELEMETRY_ENABLED:
return headers or {}
if not headers:
return {USER_AGENT_KEY: AGENT_FRAMEWORK_USER_AGENT}
headers[USER_AGENT_KEY] = (
f"{AGENT_FRAMEWORK_USER_AGENT} {headers[USER_AGENT_KEY]}"
if USER_AGENT_KEY in headers
else AGENT_FRAMEWORK_USER_AGENT
)
return headers
python/packages/main/agent_framework/_tools.py
+17 -15
View File
@@ -21,19 +21,19 @@ from typing import (
runtime_checkable,
)
from opentelemetry import metrics
from opentelemetry.metrics import Histogram
from pydantic import AnyUrl, BaseModel, Field, PrivateAttr, ValidationError, create_model, field_validator
from ._logging import get_logger
from ._pydantic import AFBaseModel
from .exceptions import ChatClientInitializationError, ToolException
from .telemetry import (
from .observability import (
OPERATION_DURATION_BUCKET_BOUNDARIES,
OtelAttr,
_capture_exception, # type: ignore
capture_exception, # type: ignore
get_function_span,
get_function_span_attributes,
meter,
get_meter,
)
if TYPE_CHECKING:
@@ -358,6 +358,16 @@ class HostedFileSearchTool(BaseTool):
super().__init__(**args, **kwargs)
def _default_histogram() -> Histogram:
"""Get the default histogram for function invocation duration."""
return get_meter().create_histogram(
name=OtelAttr.MEASUREMENT_FUNCTION_INVOCATION_DURATION,
unit=OtelAttr.DURATION_UNIT,
description="Measures the duration of a function's execution",
explicit_bucket_boundaries_advisory=OPERATION_DURATION_BUCKET_BOUNDARIES,
)
class AIFunction(BaseTool, Generic[ArgsT, ReturnT]):
"""A AITool that is callable as code.
@@ -371,14 +381,7 @@ class AIFunction(BaseTool, Generic[ArgsT, ReturnT]):
func: Callable[..., Awaitable[ReturnT] | ReturnT]
input_model: type[ArgsT]
_invocation_duration_histogram: metrics.Histogram = PrivateAttr(
default_factory=lambda: meter.create_histogram(
name=OtelAttr.MEASUREMENT_FUNCTION_INVOCATION_DURATION,
unit=OtelAttr.DURATION_UNIT,
description="Measures the duration of a function's execution",
explicit_bucket_boundaries_advisory=OPERATION_DURATION_BUCKET_BOUNDARIES,
)
)
_invocation_duration_histogram: Histogram = PrivateAttr(default_factory=_default_histogram)
def __call__(self, *args: Any, **kwargs: Any) -> ReturnT | Awaitable[ReturnT]:
"""Call the wrapped function with the provided arguments."""
@@ -398,7 +401,7 @@ class AIFunction(BaseTool, Generic[ArgsT, ReturnT]):
kwargs: keyword arguments to pass to the function, will not be used if `arguments` is provided.
"""
global OTEL_SETTINGS
from .telemetry import OTEL_SETTINGS, setup_telemetry
from .observability import OTEL_SETTINGS
tool_call_id = kwargs.pop("tool_call_id", None)
if arguments is not None:
@@ -414,7 +417,6 @@ class AIFunction(BaseTool, Generic[ArgsT, ReturnT]):
logger.debug(f"Function result: {result or 'None'}")
return result # type: ignore[reportReturnType]
setup_telemetry()
attributes = get_function_span_attributes(self, tool_call_id=tool_call_id)
if OTEL_SETTINGS.SENSITIVE_DATA_ENABLED: # type: ignore[name-defined]
attributes.update({
@@ -438,7 +440,7 @@ class AIFunction(BaseTool, Generic[ArgsT, ReturnT]):
except Exception as exception:
end_time_stamp = perf_counter()
attributes[OtelAttr.ERROR_TYPE] = type(exception).__name__
_capture_exception(span=span, exception=exception, timestamp=time_ns())
capture_exception(span=span, exception=exception, timestamp=time_ns())
logger.error(f"Function failed. Error: {exception}")
raise
else:
python/packages/main/agent_framework/_workflow/__init__.py
-4
View File
@@ -88,7 +88,6 @@ from ._runner_context import (
)
from ._sequential import SequentialBuilder
from ._shared_state import SharedState
from ._telemetry import EdgeGroupDeliveryStatus, WorkflowTracer, workflow_tracer
from ._validation import (
EdgeDuplicationError,
ExecutorDuplicationError,
@@ -116,7 +115,6 @@ __all__ = [
"Default",
"Edge",
"EdgeDuplicationError",
"EdgeGroupDeliveryStatus",
"Executor",
"ExecutorCompletedEvent",
"ExecutorDuplicationError",
@@ -184,7 +182,6 @@ __all__ = [
"WorkflowRunState",
"WorkflowStartedEvent",
"WorkflowStatusEvent",
"WorkflowTracer",
"WorkflowValidationError",
"WorkflowViz",
"create_edge_runner",
@@ -192,7 +189,6 @@ __all__ = [
"handler",
"intercepts_request",
"validate_workflow_graph",
"workflow_tracer",
]
# Rebuild models to resolve forward references after all imports are complete
python/packages/main/agent_framework/_workflow/__init__.pyi
-4
View File
@@ -84,7 +84,6 @@ from ._runner_context import (
)
from ._sequential import SequentialBuilder
from ._shared_state import SharedState
from ._telemetry import EdgeGroupDeliveryStatus, WorkflowTracer, workflow_tracer
from ._validation import (
EdgeDuplicationError,
ExecutorDuplicationError,
@@ -112,7 +111,6 @@ __all__ = [
"Default",
"Edge",
"EdgeDuplicationError",
"EdgeGroupDeliveryStatus",
"Executor",
"ExecutorCompletedEvent",
"ExecutorDuplicationError",
@@ -180,7 +178,6 @@ __all__ = [
"WorkflowRunState",
"WorkflowStartedEvent",
"WorkflowStatusEvent",
"WorkflowTracer",
"WorkflowValidationError",
"WorkflowViz",
"create_edge_runner",
@@ -188,5 +185,4 @@ __all__ = [
"handler",
"intercepts_request",
"validate_workflow_graph",
"workflow_tracer",
]
python/packages/main/agent_framework/_workflow/_agent.py
+2 -2
View File
@@ -20,9 +20,9 @@ from agent_framework import (
TextContent,
UsageDetails,
)
from agent_framework._pydantic import AFBaseModel
from agent_framework.exceptions import AgentExecutionException
from .._pydantic import AFBaseModel
from ..exceptions import AgentExecutionException
from ._events import (
AgentRunUpdateEvent,
RequestInfoEvent,
python/packages/main/agent_framework/_workflow/_edge.py
+1 -2
View File
@@ -8,8 +8,7 @@ from typing import Any, ClassVar
from pydantic import Field
from agent_framework._pydantic import AFBaseModel
from .._pydantic import AFBaseModel
from ._executor import Executor
logger = logging.getLogger(__name__)
python/packages/main/agent_framework/_workflow/_edge_runner.py
+79 -42
View File
@@ -6,11 +6,11 @@ from abc import ABC, abstractmethod
from collections import defaultdict
from typing import Any
from ..observability import EdgeGroupDeliveryStatus, OtelAttr, create_edge_group_processing_span
from ._edge import Edge, EdgeGroup, FanInEdgeGroup, FanOutEdgeGroup, SingleEdgeGroup, SwitchCaseEdgeGroup
from ._executor import Executor
from ._runner_context import Message, RunnerContext
from ._shared_state import SharedState
from ._telemetry import EdgeGroupDeliveryStatus, workflow_tracer
from ._workflow_context import WorkflowContext
logger = logging.getLogger(__name__)
@@ -90,39 +90,49 @@ class SingleEdgeRunner(EdgeRunner):
should_execute = False
target_id = None
source_id = None
with workflow_tracer.create_edge_group_processing_span(
with create_edge_group_processing_span(
self._edge_group.__class__.__name__,
edge_group_id=self._edge_group.id,
message_source_id=message.source_id,
message_target_id=message.target_id,
source_trace_contexts=message.trace_contexts,
source_span_ids=message.source_span_ids,
):
) as span:
try:
if message.target_id and message.target_id != self._edge.target_id:
workflow_tracer.set_edge_group_span_attributes(
False, EdgeGroupDeliveryStatus.DROPPED_TARGET_MISMATCH
)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: False,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.DROPPED_TARGET_MISMATCH.value,
})
return False
if self._can_handle(self._edge.target_id, message.data):
if self._edge.should_route(message.data):
workflow_tracer.set_edge_group_span_attributes(True, EdgeGroupDeliveryStatus.DELIVERED)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: True,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.DELIVERED.value,
})
should_execute = True
target_id = self._edge.target_id
source_id = self._edge.source_id
else:
workflow_tracer.set_edge_group_span_attributes(
False, EdgeGroupDeliveryStatus.DROPPED_CONDITION_FALSE
)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: False,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.DROPPED_CONDITION_FALSE.value,
})
# Return True here because message was processed, just condition failed
return True
else:
workflow_tracer.set_edge_group_span_attributes(False, EdgeGroupDeliveryStatus.DROPPED_TYPE_MISMATCH)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: False,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.DROPPED_TYPE_MISMATCH.value,
})
return False
except Exception as e:
workflow_tracer.set_edge_group_span_attributes(False, EdgeGroupDeliveryStatus.EXCEPTION)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: False,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.EXCEPTION.value,
})
raise e
# Execute outside the span
@@ -147,22 +157,24 @@ class FanOutEdgeRunner(EdgeRunner):
"""Send a message through all edges in the fan-out edge group."""
deliverable_edges = []
single_target_edge = None
# Process routing logic within span
with workflow_tracer.create_edge_group_processing_span(
with create_edge_group_processing_span(
self._edge_group.__class__.__name__,
edge_group_id=self._edge_group.id,
message_source_id=message.source_id,
message_target_id=message.target_id,
source_trace_contexts=message.trace_contexts,
source_span_ids=message.source_span_ids,
):
) as span:
try:
selection_results = (
self._selection_func(message.data, self._target_ids) if self._selection_func else self._target_ids
)
if not self._validate_selection_result(selection_results):
workflow_tracer.set_edge_group_span_attributes(False, EdgeGroupDeliveryStatus.EXCEPTION)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: False,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.EXCEPTION.value,
})
raise RuntimeError(
f"Invalid selection result: {selection_results}. "
f"Expected selections to be a subset of valid target executor IDs: {self._target_ids}."
@@ -174,24 +186,30 @@ class FanOutEdgeRunner(EdgeRunner):
edge = self._target_map.get(message.target_id)
if edge and self._can_handle(edge.target_id, message.data):
if edge.should_route(message.data):
workflow_tracer.set_edge_group_span_attributes(True, EdgeGroupDeliveryStatus.DELIVERED)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: True,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.DELIVERED.value,
})
single_target_edge = edge
else:
workflow_tracer.set_edge_group_span_attributes(
False, EdgeGroupDeliveryStatus.DROPPED_CONDITION_FALSE
)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: False,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.DROPPED_CONDITION_FALSE.value, # noqa: E501
})
# For targeted messages with condition failure, return True (message was processed)
return True
else:
workflow_tracer.set_edge_group_span_attributes(
False, EdgeGroupDeliveryStatus.DROPPED_TYPE_MISMATCH
)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: False,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.DROPPED_TYPE_MISMATCH.value, # noqa: E501
})
# For targeted messages that can't be handled, return False
return False
else:
workflow_tracer.set_edge_group_span_attributes(
False, EdgeGroupDeliveryStatus.DROPPED_TARGET_MISMATCH
)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: False,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.DROPPED_TARGET_MISMATCH.value,
})
# For targeted messages not in selection, return False
return False
else:
@@ -202,14 +220,21 @@ class FanOutEdgeRunner(EdgeRunner):
deliverable_edges.append(edge)
if len(deliverable_edges) > 0:
workflow_tracer.set_edge_group_span_attributes(True, EdgeGroupDeliveryStatus.DELIVERED)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: True,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.DELIVERED.value,
})
else:
workflow_tracer.set_edge_group_span_attributes(
False, EdgeGroupDeliveryStatus.DROPPED_TYPE_MISMATCH
)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: False,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.DROPPED_TYPE_MISMATCH.value,
})
except Exception as e:
workflow_tracer.set_edge_group_span_attributes(False, EdgeGroupDeliveryStatus.EXCEPTION)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: False,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.EXCEPTION.value,
})
raise e
# Execute outside the span
@@ -250,30 +275,36 @@ class FanInEdgeRunner(EdgeRunner):
async def send_message(self, message: Message, shared_state: SharedState, ctx: RunnerContext) -> bool:
"""Send a message through all edges in the fan-in edge group."""
execution_data: dict[str, Any] | None = None
with workflow_tracer.create_edge_group_processing_span(
with create_edge_group_processing_span(
self._edge_group.__class__.__name__,
edge_group_id=self._edge_group.id,
message_source_id=message.source_id,
message_target_id=message.target_id,
source_trace_contexts=message.trace_contexts,
source_span_ids=message.source_span_ids,
):
) as span:
try:
if message.target_id and message.target_id != self._edges[0].target_id:
workflow_tracer.set_edge_group_span_attributes(
False, EdgeGroupDeliveryStatus.DROPPED_TARGET_MISMATCH
)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: False,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.DROPPED_TARGET_MISMATCH.value,
})
return False
# Check if target can handle list of message data (fan-in aggregates multiple messages)
if self._can_handle(self._edges[0].target_id, [message.data]):
# If the edge can handle the data, buffer the message
self._buffer[message.source_id].append(message)
workflow_tracer.set_edge_group_span_attributes(True, EdgeGroupDeliveryStatus.BUFFERED)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: True,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.BUFFERED.value,
})
else:
# If the edge cannot handle the data, return False
workflow_tracer.set_edge_group_span_attributes(False, EdgeGroupDeliveryStatus.DROPPED_TYPE_MISMATCH)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: False,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.DROPPED_TYPE_MISMATCH.value,
})
return False
if self._is_ready_to_send():
@@ -294,7 +325,10 @@ class FanInEdgeRunner(EdgeRunner):
trace_contexts=trace_contexts,
source_span_ids=source_span_ids,
)
workflow_tracer.set_edge_group_span_attributes(True, EdgeGroupDeliveryStatus.DELIVERED)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: True,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.DELIVERED.value,
})
# Store execution data for later
execution_data = {
@@ -304,7 +338,10 @@ class FanInEdgeRunner(EdgeRunner):
}
except Exception as e:
workflow_tracer.set_edge_group_span_attributes(False, EdgeGroupDeliveryStatus.EXCEPTION)
span.set_attributes({
OtelAttr.EDGE_GROUP_DELIVERED: False,
OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.EXCEPTION.value,
})
raise e
# Execute outside the span if needed
python/packages/main/agent_framework/_workflow/_executor.py
+11 -11
View File
@@ -12,14 +12,13 @@ from textwrap import shorten
from types import UnionType
from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypeVar, Union, cast, get_args, get_origin, overload
if TYPE_CHECKING:
from ._workflow import Workflow
from pydantic import Field
from agent_framework import AgentProtocol, AgentRunResponse, AgentRunResponseUpdate, AgentThread, ChatMessage
from agent_framework._pydantic import AFBaseModel
from .._agents import AgentProtocol
from .._pydantic import AFBaseModel
from .._threads import AgentThread
from .._types import AgentRunResponse, AgentRunResponseUpdate, ChatMessage
from ..observability import create_processing_span
from ._checkpoint import WorkflowCheckpoint
from ._events import (
AgentRunEvent,
@@ -27,14 +26,16 @@ from ._events import (
ExecutorCompletedEvent,
ExecutorInvokedEvent,
RequestInfoEvent,
_framework_event_origin, # pyright: ignore[reportPrivateUsage]
_framework_event_origin, # type: ignore[reportPrivateUsage]
)
from ._runner_context import _decode_checkpoint_value
from ._runner_context import _decode_checkpoint_value # type: ignore[reportPrivateUsage]
from ._typing_utils import is_instance_of
from ._workflow_context import WorkflowContext
logger = logging.getLogger(__name__)
if TYPE_CHECKING:
from ._workflow import Workflow
logger = logging.getLogger(__name__)
# region Executor
@@ -114,7 +115,6 @@ class Executor(AFBaseModel):
An awaitable that resolves to the result of the execution.
"""
# Create processing span for tracing (gracefully handles disabled tracing)
from ._telemetry import workflow_tracer
source_trace_contexts = getattr(context, "_trace_contexts", None)
source_span_ids = getattr(context, "_source_span_ids", None)
@@ -125,7 +125,7 @@ class Executor(AFBaseModel):
if isinstance(message, Message):
message = message.data
with workflow_tracer.create_processing_span(
with create_processing_span(
self.id,
self.__class__.__name__,
type(message).__name__,
python/packages/main/agent_framework/_workflow/_telemetry.py
-310
View File
@@ -1,310 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
from enum import Enum
from typing import TYPE_CHECKING, Any, ClassVar
from opentelemetry.trace import Link, NoOpTracer, SpanKind, StatusCode, get_current_span, get_tracer
from opentelemetry.trace.span import SpanContext
from opentelemetry.util.types import Attributes
from agent_framework._pydantic import AFBaseSettings
if TYPE_CHECKING:
from ._workflow import Workflow
class EdgeGroupDeliveryStatus(Enum):
"""Enum for edge group delivery status values."""
DELIVERED = "delivered"
DROPPED_TYPE_MISMATCH = "dropped type mismatch"
DROPPED_TARGET_MISMATCH = "dropped target mismatch"
DROPPED_CONDITION_FALSE = "dropped condition evaluated to false"
EXCEPTION = "exception"
BUFFERED = "buffered"
# Span name constants
_WORKFLOW_BUILD_SPAN = "workflow.build"
_WORKFLOW_RUN_SPAN = "workflow.run"
_EXECUTOR_PROCESS_SPAN = "executor.process"
_MESSAGE_SEND_SPAN = "message.send"
_EDGE_GROUP_PROCESS_SPAN = "edge_group.process"
class WorkflowDiagnosticSettings(AFBaseSettings):
"""Settings for workflow tracing diagnostics."""
env_prefix: ClassVar[str] = "AGENT_FRAMEWORK_WORKFLOW_"
enable_otel: bool = False
@property
def ENABLED(self) -> bool:
return self.enable_otel
class WorkflowTracer:
"""Central tracing coordinator for workflow system.
Manages OpenTelemetry span creation and relationships for:
- Workflow build spans (workflow.build)
- Workflow execution spans (workflow.run)
- Executor processing spans (executor.process)
- Message sending spans (message.send)
- Edge group processing spans (edge_group.process)
Implements span linking for causality without unwanted nesting.
"""
def __init__(self) -> None:
self.settings = WorkflowDiagnosticSettings()
self.tracer = get_tracer("agent_framework") if self.settings.ENABLED else NoOpTracer()
@property
def enabled(self) -> bool:
return self.settings.ENABLED
def create_workflow_run_span(self, workflow: "Workflow") -> Any:
"""Create a workflow execution span."""
attributes: dict[str, str | int] = {
"workflow.id": workflow.id,
}
return self.tracer.start_as_current_span(_WORKFLOW_RUN_SPAN, kind=SpanKind.INTERNAL, attributes=attributes)
def create_workflow_build_span(self) -> Any:
"""Create a workflow build span."""
return self.tracer.start_as_current_span(_WORKFLOW_BUILD_SPAN, kind=SpanKind.INTERNAL)
def create_processing_span(
self,
executor_id: str,
executor_type: str,
message_type: str,
source_trace_contexts: list[dict[str, str]] | None = None,
source_span_ids: list[str] | None = None,
) -> Any:
"""Create an executor processing span with optional links to source spans.
Processing spans are created as children of the current workflow span and
linked (not nested) to the source publishing spans for causality tracking.
This supports multiple links for fan-in scenarios.
"""
# Create links to source spans for causality without nesting
links = []
if source_trace_contexts and source_span_ids:
# Create links for all source spans (supporting fan-in with multiple sources)
for trace_context, span_id in zip(source_trace_contexts, source_span_ids, strict=False):
try:
# Extract trace and span IDs from the trace context
# This is a simplified approach - in production you'd want more robust parsing
traceparent = trace_context.get("traceparent", "")
if traceparent:
# traceparent format: "00-{trace_id}-{parent_span_id}-{trace_flags}"
parts = traceparent.split("-")
if len(parts) >= 3:
trace_id_hex = parts[1]
# Use the source_span_id that was saved from the publishing span
# Create span context for linking
span_context = SpanContext(
trace_id=int(trace_id_hex, 16),
span_id=int(span_id, 16),
is_remote=True,
)
links.append(Link(span_context))
except (ValueError, TypeError, AttributeError):
# If linking fails, continue without link (graceful degradation)
pass
return self.tracer.start_as_current_span(
_EXECUTOR_PROCESS_SPAN,
kind=SpanKind.INTERNAL,
attributes={
"executor.id": executor_id,
"executor.type": executor_type,
"message.type": message_type,
},
links=links,
)
def create_sending_span(self, message_type: str, target_executor_id: str | None = None) -> Any:
"""Create a message send span.
Sending spans are created as children of the current processing span
to track message emission for distributed tracing.
"""
attributes: dict[str, str] = {
"message.type": message_type,
}
if target_executor_id is not None:
attributes["message.destination_executor_id"] = target_executor_id
return self.tracer.start_as_current_span(
_MESSAGE_SEND_SPAN,
kind=SpanKind.PRODUCER,
attributes=attributes,
)
def create_edge_group_processing_span(
self,
edge_group_type: str,
edge_group_id: str | None = None,
message_source_id: str | None = None,
message_target_id: str | None = None,
source_trace_contexts: list[dict[str, str]] | None = None,
source_span_ids: list[str] | None = None,
) -> Any:
"""Create an edge group processing span with optional links to source spans.
Edge group processing spans track the processing operations in edge runners
before message delivery, including condition checking and routing decisions.
Links to source spans provide causality tracking without unwanted nesting.
Args:
edge_group_type: The type of the edge group (class name).
edge_group_id: The unique ID of the edge group.
message_source_id: The source ID of the message being processed.
message_target_id: The target ID of the message being processed.
source_trace_contexts: Optional trace contexts from source spans for linking.
source_span_ids: Optional source span IDs for linking.
"""
attributes: dict[str, str] = {
"edge_group.type": edge_group_type,
}
if edge_group_id is not None:
attributes["edge_group.id"] = edge_group_id
if message_source_id is not None:
attributes["message.source_id"] = message_source_id
if message_target_id is not None:
attributes["message.target_id"] = message_target_id
# Create links to source spans for causality without nesting
links = []
if source_trace_contexts and source_span_ids:
# Create links for all source spans (supporting fan-in with multiple sources)
for trace_context, span_id in zip(source_trace_contexts, source_span_ids, strict=False):
try:
# Extract trace and span IDs from the trace context
# This is a simplified approach - in production you'd want more robust parsing
traceparent = trace_context.get("traceparent", "")
if traceparent:
# traceparent format: "00-{trace_id}-{parent_span_id}-{trace_flags}"
parts = traceparent.split("-")
if len(parts) >= 3:
trace_id_hex = parts[1]
# Use the source_span_id that was saved from the publishing span
# Create span context for linking
span_context = SpanContext(
trace_id=int(trace_id_hex, 16),
span_id=int(span_id, 16),
is_remote=True,
)
links.append(Link(span_context))
except (ValueError, TypeError, AttributeError):
# If linking fails, continue without link (graceful degradation)
pass
return self.tracer.start_as_current_span(
_EDGE_GROUP_PROCESS_SPAN,
kind=SpanKind.INTERNAL,
attributes=attributes,
links=links,
)
def set_edge_group_span_attributes(self, delivered: bool, delivery_status: EdgeGroupDeliveryStatus) -> None:
"""Set edge group span attributes for delivery status.
Args:
delivered: Whether the message was delivered.
delivery_status: The delivery status from EdgeGroupDeliveryStatus enum.
"""
span = get_current_span()
if span and span.is_recording():
span.set_attributes({
"edge_group.delivered": delivered,
"edge_group.delivery_status": delivery_status.value,
})
def add_workflow_event(self, event_name: str, attributes: Attributes | None = None) -> None:
"""Add an event to the current workflow span.
Args:
event_name: Name of the event (e.g., "workflow.started", "workflow.completed")
attributes: Optional attributes to attach to the event
"""
span = get_current_span()
if span and span.is_recording():
span.add_event(event_name, attributes)
def add_workflow_error_event(self, error: Exception, attributes: Attributes | None = None) -> None:
"""Add an error event to the current workflow span.
Args:
error: The exception that occurred
attributes: Optional additional attributes to attach to the event
"""
span = get_current_span()
if span and span.is_recording():
event_attributes: dict[str, str | bool | int | float] = {
"error.message": str(error),
"error.type": type(error).__name__,
}
if attributes:
# Safely merge attributes, ensuring type compatibility
for key, value in attributes.items():
if isinstance(value, (str, bool, int, float)):
event_attributes[key] = value
span.add_event("workflow.error", event_attributes)
span.set_status(StatusCode.ERROR, str(error))
def set_workflow_build_span_attributes(self, workflow: "Workflow") -> None:
"""Set workflow attributes on the current span.
Args:
workflow: The workflow instance to extract attributes from
"""
span = get_current_span()
if span and span.is_recording():
span.set_attributes({
"workflow.id": workflow.id,
"workflow.definition": workflow.model_dump_json(by_alias=True),
})
def add_build_event(self, event_name: str, attributes: Attributes | None = None) -> None:
"""Add an event to the current workflow build span.
Args:
event_name: Name of the build event (e.g., "build.started", "build.validation_completed")
attributes: Optional attributes to attach to the event
"""
span = get_current_span()
if span and span.is_recording():
span.add_event(event_name, attributes)
def add_build_error_event(self, error: Exception, attributes: Attributes | None = None) -> None:
"""Add an error event to the current workflow build span.
Args:
error: The exception that occurred during build
attributes: Optional additional attributes to attach to the event
"""
span = get_current_span()
if span and span.is_recording():
event_attributes: dict[str, str | bool | int | float] = {
"build.error.message": str(error),
"build.error.type": type(error).__name__,
}
if attributes:
# Safely merge attributes, ensuring type compatibility
for key, value in attributes.items():
if isinstance(value, (str, bool, int, float)):
event_attributes[key] = value
span.add_event("build.error", event_attributes)
span.set_status(StatusCode.ERROR, str(error))
# Global workflow tracer instance
workflow_tracer = WorkflowTracer()
python/packages/main/agent_framework/_workflow/_workflow.py
+36 -25
View File
@@ -11,9 +11,9 @@ from typing import Any
from pydantic import Field
from agent_framework import AgentProtocol
from agent_framework._pydantic import AFBaseModel
from .._agents import AgentProtocol
from .._pydantic import AFBaseModel
from ..observability import OtelAttr, capture_exception, create_workflow_span
from ._agent import WorkflowAgent
from ._checkpoint import CheckpointStorage
from ._const import DEFAULT_MAX_ITERATIONS
@@ -242,17 +242,19 @@ class Workflow(AFBaseModel):
Yields:
WorkflowEvent: The events generated during the workflow execution.
"""
# Import here to avoid circular imports
from ._telemetry import workflow_tracer
# Create workflow span that encompasses the entire execution
with workflow_tracer.create_workflow_run_span(self):
with create_workflow_span(
OtelAttr.WORKFLOW_RUN_SPAN,
{
OtelAttr.WORKFLOW_ID: self.id,
},
) as span:
saw_completed = False
saw_request = False
emitted_in_progress_pending = False
try:
# Add workflow started event (telemetry + surface state to consumers)
workflow_tracer.add_workflow_event("workflow.started")
span.add_event(OtelAttr.WORKFLOW_STARTED)
# Emit explicit start/status events to the stream
with _framework_event_origin():
started = WorkflowStartedEvent()
@@ -298,17 +300,24 @@ class Workflow(AFBaseModel):
terminal_status = WorkflowStatusEvent(WorkflowRunState.IDLE)
yield terminal_status
workflow_tracer.add_workflow_event("workflow.completed")
except Exception as e:
span.add_event(OtelAttr.WORKFLOW_COMPLETED)
except Exception as exc:
# Surface structured failure details before propagating exception
details = WorkflowErrorDetails.from_exception(e)
details = WorkflowErrorDetails.from_exception(exc)
with _framework_event_origin():
failed_event = WorkflowFailedEvent(details)
yield failed_event
with _framework_event_origin():
failed_status = WorkflowStatusEvent(WorkflowRunState.FAILED)
yield failed_status
workflow_tracer.add_workflow_error_event(e)
span.add_event(
name=OtelAttr.WORKFLOW_ERROR,
attributes={
"error.message": str(exc),
"error.type": type(exc).__name__,
},
)
capture_exception(span, exception=exc)
raise
async def run_stream(self, message: Any) -> AsyncIterable[WorkflowEvent]:
@@ -1074,14 +1083,11 @@ class WorkflowBuilder:
WorkflowValidationError: If workflow validation fails (includes EdgeDuplicationError,
TypeCompatibilityError, and GraphConnectivityError subclasses).
"""
# Import here to avoid circular imports
from ._telemetry import workflow_tracer
# Create workflow build span that includes validation and workflow creation
with workflow_tracer.create_workflow_build_span():
with create_workflow_span(OtelAttr.WORKFLOW_BUILD_SPAN) as span:
try:
# Add workflow build started event
workflow_tracer.add_build_event("build.started")
span.add_event(OtelAttr.BUILD_STARTED)
if not self._start_executor:
raise ValueError(
@@ -1097,7 +1103,7 @@ class WorkflowBuilder:
)
# Add validation completed event
workflow_tracer.add_build_event("build.validation_completed")
span.add_event(OtelAttr.BUILD_VALIDATION_COMPLETED)
context = InProcRunnerContext(self._checkpoint_storage)
@@ -1105,16 +1111,21 @@ class WorkflowBuilder:
workflow = Workflow(
self._edge_groups, self._executors, self._start_executor, context, self._max_iterations
)
# Set workflow attributes on the span
workflow_tracer.set_workflow_build_span_attributes(workflow)
span.set_attributes({
OtelAttr.WORKFLOW_ID: workflow.id,
OtelAttr.WORKFLOW_DEFINITION: workflow.model_dump_json(by_alias=True),
})
# Add workflow build completed event
workflow_tracer.add_build_event("build.completed")
span.add_event(OtelAttr.BUILD_COMPLETED)
return workflow
except Exception as e:
# The method already includes sufficient error info (error.message, error.type)
workflow_tracer.add_build_error_event(e)
except Exception as exc:
attributes = {
OtelAttr.BUILD_ERROR_MESSAGE: str(exc),
OtelAttr.BUILD_ERROR_TYPE: type(exc).__name__,
}
span.add_event(OtelAttr.BUILD_ERROR, attributes) # type: ignore[reportArgumentType, arg-type]
capture_exception(span, exc)
raise
python/packages/main/agent_framework/_workflow/_workflow_context.py
+10 -3
View File
@@ -4,7 +4,9 @@ import logging
from typing import Any, Generic, TypeVar, cast, get_args
from opentelemetry.propagate import inject
from opentelemetry.trace import SpanKind
from ..observability import OtelAttr, create_workflow_span
from ._events import (
WorkflowEvent,
WorkflowEventSource,
@@ -16,7 +18,6 @@ from ._events import (
)
from ._runner_context import Message, RunnerContext
from ._shared_state import SharedState
from ._telemetry import workflow_tracer
T_Out = TypeVar("T_Out")
@@ -83,13 +84,19 @@ class WorkflowContext(Generic[T_Out]):
target_id: The ID of the target executor to send the message to.
If None, the message will be sent to all target executors.
"""
global OTEL_SETTINGS
from ..observability import OTEL_SETTINGS
# Create publishing span (inherits current trace context automatically)
with workflow_tracer.create_sending_span(type(message).__name__, target_id) as span:
attributes: dict[str, str] = {OtelAttr.MESSAGE_TYPE: type(message).__name__}
if target_id:
attributes[OtelAttr.MESSAGE_DESTINATION_EXECUTOR_ID] = target_id
with create_workflow_span(OtelAttr.MESSAGE_SEND_SPAN, attributes, kind=SpanKind.PRODUCER) as span:
# Create Message wrapper
msg = Message(data=message, source_id=self._executor_id, target_id=target_id)
# Inject current trace context if tracing enabled
if workflow_tracer.enabled and span and span.is_recording():
if OTEL_SETTINGS.ENABLED and span and span.is_recording(): # type: ignore[name-defined]
trace_context: dict[str, str] = {}
inject(trace_context) # Inject current trace context for message propagation
python/packages/main/agent_framework/telemetry.py → python/packages/main/agent_framework/observability.py
+547 -214
View File
File diff suppressed because it is too large Load Diff
python/packages/main/agent_framework/openai/_assistants_client.py
+2 -2
View File
@@ -38,7 +38,7 @@ from .._types import (
UsageDetails,
)
from ..exceptions import ServiceInitializationError
from ..telemetry import use_telemetry
from ..observability import use_observability
from ._shared import OpenAIConfigMixin, OpenAISettings
if sys.version_info >= (3, 11):
@@ -51,7 +51,7 @@ __all__ = ["OpenAIAssistantsClient"]
@use_function_invocation
@use_telemetry
@use_observability
class OpenAIAssistantsClient(OpenAIConfigMixin, BaseChatClient):
"""OpenAI Assistants client."""
python/packages/main/agent_framework/openai/_chat_client.py
+2 -2
View File
@@ -40,7 +40,7 @@ from ..exceptions import (
ServiceInvalidRequestError,
ServiceResponseException,
)
from ..telemetry import use_telemetry
from ..observability import use_observability
from ._exceptions import OpenAIContentFilterException
from ._shared import OpenAIBase, OpenAIConfigMixin, OpenAISettings, prepare_function_call_results
@@ -451,7 +451,7 @@ TOpenAIChatClient = TypeVar("TOpenAIChatClient", bound="OpenAIChatClient")
@use_function_invocation
@use_telemetry
@use_observability
class OpenAIChatClient(OpenAIConfigMixin, OpenAIBaseChatClient):
"""OpenAI Chat completion class."""
python/packages/main/agent_framework/openai/_responses_client.py
+2 -2
View File
@@ -62,7 +62,7 @@ from ..exceptions import (
ServiceInvalidRequestError,
ServiceResponseException,
)
from ..telemetry import use_telemetry
from ..observability import use_observability
from ._exceptions import OpenAIContentFilterException
from ._shared import OpenAIBase, OpenAIConfigMixin, OpenAISettings, prepare_function_call_results
@@ -933,7 +933,7 @@ TOpenAIResponsesClient = TypeVar("TOpenAIResponsesClient", bound="OpenAIResponse
@use_function_invocation
@use_telemetry
@use_observability
class OpenAIResponsesClient(OpenAIConfigMixin, OpenAIBaseResponsesClient):
"""OpenAI Responses client class."""
python/packages/main/agent_framework/openai/_shared.py
+1 -1
View File
@@ -22,9 +22,9 @@ from pydantic.types import StringConstraints
from .._logging import get_logger
from .._pydantic import AFBaseModel, AFBaseSettings
from .._telemetry import APP_INFO, USER_AGENT_KEY, prepend_agent_framework_to_user_agent
from .._types import ChatOptions, Contents, SpeechToTextOptions, TextToSpeechOptions
from ..exceptions import ServiceInitializationError
from ..telemetry import APP_INFO, USER_AGENT_KEY, prepend_agent_framework_to_user_agent
logger: logging.Logger = get_logger("agent_framework.openai")