diff --git a/dotnet/Directory.Packages.props b/dotnet/Directory.Packages.props index 9e755dab26..75d67b3165 100644 --- a/dotnet/Directory.Packages.props +++ b/dotnet/Directory.Packages.props @@ -82,7 +82,7 @@ - + diff --git a/dotnet/agent-framework-dotnet.slnx b/dotnet/agent-framework-dotnet.slnx index e662856548..c9d8067e2d 100644 --- a/dotnet/agent-framework-dotnet.slnx +++ b/dotnet/agent-framework-dotnet.slnx @@ -33,6 +33,7 @@ + diff --git a/dotnet/samples/GettingStarted/AgentProviders/Agent_With_OpenAIAssistants/Agent_With_OpenAIAssistants.csproj b/dotnet/samples/GettingStarted/AgentProviders/Agent_With_OpenAIAssistants/Agent_With_OpenAIAssistants.csproj new file mode 100644 index 0000000000..27d3e4077f --- /dev/null +++ b/dotnet/samples/GettingStarted/AgentProviders/Agent_With_OpenAIAssistants/Agent_With_OpenAIAssistants.csproj @@ -0,0 +1,16 @@ + + + + Exe + net9.0 + + enable + disable + + + + + + + + diff --git a/dotnet/samples/GettingStarted/AgentProviders/Agent_With_OpenAIAssistants/Program.cs b/dotnet/samples/GettingStarted/AgentProviders/Agent_With_OpenAIAssistants/Program.cs new file mode 100644 index 0000000000..989c5663ac --- /dev/null +++ b/dotnet/samples/GettingStarted/AgentProviders/Agent_With_OpenAIAssistants/Program.cs @@ -0,0 +1,42 @@ +// Copyright (c) Microsoft. All rights reserved. + +// This sample shows how to create and use a simple AI agent with OpenAI Assistants as the backend. + +// WARNING: The Assistants API is deprecated and will be shut down. +// For more information see the OpenAI documentation: https://platform.openai.com/docs/assistants/migration + +#pragma warning disable OPENAI001 // Type is for evaluation purposes only and is subject to change or removal in future updates. Suppress this diagnostic to proceed. + +using System; +using Microsoft.Extensions.AI.Agents; +using OpenAI; + +var apiKey = Environment.GetEnvironmentVariable("OPENAI_APIKEY") ?? throw new InvalidOperationException("OPENAI_APIKEY is not set."); +var mmodel = Environment.GetEnvironmentVariable("OPENAI_MODEL") ?? "gpt-4o-mini"; + +const string JokerName = "Joker"; +const string JokerInstructions = "You are good at telling jokes."; + +// Get a client to create/retrieve server side agents with. +var assistantClient = new OpenAIClient(apiKey).GetAssistantClient(); + +// You can create a server side assistant with the OpenAI SDK. +var createResult = await assistantClient.CreateAssistantAsync(mmodel, new() { Name = JokerName, Instructions = JokerInstructions }); + +// You can retrieve an already created server side assistant as an AIAgent. +AIAgent agent1 = await assistantClient.GetAIAgentAsync(createResult.Value.Id); + +// You can also create a server side assistant and return it as an AIAgent directly. +AIAgent agent2 = await assistantClient.CreateAIAgentAsync( + model: mmodel, + name: JokerName, + instructions: JokerInstructions); + +// You can invoke the agent like any other AIAgent. +AgentThread thread = agent1.GetNewThread(); +Console.WriteLine(await agent1.RunAsync("Tell me a joke about a pirate.", thread)); + +// Cleanup for sample purposes. +await assistantClient.DeleteThreadAsync(thread.ConversationId); +await assistantClient.DeleteAssistantAsync(agent1.Id); +await assistantClient.DeleteAssistantAsync(agent2.Id); diff --git a/dotnet/samples/GettingStarted/AgentProviders/Agent_With_OpenAIAssistants/README.md b/dotnet/samples/GettingStarted/AgentProviders/Agent_With_OpenAIAssistants/README.md new file mode 100644 index 0000000000..22a4bae18c --- /dev/null +++ b/dotnet/samples/GettingStarted/AgentProviders/Agent_With_OpenAIAssistants/README.md @@ -0,0 +1,16 @@ +# Prerequisites + +WARNING: The Assistants API is deprecated and will be shut down. +For more information see the OpenAI documentation: https://platform.openai.com/docs/assistants/migration + +Before you begin, ensure you have the following prerequisites: + +- .NET 8.0 SDK or later +- OpenAI API key + +Set the following environment variables: + +```powershell +$env:OPENAI_APIKEY="*****" # Replace with your OpenAI API key +$env:OPENAI_MODEL="gpt-4o-mini" # Optional, defaults to gpt-4o-mini +``` diff --git a/dotnet/samples/GettingStarted/AgentProviders/README.md b/dotnet/samples/GettingStarted/AgentProviders/README.md index 0b5fbf1a74..aabf587785 100644 --- a/dotnet/samples/GettingStarted/AgentProviders/README.md +++ b/dotnet/samples/GettingStarted/AgentProviders/README.md @@ -21,6 +21,7 @@ See the README.md for each sample for the prerequisites for that sample. |[Creating an AIAgent with a custom implementation](./Agent_With_CustomImplementation/)|This sample demonstrates how to create an AIAgent with a custom implementation| |[Creating an AIAgent with Ollama](./Agent_With_Ollama/)|This sample demonstrates how to create an AIAgent using Ollama as the underlying inference service| |[Creating an AIAgent with ONNX](./Agent_With_ONNX/)|This sample demonstrates how to create an AIAgent using ONNX as the underlying inference service| +|[Creating an AIAgent with OpenAI Assistants](./Agent_With_OpenAIAssistants/)|This sample demonstrates how to create an AIAgent using OpenAI Assistants as the underlying inference service.
WARNING: The Assistants API is deprecated and will be shut down. For more information see the OpenAI documentation: https://platform.openai.com/docs/assistants/migration| |[Creating an AIAgent with OpenAI ChatCompletion](./Agent_With_OpenAIChatCompletion/)|This sample demonstrates how to create an AIAgent using OpenAI ChatCompletion as the underlying inference service| |[Creating an AIAgent with OpenAI Responses](./Agent_With_OpenAIResponses/)|This sample demonstrates how to create an AIAgent using OpenAI Responses as the underlying inference service| diff --git a/dotnet/samples/GettingStarted/Workflows/Declarative/Program.cs b/dotnet/samples/GettingStarted/Workflows/Declarative/Program.cs index 6f06880506..439ffff7c6 100644 --- a/dotnet/samples/GettingStarted/Workflows/Declarative/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Declarative/Program.cs @@ -33,7 +33,16 @@ internal sealed class Program { public static async Task Main(string[] args) { - Program program = new(args); + string? workflowFile = ParseWorkflowFile(args); + if (workflowFile is null) + { + Notify("\nUsage: DeclarativeWorkflow []"); + return; + } + + string? workflowInput = ParseWorkflowInput(args); + + Program program = new(workflowFile, workflowInput); await program.ExecuteAsync(); } @@ -103,7 +112,6 @@ internal sealed class Program return DeclarativeWorkflowBuilder.Build(this.WorkflowFile, options); } - private const string DefaultWorkflow = "HelloWorld.yaml"; private const string ConfigKeyFoundryEndpoint = "FOUNDRY_PROJECT_ENDPOINT"; private static Dictionary NameCache { get; } = []; @@ -116,10 +124,10 @@ internal sealed class Program private IConfiguration Configuration { get; } private CheckpointInfo? LastCheckpoint { get; set; } - private Program(string[] args) + private Program(string workflowFile, string? workflowInput) { - this.WorkflowFile = ParseWorkflowFile(args); - this.WorkflowInput = ParseWorkflowInput(args); + this.WorkflowFile = workflowFile; + this.WorkflowInput = workflowInput; this.Configuration = InitializeConfig(); @@ -270,9 +278,13 @@ internal sealed class Program return new InputResponse(userInput); } - private static string ParseWorkflowFile(string[] args) + private static string? ParseWorkflowFile(string[] args) { - string workflowFile = args.FirstOrDefault() ?? DefaultWorkflow; + string? workflowFile = args.FirstOrDefault(); + if (string.IsNullOrWhiteSpace(workflowFile)) + { + return null; + } if (!File.Exists(workflowFile) && !Path.IsPathFullyQualified(workflowFile)) { diff --git a/dotnet/src/Microsoft.Extensions.AI.Agents.OpenAI/Extensions/AssistantClientResultExtensions.cs b/dotnet/src/Microsoft.Extensions.AI.Agents.OpenAI/Extensions/AssistantClientResultExtensions.cs index 82db6f781b..79bd755ce4 100644 --- a/dotnet/src/Microsoft.Extensions.AI.Agents.OpenAI/Extensions/AssistantClientResultExtensions.cs +++ b/dotnet/src/Microsoft.Extensions.AI.Agents.OpenAI/Extensions/AssistantClientResultExtensions.cs @@ -25,7 +25,7 @@ public static class AssistantExtensions throw new ArgumentNullException(nameof(assistantClientResult)); } - return AsAIAgent(assistantClientResult, assistantClient, chatOptions); + return AsAIAgent(assistantClientResult.Value, assistantClient, chatOptions); } /// diff --git a/python/packages/main/agent_framework/_workflow/__init__.py b/python/packages/main/agent_framework/_workflow/__init__.py index f8c3955c9d..8848794a74 100644 --- a/python/packages/main/agent_framework/_workflow/__init__.py +++ b/python/packages/main/agent_framework/_workflow/__init__.py @@ -31,12 +31,14 @@ from ._events import ( ExecutorCompletedEvent, ExecutorEvent, ExecutorFailedEvent, - ExecutorInvokeEvent, + ExecutorInvokedEvent, RequestInfoEvent, WorkflowCompletedEvent, WorkflowErrorDetails, WorkflowEvent, + WorkflowEventSource, WorkflowFailedEvent, + WorkflowLifecycleEvent, WorkflowRunState, WorkflowStartedEvent, WorkflowStatusEvent, @@ -118,7 +120,7 @@ __all__ = [ "ExecutorCompletedEvent", "ExecutorEvent", "ExecutorFailedEvent", - "ExecutorInvokeEvent", + "ExecutorInvokedEvent", "FanInEdgeGroup", "FanOutEdgeGroup", "FileCheckpointStorage", @@ -172,8 +174,10 @@ __all__ = [ "WorkflowContext", "WorkflowErrorDetails", "WorkflowEvent", + "WorkflowEventSource", "WorkflowExecutor", "WorkflowFailedEvent", + "WorkflowLifecycleEvent", "WorkflowRunResult", "WorkflowRunState", "WorkflowStartedEvent", diff --git a/python/packages/main/agent_framework/_workflow/__init__.pyi b/python/packages/main/agent_framework/_workflow/__init__.pyi index 506cac18e3..90d2a4322e 100644 --- a/python/packages/main/agent_framework/_workflow/__init__.pyi +++ b/python/packages/main/agent_framework/_workflow/__init__.pyi @@ -27,12 +27,14 @@ from ._events import ( ExecutorCompletedEvent, ExecutorEvent, ExecutorFailedEvent, - ExecutorInvokeEvent, + ExecutorInvokedEvent, RequestInfoEvent, WorkflowCompletedEvent, WorkflowErrorDetails, WorkflowEvent, + WorkflowEventSource, WorkflowFailedEvent, + WorkflowLifecycleEvent, WorkflowRunState, WorkflowStartedEvent, WorkflowStatusEvent, @@ -114,7 +116,7 @@ __all__ = [ "ExecutorCompletedEvent", "ExecutorEvent", "ExecutorFailedEvent", - "ExecutorInvokeEvent", + "ExecutorInvokedEvent", "FanInEdgeGroup", "FanOutEdgeGroup", "FileCheckpointStorage", @@ -168,8 +170,10 @@ __all__ = [ "WorkflowContext", "WorkflowErrorDetails", "WorkflowEvent", + "WorkflowEventSource", "WorkflowExecutor", "WorkflowFailedEvent", + "WorkflowLifecycleEvent", "WorkflowRunResult", "WorkflowRunState", "WorkflowStartedEvent", diff --git a/python/packages/main/agent_framework/_workflow/_events.py b/python/packages/main/agent_framework/_workflow/_events.py index a04136eb0e..d8c16aeb4c 100644 --- a/python/packages/main/agent_framework/_workflow/_events.py +++ b/python/packages/main/agent_framework/_workflow/_events.py @@ -1,9 +1,12 @@ # Copyright (c) Microsoft. All rights reserved. import traceback as _traceback +from collections.abc import Iterator +from contextlib import contextmanager +from contextvars import ContextVar from dataclasses import dataclass from enum import Enum -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, TypeAlias from agent_framework import AgentRunResponse, AgentRunResponseUpdate @@ -11,32 +14,70 @@ if TYPE_CHECKING: from ._executor import RequestInfoMessage +class WorkflowEventSource(str, Enum): + """Identifies whether a workflow event came from the framework or an executor. + + Use `FRAMEWORK` for events emitted by built-in orchestration paths—even when the + code that raises them lives in runner-related modules—and `EXECUTOR` for events + surfaced by developer-provided executor implementations. + """ + + FRAMEWORK = "FRAMEWORK" # Framework-owned orchestration, regardless of module location + EXECUTOR = "EXECUTOR" # User-supplied executor code and callbacks + + +_event_origin_context: ContextVar[WorkflowEventSource] = ContextVar( + "workflow_event_origin", default=WorkflowEventSource.EXECUTOR +) + + +def _current_event_origin() -> WorkflowEventSource: + """Return the origin to associate with newly created workflow events.""" + return _event_origin_context.get() + + +@contextmanager +def _framework_event_origin() -> Iterator[None]: # pyright: ignore[reportUnusedFunction] + """Temporarily mark subsequently created events as originating from the framework (internal).""" + token = _event_origin_context.set(WorkflowEventSource.FRAMEWORK) + try: + yield + finally: + _event_origin_context.reset(token) + + class WorkflowEvent: """Base class for workflow events.""" def __init__(self, data: Any | None = None): """Initialize the workflow event with optional data.""" self.data = data + self.origin = _current_event_origin() def __repr__(self) -> str: """Return a string representation of the workflow event.""" - return f"{self.__class__.__name__}(data={self.data if self.data is not None else 'None'})" + data_repr = self.data if self.data is not None else "None" + return f"{self.__class__.__name__}(origin={self.origin}, data={data_repr})" class WorkflowStartedEvent(WorkflowEvent): - """Event triggered when a workflow starts.""" + """Built-in lifecycle event emitted when a workflow run begins.""" ... class WorkflowCompletedEvent(WorkflowEvent): - """Event triggered when a workflow completes.""" + """Built-in lifecycle event emitted when a workflow run completes successfully. + + Unlike the framework-only `WorkflowLifecycleEvent` union, this event can be + emitted by developer-provided executors to return final workflow output. + """ ... class WorkflowWarningEvent(WorkflowEvent): - """Event triggered when a warning occurs in the workflow.""" + """Executor-origin event signaling a warning surfaced by user code.""" def __init__(self, data: str): """Initialize the workflow warning event with optional data and warning message.""" @@ -44,11 +85,11 @@ class WorkflowWarningEvent(WorkflowEvent): def __repr__(self) -> str: """Return a string representation of the workflow warning event.""" - return f"{self.__class__.__name__}(message={self.data})" + return f"{self.__class__.__name__}(message={self.data}, origin={self.origin})" class WorkflowErrorEvent(WorkflowEvent): - """Event triggered when an error occurs in the workflow.""" + """Executor-origin event signaling an error surfaced by user code.""" def __init__(self, data: Exception): """Initialize the workflow error event with optional data and error message.""" @@ -56,7 +97,7 @@ class WorkflowErrorEvent(WorkflowEvent): def __repr__(self) -> str: """Return a string representation of the workflow error event.""" - return f"{self.__class__.__name__}(exception={self.data})" + return f"{self.__class__.__name__}(exception={self.data}, origin={self.origin})" class WorkflowRunState(str, Enum): @@ -108,14 +149,24 @@ class WorkflowRunState(str, Enum): class WorkflowStatusEvent(WorkflowEvent): - """Event indicating a transition in the workflow run state.""" + """Built-in lifecycle event emitted for workflow run state transitions.""" - def __init__(self, state: WorkflowRunState, data: Any | None = None): + def __init__( + self, + state: WorkflowRunState, + data: Any | None = None, + ): + """Initialize the workflow status event with a new state and optional data. + + Args: + state: The new state of the workflow run. + data: Optional additional data associated with the state change. + """ super().__init__(data) self.state = state def __repr__(self) -> str: # pragma: no cover - representation only - return f"{self.__class__.__name__}(state={self.state}, data={self.data!r})" + return f"{self.__class__.__name__}(state={self.state}, data={self.data!r}, origin={self.origin})" @dataclass @@ -151,14 +202,18 @@ class WorkflowErrorDetails: class WorkflowFailedEvent(WorkflowEvent): - """Terminal failure event for a workflow run.""" + """Built-in lifecycle event emitted when a workflow run terminates with an error.""" - def __init__(self, details: WorkflowErrorDetails, data: Any | None = None): + def __init__( + self, + details: WorkflowErrorDetails, + data: Any | None = None, + ): super().__init__(data) self.details = details def __repr__(self) -> str: # pragma: no cover - representation only - return f"{self.__class__.__name__}(details={self.details}, data={self.data!r})" + return f"{self.__class__.__name__}(details={self.details}, data={self.data!r}, origin={self.origin})" class RequestInfoEvent(WorkflowEvent): @@ -208,12 +263,12 @@ class ExecutorEvent(WorkflowEvent): return f"{self.__class__.__name__}(executor_id={self.executor_id}, data={self.data})" -class ExecutorInvokeEvent(ExecutorEvent): +class ExecutorInvokedEvent(ExecutorEvent): """Event triggered when an executor handler is invoked.""" def __repr__(self) -> str: """Return a string representation of the executor handler invoke event.""" - return f"{self.__class__.__name__}(executor_id={self.executor_id})" + return f"{self.__class__.__name__}(executor_id={self.executor_id}, data={self.data})" class ExecutorCompletedEvent(ExecutorEvent): @@ -221,13 +276,17 @@ class ExecutorCompletedEvent(ExecutorEvent): def __repr__(self) -> str: """Return a string representation of the executor handler complete event.""" - return f"{self.__class__.__name__}(executor_id={self.executor_id})" + return f"{self.__class__.__name__}(executor_id={self.executor_id}, data={self.data})" class ExecutorFailedEvent(ExecutorEvent): """Event triggered when an executor handler raises an error.""" - def __init__(self, executor_id: str, details: WorkflowErrorDetails): + def __init__( + self, + executor_id: str, + details: WorkflowErrorDetails, + ): super().__init__(executor_id, details) self.details = details @@ -257,3 +316,6 @@ class AgentRunEvent(ExecutorEvent): def __repr__(self) -> str: """Return a string representation of the agent run event.""" return f"{self.__class__.__name__}(executor_id={self.executor_id}, data={self.data})" + + +WorkflowLifecycleEvent: TypeAlias = WorkflowStartedEvent | WorkflowStatusEvent | WorkflowFailedEvent diff --git a/python/packages/main/agent_framework/_workflow/_executor.py b/python/packages/main/agent_framework/_workflow/_executor.py index 83763bc02a..8a563e21f9 100644 --- a/python/packages/main/agent_framework/_workflow/_executor.py +++ b/python/packages/main/agent_framework/_workflow/_executor.py @@ -21,8 +21,9 @@ from ._events import ( AgentRunEvent, AgentRunUpdateEvent, ExecutorCompletedEvent, - ExecutorInvokeEvent, + ExecutorInvokedEvent, RequestInfoEvent, + _framework_event_origin, # pyright: ignore[reportPrivateUsage] ) from ._typing_utils import is_instance_of from ._workflow_context import WorkflowContext @@ -102,16 +103,22 @@ class Executor(AFBaseModel): # Lazy registration for SubWorkflowRequestInfo if we have interceptors if self._request_interceptors and message.__class__.__name__ == "SubWorkflowRequestInfo": # Directly handle SubWorkflowRequestInfo - await context.add_event(ExecutorInvokeEvent(self.id)) + with _framework_event_origin(): + invoke_event = ExecutorInvokedEvent(self.id) + await context.add_event(invoke_event) try: await self._handle_sub_workflow_request(message, context) except Exception as exc: # Surface structured executor failure before propagating from ._events import ExecutorFailedEvent, WorkflowErrorDetails - await context.add_event(ExecutorFailedEvent(self.id, WorkflowErrorDetails.from_exception(exc))) + with _framework_event_origin(): + failure_event = ExecutorFailedEvent(self.id, WorkflowErrorDetails.from_exception(exc)) + await context.add_event(failure_event) raise - await context.add_event(ExecutorCompletedEvent(self.id)) + with _framework_event_origin(): + completed_event = ExecutorCompletedEvent(self.id) + await context.add_event(completed_event) return handler: Callable[[Any, WorkflowContext[Any]], Any] | None = None @@ -122,16 +129,22 @@ class Executor(AFBaseModel): if handler is None: raise RuntimeError(f"Executor {self.__class__.__name__} cannot handle message of type {type(message)}.") - await context.add_event(ExecutorInvokeEvent(self.id)) + with _framework_event_origin(): + invoke_event = ExecutorInvokedEvent(self.id) + await context.add_event(invoke_event) try: await handler(message, context) except Exception as exc: # Surface structured executor failure before propagating from ._events import ExecutorFailedEvent, WorkflowErrorDetails - await context.add_event(ExecutorFailedEvent(self.id, WorkflowErrorDetails.from_exception(exc))) + with _framework_event_origin(): + failure_event = ExecutorFailedEvent(self.id, WorkflowErrorDetails.from_exception(exc)) + await context.add_event(failure_event) raise - await context.add_event(ExecutorCompletedEvent(self.id)) + with _framework_event_origin(): + completed_event = ExecutorCompletedEvent(self.id) + await context.add_event(completed_event) def _discover_handlers(self) -> None: """Discover message handlers and request interceptors in the executor class.""" diff --git a/python/packages/main/agent_framework/_workflow/_runner.py b/python/packages/main/agent_framework/_workflow/_runner.py index 05e81619df..b501ff205f 100644 --- a/python/packages/main/agent_framework/_workflow/_runner.py +++ b/python/packages/main/agent_framework/_workflow/_runner.py @@ -11,7 +11,7 @@ if TYPE_CHECKING: from ._edge import EdgeGroup from ._edge_runner import EdgeRunner, create_edge_runner -from ._events import WorkflowCompletedEvent, WorkflowEvent +from ._events import WorkflowCompletedEvent, WorkflowEvent, _framework_event_origin from ._executor import Executor from ._runner_context import Message, RunnerContext from ._shared_state import SharedState @@ -280,7 +280,9 @@ class Runner: if isinstance(message.data, AgentExecutorResponse): final_messages = message.data.agent_run_response.messages final_text = final_messages[-1].text if final_messages else "(no content)" - await self._ctx.add_event(WorkflowCompletedEvent(final_text)) + with _framework_event_origin(): + completion_event = WorkflowCompletedEvent(final_text) + await self._ctx.add_event(completion_event) continue # Terminal handled except Exception as exc: # pragma: no cover - defensive logger.debug("Suppressed exception during terminal message type check: %s", exc) @@ -301,7 +303,9 @@ class Runner: # Emit a single completion event with final text (best-effort extraction) final_messages = message.data.agent_run_response.messages final_text = final_messages[-1].text if final_messages else "(no content)" - await self._ctx.add_event(WorkflowCompletedEvent(final_text)) + with _framework_event_origin(): + completion_event = WorkflowCompletedEvent(final_text) + await self._ctx.add_event(completion_event) continue except Exception as exc: # pragma: no cover logger.debug("Terminal completion emission failed: %s", exc) diff --git a/python/packages/main/agent_framework/_workflow/_workflow.py b/python/packages/main/agent_framework/_workflow/_workflow.py index d4915fb8bd..2fe61c2209 100644 --- a/python/packages/main/agent_framework/_workflow/_workflow.py +++ b/python/packages/main/agent_framework/_workflow/_workflow.py @@ -5,7 +5,7 @@ import logging import sys import uuid from collections.abc import AsyncIterable, Awaitable, Callable, Sequence -from typing import Any, cast +from typing import Any from pydantic import Field @@ -35,6 +35,7 @@ from ._events import ( WorkflowRunState, WorkflowStartedEvent, WorkflowStatusEvent, + _framework_event_origin, ) from ._executor import AgentExecutor, Executor, RequestInfoExecutor from ._runner import Runner @@ -52,14 +53,6 @@ else: logger = logging.getLogger(__name__) -def _default_edge_groups() -> list[EdgeGroup]: - return [] - - -def _default_executors() -> dict[str, Executor]: - return {} - - class WorkflowRunResult(list[WorkflowEvent]): """A list of events generated during the workflow execution in non-streaming mode. @@ -125,10 +118,10 @@ class Workflow(AFBaseModel): """ edge_groups: list[EdgeGroup] = Field( - default_factory=_default_edge_groups, description="List of edge groups that define the workflow edges" + default_factory=list, description="List of edge groups that define the workflow edges" ) executors: dict[str, Executor] = Field( - default_factory=_default_executors, description="Dictionary mapping executor IDs to Executor instances" + default_factory=dict, description="Dictionary mapping executor IDs to Executor instances" ) start_executor_id: str = Field(min_length=1, description="The ID of the starting executor for the workflow") max_iterations: int = Field( @@ -190,20 +183,21 @@ class Workflow(AFBaseModel): # Ensure WorkflowExecutor instances have their workflow field serialized if "executors" in data: - executors_data = cast(dict[str, Any], data["executors"]) - executor_map: dict[str, Executor] = self.executors + executors_data = data["executors"] for executor_id, executor_data in executors_data.items(): # Check if this is a WorkflowExecutor that might be missing its workflow field - if isinstance(executor_data, dict): - executor_dict = cast(dict[str, Any], executor_data) - if executor_dict.get("type") == "WorkflowExecutor" and "workflow" not in executor_dict: - # Get the original executor object and serialize its workflow - original_executor = executor_map.get(executor_id) - if original_executor is not None and hasattr(original_executor, "workflow"): - from ._executor import WorkflowExecutor + if ( + isinstance(executor_data, dict) + and executor_data.get("type") == "WorkflowExecutor" + and "workflow" not in executor_data + ): + # Get the original executor object and serialize its workflow + original_executor = self.executors.get(executor_id) + if original_executor and hasattr(original_executor, "workflow"): + from ._executor import WorkflowExecutor - if isinstance(original_executor, WorkflowExecutor): - executor_dict["workflow"] = original_executor.workflow.model_dump(**kwargs) + if isinstance(original_executor, WorkflowExecutor): + executor_data["workflow"] = original_executor.workflow.model_dump(**kwargs) return data @@ -252,8 +246,12 @@ class Workflow(AFBaseModel): # Add workflow started event (telemetry + surface state to consumers) workflow_tracer.add_workflow_event("workflow.started") # Emit explicit start/status events to the stream - yield WorkflowStartedEvent() - yield WorkflowStatusEvent(WorkflowRunState.IN_PROGRESS) + with _framework_event_origin(): + started = WorkflowStartedEvent() + yield started + with _framework_event_origin(): + in_progress = WorkflowStatusEvent(WorkflowRunState.IN_PROGRESS) + yield in_progress # Reset context for a new run if supported if reset_context: @@ -274,22 +272,34 @@ class Workflow(AFBaseModel): if isinstance(event, RequestInfoEvent) and not emitted_in_progress_pending and not saw_completed: emitted_in_progress_pending = True - yield WorkflowStatusEvent(WorkflowRunState.IN_PROGRESS_PENDING_REQUESTS) + with _framework_event_origin(): + pending_status = WorkflowStatusEvent(WorkflowRunState.IN_PROGRESS_PENDING_REQUESTS) + yield pending_status # Success path: emit a final status based on observed terminal signals if saw_completed: - yield WorkflowStatusEvent(WorkflowRunState.COMPLETED) + with _framework_event_origin(): + terminal_status = WorkflowStatusEvent(WorkflowRunState.COMPLETED) + yield terminal_status elif saw_request: - yield WorkflowStatusEvent(WorkflowRunState.IDLE_WITH_PENDING_REQUESTS) + with _framework_event_origin(): + terminal_status = WorkflowStatusEvent(WorkflowRunState.IDLE_WITH_PENDING_REQUESTS) + yield terminal_status else: - yield WorkflowStatusEvent(WorkflowRunState.IDLE) + with _framework_event_origin(): + terminal_status = WorkflowStatusEvent(WorkflowRunState.IDLE) + yield terminal_status workflow_tracer.add_workflow_event("workflow.completed") except Exception as e: # Surface structured failure details before propagating exception details = WorkflowErrorDetails.from_exception(e) - yield WorkflowFailedEvent(details) - yield WorkflowStatusEvent(WorkflowRunState.FAILED) + 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) raise diff --git a/python/packages/main/agent_framework/_workflow/_workflow_context.py b/python/packages/main/agent_framework/_workflow/_workflow_context.py index 219f164966..614b088348 100644 --- a/python/packages/main/agent_framework/_workflow/_workflow_context.py +++ b/python/packages/main/agent_framework/_workflow/_workflow_context.py @@ -1,10 +1,19 @@ # Copyright (c) Microsoft. All rights reserved. -from typing import Any, Generic, TypeVar +import logging +from typing import Any, Generic, TypeVar, cast, get_args from opentelemetry.propagate import inject -from ._events import WorkflowEvent +from ._events import ( + WorkflowEvent, + WorkflowEventSource, + WorkflowFailedEvent, + WorkflowLifecycleEvent, + WorkflowStartedEvent, + WorkflowStatusEvent, + WorkflowWarningEvent, +) from ._runner_context import Message, RunnerContext from ._shared_state import SharedState from ._telemetry import workflow_tracer @@ -12,6 +21,20 @@ from ._telemetry import workflow_tracer T_Out = TypeVar("T_Out") +logger = logging.getLogger(__name__) + + +_FRAMEWORK_LIFECYCLE_EVENT_TYPES: tuple[type[WorkflowEvent], ...] = cast( + tuple[type[WorkflowEvent], ...], + tuple(get_args(WorkflowLifecycleEvent)) + or ( + WorkflowStartedEvent, + WorkflowStatusEvent, + WorkflowFailedEvent, + ), +) + + class WorkflowContext(Generic[T_Out]): """Context for executors in a workflow. @@ -77,6 +100,16 @@ class WorkflowContext(Generic[T_Out]): async def add_event(self, event: WorkflowEvent) -> None: """Add an event to the workflow context.""" + if event.origin == WorkflowEventSource.EXECUTOR and isinstance(event, _FRAMEWORK_LIFECYCLE_EVENT_TYPES): + event_name = event.__class__.__name__ + warning_msg = ( + f"Executor '{self._executor_id}' attempted to emit {event_name}, " + "which is reserved for framework lifecycle notifications. The " + "event was ignored." + ) + logger.warning(warning_msg) + await self._runner_context.add_event(WorkflowWarningEvent(warning_msg)) + return await self._runner_context.add_event(event) async def get_shared_state(self, key: str) -> Any: diff --git a/python/packages/main/tests/workflow/test_runner.py b/python/packages/main/tests/workflow/test_runner.py index cfd01f1f08..722856134a 100644 --- a/python/packages/main/tests/workflow/test_runner.py +++ b/python/packages/main/tests/workflow/test_runner.py @@ -5,10 +5,19 @@ from dataclasses import dataclass import pytest -from agent_framework import Executor, WorkflowCompletedEvent, WorkflowContext, WorkflowEvent, handler +from agent_framework import ( + AgentExecutorResponse, + AgentRunResponse, + Executor, + WorkflowCompletedEvent, + WorkflowContext, + WorkflowEvent, + WorkflowEventSource, + handler, +) from agent_framework._workflow._edge import SingleEdgeGroup from agent_framework._workflow._runner import Runner -from agent_framework._workflow._runner_context import InProcRunnerContext, RunnerContext +from agent_framework._workflow._runner_context import InProcRunnerContext, Message, RunnerContext from agent_framework._workflow._shared_state import SharedState @@ -79,6 +88,7 @@ async def test_runner_run_until_convergence(): assert isinstance(event, WorkflowEvent) if isinstance(event, WorkflowCompletedEvent): result = event.data + assert event.origin is WorkflowEventSource.EXECUTOR assert result is not None and result == 10 @@ -148,3 +158,20 @@ async def test_runner_already_running(): pass await asyncio.gather(_run(), _run()) + + +async def test_runner_emits_runner_completion_for_agent_response_without_targets(): + ctx = InProcRunnerContext() + runner = Runner([], {}, SharedState(), ctx) + + await ctx.send_message( + Message( + data=AgentExecutorResponse("agent", AgentRunResponse()), + source_id="agent", + ) + ) + + events: list[WorkflowEvent] = [event async for event in runner.run_until_convergence()] + completions = [e for e in events if isinstance(e, WorkflowCompletedEvent)] + assert completions + assert all(e.origin is WorkflowEventSource.FRAMEWORK for e in completions) diff --git a/python/packages/main/tests/workflow/test_simple_sub_workflow.py b/python/packages/main/tests/workflow/test_simple_sub_workflow.py index 41309db05d..73585fafe9 100644 --- a/python/packages/main/tests/workflow/test_simple_sub_workflow.py +++ b/python/packages/main/tests/workflow/test_simple_sub_workflow.py @@ -3,8 +3,6 @@ import asyncio from dataclasses import dataclass -import pytest - from agent_framework import ( Executor, WorkflowBuilder, @@ -64,7 +62,6 @@ class SimpleParent(Executor): self.result = response -@pytest.mark.asyncio async def test_simple_sub_workflow(): """Test the simplest possible sub-workflow.""" # Create sub-workflow with dummy executor to satisfy validation diff --git a/python/packages/main/tests/workflow/test_sub_workflow.py b/python/packages/main/tests/workflow/test_sub_workflow.py index 7bcb72e54b..3fe8109cb8 100644 --- a/python/packages/main/tests/workflow/test_sub_workflow.py +++ b/python/packages/main/tests/workflow/test_sub_workflow.py @@ -4,7 +4,6 @@ import asyncio from dataclasses import dataclass from typing import Any -import pytest from pydantic import Field from agent_framework import ( @@ -120,7 +119,6 @@ class ParentOrchestrator(Executor): self.results.append(result) -@pytest.mark.asyncio async def test_basic_sub_workflow() -> None: """Test basic sub-workflow execution without interception.""" # Create sub-workflow @@ -185,7 +183,6 @@ async def test_basic_sub_workflow() -> None: assert parent.result.is_valid is True -@pytest.mark.asyncio async def test_sub_workflow_with_interception(): """Test sub-workflow with parent interception of requests.""" # Create sub-workflow @@ -248,7 +245,6 @@ async def test_sub_workflow_with_interception(): assert parent.results[0].reason == "Domain not approved" -@pytest.mark.asyncio async def test_conditional_forwarding() -> None: """Test conditional forwarding with RequestResponse.forward().""" @@ -326,7 +322,6 @@ async def test_conditional_forwarding() -> None: assert parent.result.is_valid is True -@pytest.mark.asyncio async def test_workflow_scoped_interception() -> None: """Test interception scoped to specific sub-workflows.""" diff --git a/python/packages/main/tests/workflow/test_tracing.py b/python/packages/main/tests/workflow/test_tracing.py index 5b280727c4..e0cf624850 100644 --- a/python/packages/main/tests/workflow/test_tracing.py +++ b/python/packages/main/tests/workflow/test_tracing.py @@ -273,7 +273,6 @@ async def test_trace_context_handling(tracing_enabled: Any, span_exporter: InMem assert processing_span.attributes.get("message.type") == "str" -@pytest.mark.asyncio async def test_trace_context_disabled_when_tracing_disabled() -> None: """Test that no trace context is added when tracing is disabled.""" # Tracing should be disabled by default @@ -299,7 +298,6 @@ async def test_trace_context_disabled_when_tracing_disabled() -> None: assert message.source_span_id is None -@pytest.mark.asyncio async def test_end_to_end_workflow_tracing(tracing_enabled: Any, span_exporter: InMemorySpanExporter) -> None: """Test end-to-end tracing including workflow build, execution, and span linking with fan-in edges.""" # Create executors for fan-in scenario @@ -424,7 +422,6 @@ async def test_end_to_end_workflow_tracing(tracing_enabled: Any, span_exporter: assert len(aggregator_span.links) >= 2, f"Expected at least 2 links, got {len(aggregator_span.links)}" -@pytest.mark.asyncio async def test_workflow_error_handling_in_tracing(tracing_enabled: Any, span_exporter: InMemorySpanExporter) -> None: """Test that workflow errors are properly recorded in traces.""" @@ -460,7 +457,6 @@ async def test_workflow_error_handling_in_tracing(tracing_enabled: Any, span_exp assert workflow_span.status.status_code.name == "ERROR" -@pytest.mark.asyncio async def test_message_trace_context_serialization() -> None: """Test that message trace context is properly serialized/deserialized.""" ctx = InProcRunnerContext() @@ -495,7 +491,6 @@ async def test_message_trace_context_serialization() -> None: assert restored_msg.source_span_ids == ["span123"] # Test new format -@pytest.mark.asyncio async def test_workflow_build_error_tracing(tracing_enabled: Any, span_exporter: InMemorySpanExporter) -> None: """Test that build errors are properly recorded in build spans.""" diff --git a/python/packages/main/tests/workflow/test_workflow.py b/python/packages/main/tests/workflow/test_workflow.py index e1b5956c3c..b644ba2cb4 100644 --- a/python/packages/main/tests/workflow/test_workflow.py +++ b/python/packages/main/tests/workflow/test_workflow.py @@ -231,7 +231,7 @@ async def test_fan_out(): events = await workflow.run(NumberMessage(data=0)) - # Each executor will emit two events: ExecutorInvokeEvent and ExecutorCompletedEvent + # Each executor will emit two events: ExecutorInvokedEvent and ExecutorCompletedEvent # executor_b will also emit a WorkflowCompletedEvent assert len(events) == 7 @@ -251,7 +251,7 @@ async def test_fan_out_multiple_completed_events(): events = await workflow.run(NumberMessage(data=0)) - # Each executor will emit two events: ExecutorInvokeEvent and ExecutorCompletedEvent + # Each executor will emit two events: ExecutorInvokedEvent and ExecutorCompletedEvent # executor_a and executor_b will also emit a WorkflowCompletedEvent assert len(events) == 8 @@ -276,7 +276,7 @@ async def test_fan_in(): events = await workflow.run(NumberMessage(data=0)) - # Each executor will emit two events: ExecutorInvokeEvent and ExecutorCompletedEvent + # Each executor will emit two events: ExecutorInvokedEvent and ExecutorCompletedEvent # aggregator will also emit a WorkflowCompletedEvent assert len(events) == 9 @@ -667,10 +667,10 @@ async def test_workflow_with_simple_cycle_and_exit_condition(): ) # Should complete when executor_a reaches its limit # Verify cycling occurred (should have events from both executors) - # Check for ExecutorInvokeEvent and ExecutorCompletedEvent types that have executor_id - from agent_framework import ExecutorCompletedEvent, ExecutorInvokeEvent + # Check for ExecutorInvokedEvent and ExecutorCompletedEvent types that have executor_id + from agent_framework import ExecutorCompletedEvent, ExecutorInvokedEvent - executor_events = [e for e in events if isinstance(e, (ExecutorInvokeEvent, ExecutorCompletedEvent))] + executor_events = [e for e in events if isinstance(e, (ExecutorInvokedEvent, ExecutorCompletedEvent))] executor_ids = {e.executor_id for e in executor_events} assert "exec_a" in executor_ids, "Should have events from executor A" assert "exec_b" in executor_ids, "Should have events from executor B" diff --git a/python/packages/main/tests/workflow/test_workflow_agent.py b/python/packages/main/tests/workflow/test_workflow_agent.py index a194940d27..145ac1c06a 100644 --- a/python/packages/main/tests/workflow/test_workflow_agent.py +++ b/python/packages/main/tests/workflow/test_workflow_agent.py @@ -76,7 +76,6 @@ class RequestingExecutor(Executor): class TestWorkflowAgent: """Test cases for WorkflowAgent end-to-end functionality.""" - @pytest.mark.asyncio async def test_end_to_end_basic_workflow(self): """Test basic end-to-end workflow execution with 2 executors emitting AgentRunEvent.""" # Create workflow with two executors @@ -117,7 +116,6 @@ class TestWorkflowAgent: assert "Step1: Hello World" in step1_text assert "Step2: Step1: Hello World" in step2_text - @pytest.mark.asyncio async def test_end_to_end_basic_workflow_streaming(self): """Test end-to-end workflow with streaming executor that emits AgentRunStreamingEvent.""" # Create a single streaming executor @@ -146,7 +144,6 @@ class TestWorkflowAgent: assert isinstance(second_content, TextContent) assert "Streaming2: Streaming1: Test input" in second_content.text - @pytest.mark.asyncio async def test_end_to_end_request_info_handling(self): """Test end-to-end workflow with RequestInfoEvent handling.""" # Create workflow with requesting executor -> request info executor (no cycle) diff --git a/python/packages/main/tests/workflow/test_workflow_context.py b/python/packages/main/tests/workflow/test_workflow_context.py new file mode 100644 index 0000000000..b78de9af67 --- /dev/null +++ b/python/packages/main/tests/workflow/test_workflow_context.py @@ -0,0 +1,64 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from typing import TYPE_CHECKING + +from agent_framework import ( + WorkflowCompletedEvent, + WorkflowContext, + WorkflowEvent, + WorkflowRunState, + WorkflowStatusEvent, +) + +if TYPE_CHECKING: + from _pytest.logging import LogCaptureFixture + + from agent_framework._workflow._runner_context import InProcRunnerContext + + +@asynccontextmanager +async def make_context( + executor_id: str = "exec", +) -> AsyncIterator[tuple[WorkflowContext[object], "InProcRunnerContext"]]: + from agent_framework._workflow._runner_context import InProcRunnerContext + from agent_framework._workflow._shared_state import SharedState + + runner_ctx = InProcRunnerContext() + shared_state = SharedState() + workflow_ctx: WorkflowContext[object] = WorkflowContext( + executor_id, + ["source"], + shared_state, + runner_ctx, + ) + try: + yield workflow_ctx, runner_ctx + finally: + await asyncio.sleep(0) + + +async def test_executor_cannot_emit_framework_lifecycle_event(caplog: "LogCaptureFixture") -> None: + async with make_context() as (ctx, runner_ctx): + caplog.clear() + with caplog.at_level("WARNING"): + await ctx.add_event(WorkflowStatusEvent(state=WorkflowRunState.IN_PROGRESS)) + + events: list[WorkflowEvent] = await runner_ctx.drain_events() + assert len(events) == 1 + assert type(events[0]).__name__ == "WorkflowWarningEvent" + data = getattr(events[0], "data", None) + assert isinstance(data, str) + assert "reserved for framework lifecycle notifications" in data + assert any("attempted to emit WorkflowStatusEvent" in message for message in list(caplog.messages)) + + +async def test_executor_emits_normal_event() -> None: + async with make_context() as (ctx, runner_ctx): + await ctx.add_event(WorkflowCompletedEvent("done")) + + events: list[WorkflowEvent] = await runner_ctx.drain_events() + assert len(events) == 1 + assert isinstance(events[0], WorkflowCompletedEvent) diff --git a/python/packages/main/tests/workflow/test_workflow_states.py b/python/packages/main/tests/workflow/test_workflow_states.py index 51e16d5afb..c419682c7c 100644 --- a/python/packages/main/tests/workflow/test_workflow_states.py +++ b/python/packages/main/tests/workflow/test_workflow_states.py @@ -14,9 +14,11 @@ from agent_framework import ( WorkflowBuilder, WorkflowCompletedEvent, WorkflowContext, + WorkflowEventSource, WorkflowFailedEvent, WorkflowRunResult, WorkflowRunState, + WorkflowStartedEvent, WorkflowStatusEvent, handler, ) @@ -41,9 +43,12 @@ async def test_executor_failed_and_workflow_failed_events_streaming(): events.append(ev) # Workflow-level failure and FAILED status should be surfaced - assert any(isinstance(e, WorkflowFailedEvent) for e in events) + failed_events = [e for e in events if isinstance(e, WorkflowFailedEvent)] + assert failed_events + assert all(e.origin is WorkflowEventSource.FRAMEWORK for e in failed_events) status = [e for e in events if isinstance(e, WorkflowStatusEvent)] assert status and status[-1].state == WorkflowRunState.FAILED + assert all(e.origin is WorkflowEventSource.FRAMEWORK for e in status) async def test_executor_failed_event_emitted_on_direct_execute(): @@ -58,7 +63,9 @@ async def test_executor_failed_event_emitted_on_direct_execute(): with pytest.raises(RuntimeError, match="boom"): await failing.execute(0, wf_ctx) drained = await ctx.drain_events() - assert any(isinstance(e, ExecutorFailedEvent) for e in drained) + failed = [e for e in drained if isinstance(e, ExecutorFailedEvent)] + assert failed + assert all(e.origin is WorkflowEventSource.FRAMEWORK for e in failed) class Requester(Executor): @@ -99,6 +106,19 @@ async def test_completed_status_streaming(): # Last status should be COMPLETED status = [e for e in events if isinstance(e, WorkflowStatusEvent)] assert status and status[-1].state == WorkflowRunState.COMPLETED + assert all(e.origin is WorkflowEventSource.FRAMEWORK for e in status) + + +async def test_started_and_completed_event_origins(): + c = Completer(id="c-origin") + wf = WorkflowBuilder().set_start_executor(c).build() + events = [ev async for ev in wf.run_stream("payload")] + + started = next(e for e in events if isinstance(e, WorkflowStartedEvent)) + assert started.origin is WorkflowEventSource.FRAMEWORK + + completed = next(e for e in events if isinstance(e, WorkflowCompletedEvent)) + assert completed.origin is WorkflowEventSource.EXECUTOR async def test_non_streaming_final_state_helpers(): diff --git a/python/samples/getting_started/workflow/_start-here/step3_streaming.py b/python/samples/getting_started/workflow/_start-here/step3_streaming.py index 423fcd53da..f5d29e178a 100644 --- a/python/samples/getting_started/workflow/_start-here/step3_streaming.py +++ b/python/samples/getting_started/workflow/_start-here/step3_streaming.py @@ -29,7 +29,8 @@ Purpose: Show how to wrap chat agents created by AzureChatClient inside workflow executors, wire them with WorkflowBuilder, and consume streaming events from the workflow. Demonstrate the @handler pattern with typed inputs and typed WorkflowContext[T] outputs, and finish by emitting a WorkflowCompletedEvent from the terminal node while printing -intermediate events for observability. +intermediate events for observability. The streaming loop also surfaces WorkflowEvent.origin so you can +distinguish runner-generated lifecycle events from executor-generated data-plane events. Prerequisites: - Azure OpenAI configured for AzureChatClient with required environment variables. @@ -123,36 +124,42 @@ async def main(): ChatMessage(role="user", text="Create a slogan for a new electric SUV that is affordable and fun to drive.") ): if isinstance(event, WorkflowStatusEvent): + prefix = f"State ({event.origin.value}): " if event.state == WorkflowRunState.IN_PROGRESS: - print("State: IN_PROGRESS") + print(prefix + "IN_PROGRESS") elif event.state == WorkflowRunState.COMPLETED: - print("State: COMPLETED") + print(prefix + "COMPLETED") elif event.state == WorkflowRunState.IN_PROGRESS_PENDING_REQUESTS: - print("State: IN_PROGRESS_PENDING_REQUESTS (requests in flight)") + print(prefix + "IN_PROGRESS_PENDING_REQUESTS (requests in flight)") elif event.state == WorkflowRunState.IDLE: - print("State: IDLE (no active work)") + print(prefix + "IDLE (no active work)") elif event.state == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS: - print("State: IDLE_WITH_PENDING_REQUESTS (prompt user or UI now)") + print(prefix + "IDLE_WITH_PENDING_REQUESTS (prompt user or UI now)") else: - print(f"State: {event.state}") + print(prefix + str(event.state)) + elif isinstance(event, WorkflowCompletedEvent): + print(f"Workflow completed ({event.origin.value}): {event.data}") elif isinstance(event, ExecutorFailedEvent): - print(f"Executor failed: {event.executor_id} {event.details.error_type}: {event.details.message}") + print( + f"Executor failed ({event.origin.value}): " + f"{event.executor_id} {event.details.error_type}: {event.details.message}" + ) elif isinstance(event, WorkflowFailedEvent): details = event.details - print(f"Workflow failed: {details.error_type}: {details.message}") + print(f"Workflow failed ({event.origin.value}): {details.error_type}: {details.message}") else: - print(event) + print(f"{event.__class__.__name__} ({event.origin.value}): {event}") """ Sample Output: - State: IN_PROGRESS - ExecutorInvokeEvent(executor_id=writer) - ExecutorCompletedEvent(executor_id=writer) - ExecutorInvokeEvent(executor_id=reviewer) - WorkflowCompletedEvent(data=Drive the Future. Affordable Adventure, Electrified.) - ExecutorCompletedEvent(executor_id=reviewer) - State: COMPLETED + State (RUNNER): IN_PROGRESS + ExecutorInvokeEvent (RUNNER): ExecutorInvokeEvent(executor_id=writer) + ExecutorCompletedEvent (RUNNER): ExecutorCompletedEvent(executor_id=writer) + ExecutorInvokeEvent (RUNNER): ExecutorInvokeEvent(executor_id=reviewer) + Workflow completed (EXECUTOR): Drive the Future. Affordable Adventure, Electrified. + ExecutorCompletedEvent (RUNNER): ExecutorCompletedEvent(executor_id=reviewer) + State (RUNNER): COMPLETED """ diff --git a/python/samples/getting_started/workflow/checkpoint/checkpoint_with_resume.py b/python/samples/getting_started/workflow/checkpoint/checkpoint_with_resume.py index 839b3ad12c..42dbb789fd 100644 --- a/python/samples/getting_started/workflow/checkpoint/checkpoint_with_resume.py +++ b/python/samples/getting_started/workflow/checkpoint/checkpoint_with_resume.py @@ -275,15 +275,15 @@ async def main(): Running workflow with initial message... UpperCaseExecutor: 'hello world' -> 'HELLO WORLD' - Event: ExecutorInvokeEvent(executor_id=upper_case_executor) + Event: ExecutorInvokedEvent(executor_id=upper_case_executor) Event: ExecutorCompletedEvent(executor_id=upper_case_executor) ReverseTextExecutor: 'HELLO WORLD' -> 'DLROW OLLEH' - Event: ExecutorInvokeEvent(executor_id=reverse_text_executor) + Event: ExecutorInvokedEvent(executor_id=reverse_text_executor) Event: ExecutorCompletedEvent(executor_id=reverse_text_executor) LowerAgent (shared_state): original_input='hello world', upper_output='HELLO WORLD' - Event: ExecutorInvokeEvent(executor_id=submit_lower) - Event: ExecutorInvokeEvent(executor_id=lower_agent) - Event: ExecutorInvokeEvent(executor_id=finalize) + Event: ExecutorInvokedEvent(executor_id=submit_lower) + Event: ExecutorInvokedEvent(executor_id=lower_agent) + Event: ExecutorInvokedEvent(executor_id=finalize) Event: WorkflowCompletedEvent(data=dlrow olleh) Checkpoint summary: @@ -300,9 +300,9 @@ async def main(): Resuming from checkpoint: a78c345a-e5d9-45ba-82c0-cb725452d91b LowerAgent (shared_state): original_input='hello world', upper_output='HELLO WORLD' - Resumed Event: ExecutorInvokeEvent(executor_id=submit_lower) - Resumed Event: ExecutorInvokeEvent(executor_id=lower_agent) - Resumed Event: ExecutorInvokeEvent(executor_id=finalize) + Resumed Event: ExecutorInvokedEvent(executor_id=submit_lower) + Resumed Event: ExecutorInvokedEvent(executor_id=lower_agent) + Resumed Event: ExecutorInvokedEvent(executor_id=finalize) Resumed Event: WorkflowCompletedEvent(data=dlrow olleh) """ # noqa: E501 diff --git a/python/samples/getting_started/workflow/control-flow/sequential_streaming.py b/python/samples/getting_started/workflow/control-flow/sequential_streaming.py index bf194ba901..fe6056cfa6 100644 --- a/python/samples/getting_started/workflow/control-flow/sequential_streaming.py +++ b/python/samples/getting_started/workflow/control-flow/sequential_streaming.py @@ -13,7 +13,7 @@ The second reverses the text and completes the workflow. Events are printed as t Purpose: Show how to declare executors with the @executor decorator, connect them with WorkflowBuilder, pass intermediate values using ctx.send_message, and signal completion with ctx.add_event by emitting a -WorkflowCompletedEvent. Demonstrate how streaming exposes ExecutorInvokeEvent and WorkflowCompletedEvent +WorkflowCompletedEvent. Demonstrate how streaming exposes ExecutorInvokedEvent and WorkflowCompletedEvent for observability. Prerequisites: @@ -72,9 +72,9 @@ async def main(): """ Sample Output: - Event: ExecutorInvokeEvent(executor_id=upper_case_executor) + Event: ExecutorInvokedEvent(executor_id=upper_case_executor) Event: ExecutorCompletedEvent(executor_id=upper_case_executor) - Event: ExecutorInvokeEvent(executor_id=reverse_text_executor) + Event: ExecutorInvokedEvent(executor_id=reverse_text_executor) Event: WorkflowCompletedEvent(data=DLROW OLLEH) Event: ExecutorCompletedEvent(executor_id=reverse_text_executor) Workflow completed with result: DLROW OLLEH