diff --git a/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py index 7a418c6bdb6..54a14bd3113 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py +++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py @@ -517,7 +517,10 @@ async def finalizer(message: str, ctx: WorkflowContext[None, str]) -> None: assert not [event for event in events if event.type == "CUSTOM" and event.name == "workflow_output"] # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] -async def test_workflow_and_agent_spans_use_supplied_agui_thread_id(monkeypatch: pytest.MonkeyPatch) -> None: +@pytest.mark.parametrize("workflow_name", [None, "customer_support_pipeline"]) +async def test_workflow_and_agent_spans_use_supplied_agui_thread_id( + monkeypatch: pytest.MonkeyPatch, workflow_name: str | None +) -> None: """Workflow spans use supplied AG-UI threads without replacing provider fallback behavior.""" import agent_framework.observability as observability from opentelemetry.sdk.trace import TracerProvider @@ -561,7 +564,7 @@ async def passthrough_middleware(_context: AgentContext, call_next: Any) -> None name="workflow-agent", middleware=[passthrough_middleware], ) - workflow = WorkflowBuilder(start_executor=participant, output_from="all").build() + workflow = WorkflowBuilder(start_executor=participant, output_from="all", name=workflow_name).build() for run_number in (1, 2): events = [ @@ -601,11 +604,16 @@ async def passthrough_middleware(_context: AgentContext, call_next: Any) -> None "ag-ui-workflow-thread", ] - workflow_spans = [span for span in exporter.get_finished_spans() if span.name == "workflow.run"] + expected_span_name = f"invoke_workflow {workflow_name}" if workflow_name else "invoke_workflow" + workflow_spans = [span for span in exporter.get_finished_spans() if span.name == expected_span_name] assert len(workflow_spans) == 2 workflow_conversation_ids = [] for span in workflow_spans: assert span.attributes is not None + if workflow_name: + assert span.attributes["gen_ai.workflow.name"] == workflow_name + else: + assert "gen_ai.workflow.name" not in span.attributes workflow_conversation_ids.append(span.attributes.get("gen_ai.conversation.id")) assert workflow_conversation_ids == [ @@ -631,7 +639,7 @@ async def passthrough_middleware(_context: AgentContext, call_next: Any) -> None assert fallback_agent_span.attributes is not None assert fallback_agent_span.attributes.get("gen_ai.conversation.id") == "provider-conversation" - fallback_workflow_span = next(span for span in exporter.get_finished_spans() if span.name == "workflow.run") + fallback_workflow_span = next(span for span in exporter.get_finished_spans() if span.name == expected_span_name) assert fallback_workflow_span.attributes is not None assert "gen_ai.conversation.id" not in fallback_workflow_span.attributes diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py index c60617dabd5..cd566778484 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow.py +++ b/python/packages/core/agent_framework/_workflows/_workflow.py @@ -366,6 +366,7 @@ def __init__( self.start_executor_id = start_executor.id self.max_iterations = max_iterations self.name = name + self._generated_name: str | None = None self.description = description # Generate a unique ID for the workflow instance for monitoring purposes. This is not intended to be a # stable identifier across instances created from the same builder, for that, use the name field. @@ -559,7 +560,7 @@ async def _run_workflow_with_tracing( """ # Create workflow span that encompasses the entire execution attributes: dict[str, Any] = {OtelAttr.WORKFLOW_ID: self.id} - if self.name: + if self.name and self.name != self._generated_name: attributes[OtelAttr.WORKFLOW_NAME] = self.name if self.description: attributes[OtelAttr.WORKFLOW_DESCRIPTION] = self.description diff --git a/python/packages/core/agent_framework/_workflows/_workflow_builder.py b/python/packages/core/agent_framework/_workflows/_workflow_builder.py index 3eed9ca800a..084ddcf8fc4 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow_builder.py +++ b/python/packages/core/agent_framework/_workflows/_workflow_builder.py @@ -1,5 +1,7 @@ # Copyright (c) Microsoft. All rights reserved. +from __future__ import annotations + import logging import sys import uuid @@ -110,7 +112,8 @@ def __init__( for all workflow instances created from this builder. If not provided, a unique name will be generated. This will be useful for versioning, monitoring, checkpointing, and debugging workflows. Keeping this name unique across versions of your workflow definitions - is recommended for better observability and management. + is recommended for better observability and management. Generated names are omitted from + workflow invocation span names and the ``gen_ai.workflow.name`` attribute. description: Optional description of what the workflow does. start_executor: The starting executor for the workflow. Can be an Executor instance or SupportsAgentRun instance. @@ -153,6 +156,7 @@ def __init__( self._checkpoint_storage: CheckpointStorage | None = checkpoint_storage self._max_iterations: int = max_iterations self._name: str = name or f"WorkflowBuilder-{uuid.uuid4()!s}" + self._generated_name: str | None = self._name if not name else None self._description: str | None = description # Maps underlying SupportsAgentRun object id -> wrapped Executor so we reuse the same wrapper # across start_executor / add_edge calls. This avoids multiple AgentExecutor instances @@ -866,6 +870,7 @@ async def process(self, text: str, ctx: WorkflowContext[Never, str]) -> None: output_from=output_for_workflow, intermediate_output_from=intermediate_output_for_workflow, ) + workflow._generated_name = self._generated_name # pyright: ignore[reportPrivateUsage] build_attributes: dict[str, Any] = { OtelAttr.WORKFLOW_BUILDER_NAME: self._name, OtelAttr.WORKFLOW_ID: workflow.id, diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py index 9fec610e248..71afdf94bbd 100644 --- a/python/packages/core/agent_framework/observability.py +++ b/python/packages/core/agent_framework/observability.py @@ -295,11 +295,11 @@ class OtelAttr(str, Enum): WORKFLOW_ID = "workflow.id" WORKFLOW_BUILDER_NAME = "workflow_builder.name" WORKFLOW_BUILDER_DESCRIPTION = "workflow_builder.description" - WORKFLOW_NAME = "workflow.name" + WORKFLOW_NAME = "gen_ai.workflow.name" WORKFLOW_DESCRIPTION = "workflow.description" WORKFLOW_DEFINITION = "workflow.definition" WORKFLOW_BUILD_SPAN = "workflow.build" - WORKFLOW_RUN_SPAN = "workflow.run" + WORKFLOW_RUN_SPAN = "invoke_workflow" WORKFLOW_STARTED = "workflow.started" WORKFLOW_COMPLETED = "workflow.completed" WORKFLOW_ERROR = "workflow.error" @@ -3657,15 +3657,19 @@ def workflow_tracer() -> Tracer: return get_tracer() if OBSERVABILITY_SETTINGS.ENABLED else trace.NoOpTracer() -def _workflow_span_attributes( +def _prepare_workflow_span( name: str, attributes: Mapping[str, str | int] | None = None, -) -> dict[str, str | int] | None: +) -> tuple[str, dict[str, str | int] | None]: span_attributes: dict[str, str | int] = dict(attributes) if attributes is not None else {} conversation_id = _TELEMETRY_CONVERSATION_ID.get() - if name == OtelAttr.WORKFLOW_RUN_SPAN and conversation_id is not None: - span_attributes.setdefault(OtelAttr.CONVERSATION_ID, conversation_id) - return span_attributes or None + if name == OtelAttr.WORKFLOW_RUN_SPAN: + span_attributes[OtelAttr.OPERATION] = "invoke_workflow" + if conversation_id is not None: + span_attributes.setdefault(OtelAttr.CONVERSATION_ID, conversation_id) + if workflow_name := span_attributes.get(OtelAttr.WORKFLOW_NAME): + name = f"{name} {workflow_name}" + return name, span_attributes or None @contextlib.contextmanager @@ -3693,11 +3697,12 @@ def create_workflow_span( generator is later closed on GC from a different ``Context``. Streaming workflow runs should use :func:`start_workflow_span` instead. """ + name, span_attributes = _prepare_workflow_span(name, attributes) return _capture_span_exceptions( workflow_tracer().start_as_current_span( name, kind=kind, - attributes=_workflow_span_attributes(name, attributes), + attributes=span_attributes, record_exception=False, set_status_on_exception=False, ) @@ -3717,7 +3722,8 @@ def start_workflow_span( across a yield causes ``ValueError: Token was created in a different Context`` when the generator is garbage-collected. """ - return workflow_tracer().start_span(name, kind=kind, attributes=_workflow_span_attributes(name, attributes)) + name, span_attributes = _prepare_workflow_span(name, attributes) + return workflow_tracer().start_span(name, kind=kind, attributes=span_attributes) def create_processing_span( diff --git a/python/packages/core/tests/core/test_observability.py b/python/packages/core/tests/core/test_observability.py index fab710a7c53..a3e8c8b4002 100644 --- a/python/packages/core/tests/core/test_observability.py +++ b/python/packages/core/tests/core/test_observability.py @@ -3074,31 +3074,44 @@ def test_start_workflow_span_does_not_attach_as_current(span_exporter: InMemoryS assert spans[0].attributes["key"] == "value" # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] -def test_create_workflow_span_uses_scoped_conversation_id(span_exporter: InMemorySpanExporter) -> None: - """An ambient conversation id is applied only within its workflow execution scope.""" +@pytest.mark.parametrize("workflow_name", [None, "", "customer_support_pipeline"]) +@pytest.mark.parametrize("unattached", [False, True]) +def test_workflow_span_uses_scoped_conversation_id( + span_exporter: InMemorySpanExporter, workflow_name: str | None, unattached: bool +) -> None: + """Named and unnamed workflow spans retain scoped conversation ids with either span factory.""" from agent_framework.observability import ( OtelAttr, _use_telemetry_conversation_id, # pyright: ignore[reportPrivateUsage] create_workflow_span, + start_workflow_span, ) + span_factory = start_workflow_span if unattached else create_workflow_span + attributes: dict[str, str | int] = {OtelAttr.WORKFLOW_NAME: workflow_name} if workflow_name is not None else {} span_exporter.clear() # type: ignore[attr-defined] with _use_telemetry_conversation_id("application-thread"): - with create_workflow_span(OtelAttr.WORKFLOW_RUN_SPAN): + with span_factory(OtelAttr.WORKFLOW_RUN_SPAN, attributes=attributes): pass - with create_workflow_span( + with span_factory( OtelAttr.WORKFLOW_RUN_SPAN, - attributes={OtelAttr.CONVERSATION_ID: "explicit-thread"}, + attributes={**attributes, OtelAttr.CONVERSATION_ID: "explicit-thread"}, ): pass - with create_workflow_span(OtelAttr.MESSAGE_SEND_SPAN): + with span_factory(OtelAttr.MESSAGE_SEND_SPAN, attributes=attributes): pass - with create_workflow_span(OtelAttr.WORKFLOW_RUN_SPAN): + with span_factory(OtelAttr.WORKFLOW_RUN_SPAN, attributes=attributes): pass spans = span_exporter.get_finished_spans() # type: ignore[attr-defined] - workflow_spans = [span for span in spans if span.name == OtelAttr.WORKFLOW_RUN_SPAN] + workflow_spans = [ + span + for span in spans + if span.attributes is not None and span.attributes.get(OtelAttr.OPERATION) == "invoke_workflow" + ] assert len(workflow_spans) == 3 + expected_name = f"invoke_workflow {workflow_name}" if workflow_name else "invoke_workflow" + assert all(span.name == expected_name for span in workflow_spans) ambient_attributes = workflow_spans[0].attributes explicit_attributes = workflow_spans[1].attributes unscoped_attributes = workflow_spans[2].attributes @@ -3112,6 +3125,9 @@ def test_create_workflow_span_uses_scoped_conversation_id(span_exporter: InMemor message_send_attributes = message_send_span.attributes assert message_send_attributes is not None assert OtelAttr.CONVERSATION_ID not in message_send_attributes + assert OtelAttr.OPERATION not in message_send_attributes + assert OtelAttr.CONVERSATION_ID not in attributes + assert OtelAttr.OPERATION not in attributes def test_create_processing_span(span_exporter): diff --git a/python/packages/core/tests/workflow/test_functional_workflow.py b/python/packages/core/tests/workflow/test_functional_workflow.py index 08b02ea84a8..3709be62374 100644 --- a/python/packages/core/tests/workflow/test_functional_workflow.py +++ b/python/packages/core/tests/workflow/test_functional_workflow.py @@ -631,7 +631,7 @@ async def pipeline(x: int) -> int: assert follow_up.get_outputs() == [7] async def test_nested_processing_span_parents_under_workflow_run(self, span_exporter: Any) -> None: - """Spans opened during ``_execute`` must parent under the unattached ``workflow.run`` span.""" + """Spans opened during ``_execute`` must parent under the unattached ``invoke_workflow`` span.""" from agent_framework.observability import OtelAttr, create_processing_span @step @@ -648,7 +648,7 @@ async def pipeline(x: int) -> int: assert result.get_outputs() == [6] spans = span_exporter.get_finished_spans() # type: ignore[attr-defined] - run_spans = [s for s in spans if s.name == OtelAttr.WORKFLOW_RUN_SPAN] + run_spans = [s for s in spans if s.name == f"invoke_workflow {pipeline.name}"] process_spans = [s for s in spans if s.name == f"{OtelAttr.EXECUTOR_PROCESS_SPAN} traced_add"] assert len(run_spans) == 1 assert len(process_spans) == 1 diff --git a/python/packages/core/tests/workflow/test_workflow.py b/python/packages/core/tests/workflow/test_workflow.py index 2ecc00c9d35..84df5e5001c 100644 --- a/python/packages/core/tests/workflow/test_workflow.py +++ b/python/packages/core/tests/workflow/test_workflow.py @@ -1434,7 +1434,7 @@ def capture_loop_exception(_loop: asyncio.AbstractEventLoop, context: dict[str, if "Failed to detach context" in rec.getMessage() or "was created in a different Context" in rec.getMessage() ] assert otel_errors == [], f"Abandoned stream leaked OpenTelemetry errors: {otel_errors!r}" - assert any(span.name == "workflow.run" for span in span_exporter.get_finished_spans()) + assert any(span.name == "invoke_workflow" for span in span_exporter.get_finished_spans()) async def test_workflow_unawaited_run_coroutine_releases_run_lock() -> None: diff --git a/python/packages/core/tests/workflow/test_workflow_observability.py b/python/packages/core/tests/workflow/test_workflow_observability.py index b6a8082c8f7..676e678531f 100644 --- a/python/packages/core/tests/workflow/test_workflow_observability.py +++ b/python/packages/core/tests/workflow/test_workflow_observability.py @@ -7,7 +7,7 @@ from opentelemetry import trace from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter -from agent_framework import InMemoryCheckpointStorage, WorkflowBuilder, step, workflow +from agent_framework import FunctionalWorkflow, InMemoryCheckpointStorage, WorkflowBuilder, step, workflow from agent_framework._workflows._events import WorkflowErrorDetails, WorkflowEvent from agent_framework._workflows._executor import Executor, handler from agent_framework._workflows._runner_context import InProcRunnerContext, MessageType, WorkflowMessage @@ -16,6 +16,7 @@ from agent_framework._workflows._workflow_context import WorkflowContext from agent_framework.observability import ( OtelAttr, + _use_telemetry_conversation_id, # pyright: ignore[reportPrivateUsage] create_processing_span, create_workflow_span, ) @@ -120,6 +121,94 @@ def processed_messages(self) -> list[Any]: return self._processed_messages +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize( + ("functional", "workflow_name"), + [ + (False, "customer_support_pipeline"), + (False, None), + (False, ""), + (False, "WorkflowBuilder-00000000-0000-0000-0000-000000000000"), + (True, "customer_support_pipeline"), + (True, None), + ], +) +async def test_workflow_run_span_uses_semantic_convention_name( + span_exporter: InMemorySpanExporter, streaming: bool, functional: bool, workflow_name: str | None +) -> None: + """Graph and functional workflow invocations use standard span and workflow attribute names.""" + workflow_instance: Workflow | FunctionalWorkflow + if functional: + + @workflow(name=workflow_name) + async def simple_workflow(message: str) -> str: + return message + + workflow_instance = simple_workflow.build() + else: + workflow_instance = WorkflowBuilder(start_executor=MockExecutor(), name=workflow_name).build() + + span_exporter.clear() + with _use_telemetry_conversation_id("workflow-thread"): + if streaming: + async for _ in workflow_instance.run("test input", stream=True): + pass + else: + await workflow_instance.run("test input") + + run_spans = [ + span + for span in span_exporter.get_finished_spans() + if any(event.name == "workflow.started" for event in span.events) + ] + assert len(run_spans) == 1 + expected_name = workflow_name or ("simple_workflow" if functional else None) + assert run_spans[0].name == (f"invoke_workflow {expected_name}" if expected_name else "invoke_workflow") + attributes = run_spans[0].attributes + assert attributes is not None + assert "workflow.name" not in attributes + if expected_name: + assert attributes["gen_ai.workflow.name"] == expected_name + else: + assert "gen_ai.workflow.name" not in attributes + assert attributes["gen_ai.operation.name"] == "invoke_workflow" + assert attributes["gen_ai.conversation.id"] == "workflow-thread" + + +@pytest.mark.parametrize("streaming", [False, True]) +async def test_unnamed_workflow_preserves_runtime_name_and_checkpoint_identity( + span_exporter: InMemorySpanExporter, streaming: bool +) -> None: + """Omitting generated telemetry names does not change runtime or checkpoint identity.""" + storage = InMemoryCheckpointStorage() + workflow_instance = WorkflowBuilder(start_executor=MockExecutor(), checkpoint_storage=storage).build() + runtime_name = workflow_instance.name + assert runtime_name.startswith("WorkflowBuilder-") + other_workflow = WorkflowBuilder(start_executor=MockExecutor()).build() + assert other_workflow.name != runtime_name + + span_exporter.clear() + if streaming: + async for _ in workflow_instance.run("test input", stream=True): + pass + else: + await workflow_instance.run("test input") + + assert workflow_instance.name == runtime_name + assert workflow_instance.to_dict()["name"] == runtime_name + checkpoint = await storage.get_latest(workflow_name=runtime_name) + assert checkpoint is not None + assert checkpoint.workflow_name == runtime_name + run_span = next( + span + for span in span_exporter.get_finished_spans() + if span.attributes is not None and span.attributes.get("gen_ai.operation.name") == "invoke_workflow" + ) + assert run_span.name == "invoke_workflow" + assert run_span.attributes is not None + assert "gen_ai.workflow.name" not in run_span.attributes + + async def test_span_creation_and_attributes(span_exporter: InMemorySpanExporter) -> None: """Test creation and attributes of all span types (workflow, processing, sending).""" # Create a mock workflow object @@ -143,6 +232,7 @@ async def test_span_creation_and_attributes(span_exporter: InMemorySpanExporter) OtelAttr.WORKFLOW_ID: mock_workflow.id, }, ) as workflow_span: + assert _span_attributes(workflow_span)["gen_ai.operation.name"] == "invoke_workflow" workflow_span.add_event(OtelAttr.WORKFLOW_STARTED) sending_attributes: dict[str, str | int] = { OtelAttr.MESSAGE_TYPE: "ResponseMessage", @@ -160,12 +250,14 @@ async def test_span_creation_and_attributes(span_exporter: InMemorySpanExporter) assert workflow_span is not None and workflow_span.is_recording() assert processing_span is not None and processing_span.is_recording() assert sending_span is not None and sending_span.is_recording() + assert "gen_ai.operation.name" not in _span_attributes(processing_span) + assert "gen_ai.operation.name" not in _span_attributes(sending_span) spans = span_exporter.get_finished_spans() assert len(spans) == 3 # Check workflow span - workflow_span = next(s for s in spans if s.name == "workflow.run") # type: ignore[assignment, misc] + workflow_span = next(s for s in spans if s.name == "invoke_workflow") # type: ignore[assignment, misc] assert workflow_span.kind == trace.SpanKind.INTERNAL # type: ignore[attr-defined] assert workflow_span.attributes is not None # type: ignore[attr-defined] assert workflow_span.attributes.get(OtelAttr.WORKFLOW_ID) == "test-workflow-123" # type: ignore[attr-defined] @@ -359,7 +451,7 @@ async def test_end_to_end_workflow_tracing(span_exporter: InMemorySpanExporter) # Should have workflow span, processing spans, and sending spans # Processing spans now use executor_id as the span name, filter by executor.id attribute - workflow_spans = [s for s in spans if s.name == "workflow.run"] + workflow_spans = [s for s in spans if s.name == "invoke_workflow"] processing_spans = [s for s in spans if _span_attributes(s).get("executor.id") is not None] sending_spans = [s for s in spans if s.name == "message.send"] build_spans_after_run = [s for s in spans if s.name == "workflow.build"] @@ -377,7 +469,7 @@ async def test_end_to_end_workflow_tracing(span_exporter: InMemorySpanExporter) assert all( processing_span.parent is not None and processing_span.parent.span_id == workflow_span.context.span_id for processing_span in processing_spans - ), "Executor processing spans should be children of workflow.run" + ), "Executor processing spans should be children of invoke_workflow" assert workflow_span.events is not None event_names = [event.name for event in workflow_span.events] assert "workflow.started" in event_names @@ -464,7 +556,7 @@ async def handle_message(self, message: str, ctx: WorkflowContext) -> None: spans = span_exporter.get_finished_spans() # Find workflow span - workflow_spans = [s for s in spans if s.name == "workflow.run"] + workflow_spans = [s for s in spans if s.name == "invoke_workflow"] assert len(workflow_spans) == 1 workflow_span = workflow_spans[0] @@ -542,7 +634,7 @@ async def failing_workflow(message: str) -> None: _assert_workflow_error_details(executor_failed_events[0].details, private_error_detail, enable_sensitive_data) spans = span_exporter.get_finished_spans() - workflow_spans = [span for span in spans if span.name == "workflow.run"] + workflow_spans = [span for span in spans if span.name == f"invoke_workflow {workflow_instance.name}"] assert len(workflow_spans) == 1 span = workflow_spans[0] assert span.attributes is not None diff --git a/skills/agent-framework-builder-python/references/observability.md b/skills/agent-framework-builder-python/references/observability.md index b3f7b3e1abe..5337e27e9b4 100644 --- a/skills/agent-framework-builder-python/references/observability.md +++ b/skills/agent-framework-builder-python/references/observability.md @@ -17,6 +17,12 @@ runtime evidence; it does not determine whether output is correct. Inspect run boundaries, model calls, tool and MCP calls, workflow transitions, checkpoints, approvals, retries, usage, and hosted-service continuations. Use stable agent, workflow, and service identifiers. +Workflow invocation spans are named `invoke_workflow {gen_ai.workflow.name}` when the workflow-name attribute is +non-empty, or `invoke_workflow` otherwise. Their `gen_ai.operation.name` is always `invoke_workflow`; supplied +conversation identifiers remain available as `gen_ai.conversation.id`. +Graph workflows without an application-provided name retain their generated runtime/checkpoint name, but omit it +from `gen_ai.workflow.name` and use the bare invocation span name. Functional workflows default to their function name. + ## Learn more - [Agent observability](https://learn.microsoft.com/agent-framework/agents/observability)