diff --git a/python/packages/main/agent_framework/workflow/__init__.py b/python/packages/main/agent_framework/workflow/__init__.py index f74b1f1ff3..23f28d1dc0 100644 --- a/python/packages/main/agent_framework/workflow/__init__.py +++ b/python/packages/main/agent_framework/workflow/__init__.py @@ -19,7 +19,7 @@ _IMPORTS = [ "WorkflowEvent", "WorkflowStartedEvent", "AgentRunEvent", - "AgentRunStreamingEvent", + "AgentRunUpdateEvent", "handler", "AgentExecutor", "MagenticAgentDeltaEvent", @@ -35,6 +35,7 @@ _IMPORTS = [ "RequestInfoMessage", "WorkflowRunResult", "Workflow", + "WorkflowAgent", "WorkflowViz", "FileCheckpointStorage", "InMemoryCheckpointStorage", diff --git a/python/packages/main/agent_framework/workflow/__init__.pyi b/python/packages/main/agent_framework/workflow/__init__.pyi index 5f0e2f7b21..063921b69c 100644 --- a/python/packages/main/agent_framework/workflow/__init__.pyi +++ b/python/packages/main/agent_framework/workflow/__init__.pyi @@ -5,7 +5,7 @@ from agent_framework_workflow import ( AgentExecutorRequest, AgentExecutorResponse, AgentRunEvent, - AgentRunStreamingEvent, + AgentRunUpdateEvent, Case, CheckpointStorage, Default, @@ -31,6 +31,7 @@ from agent_framework_workflow import ( SubWorkflowRequestInfo, SubWorkflowResponse, Workflow, + WorkflowAgent, WorkflowBuilder, WorkflowCheckpoint, WorkflowCompletedEvent, @@ -50,7 +51,7 @@ __all__ = [ "AgentExecutorRequest", "AgentExecutorResponse", "AgentRunEvent", - "AgentRunStreamingEvent", + "AgentRunUpdateEvent", "Case", "CheckpointStorage", "Default", @@ -69,7 +70,6 @@ __all__ = [ "MagenticProgressLedger", "MagenticProgressLedgerItem", "RequestInfoEvent", - "RequestInfoEvent", "RequestInfoExecutor", "RequestInfoMessage", "RequestResponse", @@ -77,6 +77,7 @@ __all__ = [ "SubWorkflowRequestInfo", "SubWorkflowResponse", "Workflow", + "WorkflowAgent", "WorkflowBuilder", "WorkflowCheckpoint", "WorkflowCompletedEvent", diff --git a/python/packages/workflow/agent_framework_workflow/__init__.py b/python/packages/workflow/agent_framework_workflow/__init__.py index 0214193061..ee4bc5d246 100644 --- a/python/packages/workflow/agent_framework_workflow/__init__.py +++ b/python/packages/workflow/agent_framework_workflow/__init__.py @@ -2,6 +2,7 @@ import importlib.metadata +from ._agent import WorkflowAgent from ._checkpoint import ( CheckpointStorage, FileCheckpointStorage, @@ -14,7 +15,7 @@ from ._const import ( from ._edge import Case, Default from ._events import ( AgentRunEvent, - AgentRunStreamingEvent, + AgentRunUpdateEvent, ExecutorCompletedEvent, ExecutorEvent, ExecutorInvokeEvent, @@ -88,7 +89,7 @@ __all__ = [ "AgentExecutorRequest", "AgentExecutorResponse", "AgentRunEvent", - "AgentRunStreamingEvent", + "AgentRunUpdateEvent", "Case", "CheckpointStorage", "Default", @@ -132,6 +133,7 @@ __all__ = [ "TypeCompatibilityError", "ValidationTypeEnum", "Workflow", + "WorkflowAgent", "WorkflowBuilder", "WorkflowCheckpoint", "WorkflowCompletedEvent", @@ -155,3 +157,5 @@ import contextlib with contextlib.suppress(AttributeError, TypeError, ValueError): # Rebuild WorkflowExecutor to resolve Workflow forward reference WorkflowExecutor.model_rebuild() + # Rebuild WorkflowAgent to resolve Workflow forward reference + WorkflowAgent.model_rebuild() diff --git a/python/packages/workflow/agent_framework_workflow/_agent.py b/python/packages/workflow/agent_framework_workflow/_agent.py new file mode 100644 index 0000000000..415f526948 --- /dev/null +++ b/python/packages/workflow/agent_framework_workflow/_agent.py @@ -0,0 +1,444 @@ +# Copyright (c) Microsoft. All rights reserved. + +import logging +import uuid +from collections.abc import AsyncIterable +from datetime import datetime +from typing import TYPE_CHECKING, Any, ClassVar, TypedDict, cast + +from agent_framework import ( + AgentBase, + AgentRunResponse, + AgentRunResponseUpdate, + AgentThread, + ChatMessage, + ChatRole, + FunctionCallContent, + FunctionResultContent, + TextContent, + UsageDetails, +) +from agent_framework._pydantic import AFBaseModel +from agent_framework.exceptions import AgentExecutionException +from pydantic import Field + +from ._events import ( + AgentRunUpdateEvent, + RequestInfoEvent, + WorkflowEvent, +) + +if TYPE_CHECKING: + from ._workflow import Workflow + +logger = logging.getLogger(__name__) + + +class WorkflowAgent(AgentBase): + """An `AIAgent` subclass that wraps a workflow and exposes it as an agent.""" + + # Class variable for the request info function name + REQUEST_INFO_FUNCTION_NAME: ClassVar[str] = "request_info" + + class RequestInfoFunctionArgs(AFBaseModel): + request_id: str + data: Any + + workflow: "Workflow" = Field(description="The workflow wrapped as an agent") + pending_requests: dict[str, RequestInfoEvent] = Field( + default_factory=dict, description="Pending request info events" + ) + + def __init__( + self, + workflow: "Workflow", + *, + id: str | None = None, + name: str | None = None, + description: str | None = None, + **kwargs: Any, + ) -> None: + """Initialize the WorkflowAgent. + + Args: + workflow: The workflow to wrap as an agent. + id: Unique identifier for the agent. If None, will be generated. + name: Optional name for the agent. + description: Optional description of the agent. + **kwargs: Additional keyword arguments passed to AgentBase. + """ + if id is None: + id = f"WorkflowAgent_{uuid.uuid4().hex[:8]}" + # Initialize with standard AgentBase parameters first + kwargs["workflow"] = workflow + + # Validate the workflow's start executor can handle agent-facing message inputs + start_executor = workflow.get_start_executor() + if start_executor is None: + raise ValueError("Workflow's start executor is not defined.") + + if not start_executor.can_handle_type(list[ChatMessage]): + raise ValueError("Workflow's start executor cannot handle list[ChatMessage]") + + super().__init__(id=id, name=name, description=description, **kwargs) + + async def run( + self, + messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None, + *, + thread: AgentThread | None = None, + **kwargs: Any, + ) -> AgentRunResponse: + """Get a response from the workflow agent (non-streaming). + + This method collects all streaming updates and merges them into a single response. + + Args: + messages: The message(s) to send to the workflow. + thread: The conversation thread. If None, a new thread will be created. + **kwargs: Additional keyword arguments. + + Returns: + The final workflow response as an AgentRunResponse. + """ + # Collect all streaming updates + response_updates: list[AgentRunResponseUpdate] = [] + input_messages = self._normalize_messages(messages) + thread = thread or self.get_new_thread() + response_id = str(uuid.uuid4()) + + async for update in self._run_streaming_impl(input_messages, response_id): + response_updates.append(update) + + # Convert updates to final response. + response = self.merge_updates(response_updates, response_id) + + # Notify thread of new messages (both input and response messages) + await self._notify_thread_of_new_messages(thread, input_messages) + await self._notify_thread_of_new_messages(thread, response.messages) + + return response + + async def run_streaming( + self, + messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None, + *, + thread: AgentThread | None = None, + **kwargs: Any, + ) -> AsyncIterable[AgentRunResponseUpdate]: + """Stream response updates from the workflow agent. + + Args: + messages: The message(s) to send to the workflow. + thread: The conversation thread. If None, a new thread will be created. + **kwargs: Additional keyword arguments. + + Yields: + AgentRunResponseUpdate objects representing the workflow execution progress. + """ + input_messages = self._normalize_messages(messages) + thread = thread or self.get_new_thread() + response_updates: list[AgentRunResponseUpdate] = [] + response_id = str(uuid.uuid4()) + + async for update in self._run_streaming_impl(input_messages, response_id): + response_updates.append(update) + yield update + + # Convert updates to final response. + response = self.merge_updates(response_updates, response_id) + + # Notify thread of new messages (both input and response messages) + await self._notify_thread_of_new_messages(thread, input_messages) + await self._notify_thread_of_new_messages(thread, response.messages) + + async def _run_streaming_impl( + self, + input_messages: list[ChatMessage], + response_id: str, + ) -> AsyncIterable[AgentRunResponseUpdate]: + """Internal implementation of streaming execution. + + Args: + input_messages: Normalized input messages to process. + response_id: The unique response ID for this workflow execution. + + Yields: + AgentRunResponseUpdate objects representing the workflow execution progress. + """ + # Determine the event stream based on whether we have function responses + if bool(self.pending_requests): + # This is a continuation - use send_responses_streaming to send function responses back + logger.info(f"Continuing workflow to address {len(self.pending_requests)} requests") + + # Extract function responses from input messages, and ensure that + # only function responses are present in messages if there is any + # pending request. + function_responses = self._extract_function_responses(input_messages) + + # Pop pending requests if fulfilled. + for request_id in list(self.pending_requests.keys()): + if request_id in function_responses: + self.pending_requests.pop(request_id) + + # NOTE: It is possible that some pending requests are not fulfilled, + # and we will let the workflow to handle this -- the agent does not + # have an opinion on this. + event_stream = self.workflow.send_responses_streaming(function_responses) + else: + # Execute workflow with streaming (initial run or no function responses) + # Pass the new input messages directly to the workflow + event_stream = self.workflow.run_streaming(input_messages) + + # Process events from the stream + async for event in event_stream: + # Convert workflow event to agent update + update = self._convert_workflow_event_to_agent_update(response_id, event) + if update: + yield update + + def _normalize_messages( + self, + messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None, + ) -> list[ChatMessage]: + """Normalize input messages to a list of ChatMessage objects.""" + if messages is None: + return [] + + if isinstance(messages, str): + return [ChatMessage(role=ChatRole.USER, contents=[TextContent(text=messages)])] + + if isinstance(messages, ChatMessage): + return [messages] + + normalized = [] + for msg in messages: + if isinstance(msg, str): + normalized.append(ChatMessage(role=ChatRole.USER, contents=[TextContent(text=msg)])) + elif isinstance(msg, ChatMessage): + normalized.append(msg) + return normalized + + def _convert_workflow_event_to_agent_update( + self, + response_id: str, + event: WorkflowEvent, + ) -> AgentRunResponseUpdate | None: + """Convert a workflow event to an AgentRunResponseUpdate. + + Only AgentRunUpdateEvent and RequestInfoEvent are processed and the rest + are not relevant. Returns None if the event is not relevant. + """ + match event: + case AgentRunUpdateEvent(data=update): + # Direct pass-through of update in an agent streaming event + if update: + return cast(AgentRunResponseUpdate, update) + return None + + case RequestInfoEvent(request_id=request_id): + # Store the pending request for later correlation + self.pending_requests[request_id] = event + + # Convert to function call content + # TODO(ekzhu): update this to FunctionApprovalRequestContent + # monitor: https://github.com/microsoft/agent-framework/issues/285 + function_call = FunctionCallContent( + call_id=request_id, + name=self.REQUEST_INFO_FUNCTION_NAME, + arguments=self.RequestInfoFunctionArgs(request_id=request_id, data=event.data).model_dump(), + ) + return AgentRunResponseUpdate( + contents=[function_call], + role=ChatRole.ASSISTANT, + author_name=self.name, + response_id=response_id, + message_id=str(uuid.uuid4()), + created_at=datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%fZ"), + ) + # We only care about the above two events and discard the rest. + return None + + def _extract_function_responses(self, input_messages: list[ChatMessage]) -> dict[str, Any]: + """Extract function responses from input messages.""" + function_responses: dict[str, Any] = {} + for message in input_messages: + for content in message.contents: + # TODO(ekzhu): update this to FunctionApprovalResponseContent + # monitor: https://github.com/microsoft/agent-framework/issues/285 + if isinstance(content, FunctionResultContent): + request_id = content.call_id + # Check if we have a pending request for this call_id + if request_id in self.pending_requests: + response_data = content.result if hasattr(content, "result") else str(content) + function_responses[request_id] = response_data + elif bool(self.pending_requests): + # Function result for unknown request when we have pending requests - this is an error + raise AgentExecutionException( + "Only FunctionResultContent for pending requests is allowed in input messages " + "when there are pending requests." + ) + else: + if bool(self.pending_requests): + # Non-function content when we have pending requests - this is an error + raise AgentExecutionException( + "Only FunctionResultContent is allowed in input messages when there are pending requests." + ) + return function_responses + + class _ResponseState(TypedDict): + """State for grouping response updates by message_id.""" + + by_msg: dict[str, list[AgentRunResponseUpdate]] + dangling: list[AgentRunResponseUpdate] + + @staticmethod + def merge_updates(updates: list[AgentRunResponseUpdate], response_id: str) -> AgentRunResponse: + """Merge streaming updates into a single AgentRunResponse. + + Behavior: + - Group updates by response_id; within each response_id, group by message_id and keep a dangling bucket for + updates without message_id. + - Convert each group (per message and dangling) into an intermediate AgentRunResponse via + AgentRunResponse.from_agent_run_response_updates, then sort by created_at and merge. + - Append messages from updates without any response_id at the end (global dangling), while aggregating metadata. + + Args: + updates: The list of AgentRunResponseUpdate objects to merge. + response_id: The response identifier to set on the returned AgentRunResponse. + + Returns: + An AgentRunResponse with messages in processing order and aggregated metadata. + """ + # PHASE 1: GROUP UPDATES BY RESPONSE_ID AND MESSAGE_ID + states: dict[str, WorkflowAgent._ResponseState] = {} + global_dangling: list[AgentRunResponseUpdate] = [] + + for u in updates: + if u.response_id: + state = states.setdefault(u.response_id, {"by_msg": {}, "dangling": []}) + by_msg = state["by_msg"] + dangling = state["dangling"] + if u.message_id: + by_msg.setdefault(u.message_id, []).append(u) + else: + dangling.append(u) + else: + global_dangling.append(u) + + # HELPER FUNCTIONS + def _parse_dt(value: str | None) -> tuple[int, datetime | str | None]: + if not value: + return (1, None) + v = value + if v.endswith("Z"): + v = v[:-1] + "+00:00" + try: + return (0, datetime.fromisoformat(v)) + except Exception: + return (0, v) + + def _sum_usage(a: UsageDetails | None, b: UsageDetails | None) -> UsageDetails | None: + if a is None: + return b + if b is None: + return a + return a + b + + def _merge_responses(current: AgentRunResponse | None, incoming: AgentRunResponse) -> AgentRunResponse: + if current is None: + return incoming + raw_list: list[object] = [] + if current.raw_representation is not None: + if isinstance(current.raw_representation, list): + raw_list.extend(current.raw_representation) + else: + raw_list.append(current.raw_representation) + if incoming.raw_representation is not None: + if isinstance(incoming.raw_representation, list): + raw_list.extend(incoming.raw_representation) + else: + raw_list.append(incoming.raw_representation) + return AgentRunResponse( + messages=(current.messages or []) + (incoming.messages or []), + response_id=current.response_id or incoming.response_id, + created_at=incoming.created_at or current.created_at, + usage_details=_sum_usage(current.usage_details, incoming.usage_details), + raw_representation=raw_list if raw_list else None, + additional_properties=incoming.additional_properties or current.additional_properties, + ) + + # PHASE 2: CONVERT GROUPED UPDATES TO RESPONSES AND MERGE + final_messages: list[ChatMessage] = [] + merged_usage: UsageDetails | None = None + latest_created_at: str | None = None + merged_additional_properties: dict[str, Any] | None = None + raw_representations: list[object] = [] + + for grouped_response_id in states: + state = states[grouped_response_id] + by_msg = state["by_msg"] + dangling = state["dangling"] + + per_message_responses: list[AgentRunResponse] = [] + for _, msg_updates in by_msg.items(): + if msg_updates: + per_message_responses.append(AgentRunResponse.from_agent_run_response_updates(msg_updates)) + if dangling: + per_message_responses.append(AgentRunResponse.from_agent_run_response_updates(dangling)) + + per_message_responses.sort(key=lambda r: _parse_dt(r.created_at)) + + aggregated: AgentRunResponse | None = None + for resp in per_message_responses: + if resp.response_id and grouped_response_id and resp.response_id != grouped_response_id: + resp.response_id = grouped_response_id + aggregated = _merge_responses(aggregated, resp) + + if aggregated: + final_messages.extend(aggregated.messages) + if aggregated.usage_details: + merged_usage = _sum_usage(merged_usage, aggregated.usage_details) + if aggregated.created_at and ( + not latest_created_at or _parse_dt(aggregated.created_at) > _parse_dt(latest_created_at) + ): + latest_created_at = aggregated.created_at + if aggregated.additional_properties: + if merged_additional_properties is None: + merged_additional_properties = {} + merged_additional_properties.update(aggregated.additional_properties) + if aggregated.raw_representation: + if isinstance(aggregated.raw_representation, list): + raw_representations.extend(aggregated.raw_representation) + else: + raw_representations.append(aggregated.raw_representation) + + # PHASE 3: HANDLE GLOBAL DANGLING UPDATES (NO RESPONSE_ID) + if global_dangling: + flattened = AgentRunResponse.from_agent_run_response_updates(global_dangling) + final_messages.extend(flattened.messages) + if flattened.usage_details: + merged_usage = _sum_usage(merged_usage, flattened.usage_details) + if flattened.created_at and ( + not latest_created_at or _parse_dt(flattened.created_at) > _parse_dt(latest_created_at) + ): + latest_created_at = flattened.created_at + if flattened.additional_properties: + if merged_additional_properties is None: + merged_additional_properties = {} + merged_additional_properties.update(flattened.additional_properties) + if flattened.raw_representation: + if isinstance(flattened.raw_representation, list): + raw_representations.extend(flattened.raw_representation) + else: + raw_representations.append(flattened.raw_representation) + + # PHASE 4: CONSTRUCT FINAL RESPONSE WITH INPUT RESPONSE_ID + return AgentRunResponse( + messages=final_messages, + response_id=response_id, + created_at=latest_created_at, + usage_details=merged_usage, + raw_representation=raw_representations if raw_representations else None, + additional_properties=merged_additional_properties, + ) diff --git a/python/packages/workflow/agent_framework_workflow/_events.py b/python/packages/workflow/agent_framework_workflow/_events.py index f4a67036ff..67075a617f 100644 --- a/python/packages/workflow/agent_framework_workflow/_events.py +++ b/python/packages/workflow/agent_framework_workflow/_events.py @@ -119,7 +119,7 @@ class ExecutorCompletedEvent(ExecutorEvent): return f"{self.__class__.__name__}(executor_id={self.executor_id})" -class AgentRunStreamingEvent(ExecutorEvent): +class AgentRunUpdateEvent(ExecutorEvent): """Event triggered when an agent is streaming messages.""" def __init__(self, executor_id: str, data: AgentRunResponseUpdate | None = None): diff --git a/python/packages/workflow/agent_framework_workflow/_executor.py b/python/packages/workflow/agent_framework_workflow/_executor.py index 19ccbe6224..50f54f5489 100644 --- a/python/packages/workflow/agent_framework_workflow/_executor.py +++ b/python/packages/workflow/agent_framework_workflow/_executor.py @@ -18,7 +18,7 @@ from pydantic import Field from ._events import ( AgentRunEvent, - AgentRunStreamingEvent, + AgentRunUpdateEvent, ExecutorCompletedEvent, ExecutorInvokeEvent, RequestInfoEvent, @@ -39,7 +39,7 @@ class Executor(AFBaseModel): min_length=1, description="Unique identifier for the executor", ) - type: str = Field(default="", description="The type of executor, corresponding to the class name") + type_: str = Field(default="", alias="type", description="The type of executor, corresponding to the class name") def __init__(self, id: str | None = None, **kwargs: Any) -> None: """Initialize the executor with a unique identifier. @@ -52,8 +52,8 @@ class Executor(AFBaseModel): executor_id = f"{self.__class__.__name__}/{uuid.uuid4()}" if id is None else id kwargs.update({"id": executor_id}) - if "type" not in kwargs: - kwargs["type"] = self.__class__.__name__ + if "type" not in kwargs and "type_" not in kwargs: + kwargs["type_"] = self.__class__.__name__ super().__init__(**kwargs) @@ -254,6 +254,17 @@ class Executor(AFBaseModel): """ return any(is_instance_of(message, message_type) for message_type in self._handlers) + def can_handle_type(self, message_type: type[Any]) -> bool: + """Check if the executor can handle a given message type. + + Args: + message_type: The message type to check. + + Returns: + True if the executor can handle the message type, False otherwise. + """ + return message_type in self._handlers + # endregion: Executor @@ -783,7 +794,7 @@ class AgentExecutor(Executor): thread=self._agent_thread, ): updates.append(update) - await ctx.add_event(AgentRunStreamingEvent(self.id, update)) + await ctx.add_event(AgentRunUpdateEvent(self.id, update)) response = AgentRunResponse.from_agent_run_response_updates(updates) else: response = await self._agent.run( diff --git a/python/packages/workflow/agent_framework_workflow/_telemetry.py b/python/packages/workflow/agent_framework_workflow/_telemetry.py index fad5233a7f..f90d439665 100644 --- a/python/packages/workflow/agent_framework_workflow/_telemetry.py +++ b/python/packages/workflow/agent_framework_workflow/_telemetry.py @@ -173,7 +173,7 @@ class WorkflowTracer: if span and span.is_recording(): span.set_attributes({ "workflow.id": workflow.id, - "workflow.definition": workflow.model_dump_json(), + "workflow.definition": workflow.model_dump_json(by_alias=True), }) def add_build_event(self, event_name: str, attributes: Attributes | None = None) -> None: diff --git a/python/packages/workflow/agent_framework_workflow/_workflow.py b/python/packages/workflow/agent_framework_workflow/_workflow.py index 683697218d..d013478e39 100644 --- a/python/packages/workflow/agent_framework_workflow/_workflow.py +++ b/python/packages/workflow/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 +from typing import TYPE_CHECKING, Any from agent_framework._pydantic import AFBaseModel from pydantic import Field @@ -39,6 +39,9 @@ else: logger = logging.getLogger(__name__) +if TYPE_CHECKING: # Avoid runtime import cycles; enables proper type checking of as_agent return type + from ._agent import WorkflowAgent + class WorkflowRunResult(list[WorkflowEvent]): """A list of events generated during the workflow execution in non-streaming mode.""" @@ -534,6 +537,20 @@ class Workflow(AFBaseModel): ) ) + def as_agent(self, name: str | None = None) -> "WorkflowAgent": + """Create a WorkflowAgent that wraps this workflow. + + Args: + name: Optional name for the agent. If None, a default name will be generated. + + Returns: + A WorkflowAgent instance that wraps this workflow. + """ + # Import here to avoid circular imports + from ._agent import WorkflowAgent + + return WorkflowAgent(workflow=self, name=name) + # region WorkflowBuilder diff --git a/python/packages/workflow/tests/test_serialization.py b/python/packages/workflow/tests/test_serialization.py index 46de4d3f04..e3b0177c57 100644 --- a/python/packages/workflow/tests/test_serialization.py +++ b/python/packages/workflow/tests/test_serialization.py @@ -47,7 +47,7 @@ class TestSerializationWorkflowClasses: executor = SampleExecutor(id="test-executor") # Test model_dump - data = executor.model_dump() + data = executor.model_dump(by_alias=True) assert data["id"] == "test-executor" # Test type field @@ -55,7 +55,7 @@ class TestSerializationWorkflowClasses: assert data["type"] == "SampleExecutor", f"Expected type 'SampleExecutor', got {data['type']}" # Test model_dump_json - json_str = executor.model_dump_json() + json_str = executor.model_dump_json(by_alias=True) parsed = json.loads(json_str) assert parsed["id"] == "test-executor" @@ -124,7 +124,7 @@ class TestSerializationWorkflowClasses: edge_group = SingleEdgeGroup(source_id="source", target_id="target") # Test model_dump - data = edge_group.model_dump() + data = edge_group.model_dump(by_alias=True) assert "id" in data assert data["id"].startswith("SingleEdgeGroup/") @@ -434,7 +434,7 @@ class TestSerializationWorkflowClasses: ) # Test serialization of the nested structure - data = outer_workflow.model_dump() + data = outer_workflow.model_dump(by_alias=True) # Verify outer structure assert data["start_executor_id"] == "outer-exec" @@ -473,7 +473,7 @@ class TestSerializationWorkflowClasses: assert "inner-exec" in innermost_workflow_data["executors"] # Test JSON serialization preserves the complete nested structure - json_str = outer_workflow.model_dump_json() + json_str = outer_workflow.model_dump_json(by_alias=True) parsed = json.loads(json_str) # Verify the complete structure is preserved in JSON @@ -499,7 +499,7 @@ class TestSerializationWorkflowClasses: assert "inner-exec" in innermost_workflow_json["executors"] # Test that WorkflowExecutor also serializes correctly when accessed directly - direct_middle_data = middle_workflow_executor.model_dump() + direct_middle_data = middle_workflow_executor.model_dump(by_alias=True) assert "workflow" in direct_middle_data assert direct_middle_data["type"] == "WorkflowExecutor" assert "executors" in direct_middle_data["workflow"] diff --git a/python/packages/workflow/tests/test_tracing.py b/python/packages/workflow/tests/test_tracing.py index 5dda235393..c196180ae2 100644 --- a/python/packages/workflow/tests/test_tracing.py +++ b/python/packages/workflow/tests/test_tracing.py @@ -329,7 +329,7 @@ async def test_end_to_end_workflow_tracing(tracing_enabled: Any, span_exporter: assert build_span.attributes.get("workflow.id") == workflow.id assert build_span.attributes.get("workflow.definition") is not None definition = build_span.attributes.get("workflow.definition") - assert definition == workflow.model_dump_json() + assert definition == workflow.model_dump_json(by_alias=True) # Check build events assert build_span.events is not None diff --git a/python/packages/workflow/tests/test_workflow_agent.py b/python/packages/workflow/tests/test_workflow_agent.py new file mode 100644 index 0000000000..35bf5a041c --- /dev/null +++ b/python/packages/workflow/tests/test_workflow_agent.py @@ -0,0 +1,421 @@ +# Copyright (c) Microsoft. All rights reserved. + +import uuid +from typing import Any + +import pytest +from agent_framework import ( + AgentRunResponse, + AgentRunResponseUpdate, + ChatMessage, + ChatRole, + FunctionResultContent, + TextContent, + UsageContent, + UsageDetails, +) +from agent_framework.workflow import ( + AgentRunUpdateEvent, + Executor, + RequestInfoExecutor, + RequestInfoMessage, + WorkflowAgent, + WorkflowBuilder, + WorkflowContext, + handler, +) + + +class SimpleExecutor(Executor): + """Simple executor that emits AgentRunEvent or AgentRunStreamingEvent.""" + + response_text: str + emit_streaming: bool = False + + def __init__(self, id: str, response_text: str, emit_streaming: bool = False): + super().__init__(id=id, response_text=response_text, emit_streaming=emit_streaming) + + @handler + async def handle_message(self, message: list[ChatMessage], ctx: WorkflowContext[list[ChatMessage]]) -> None: + input_text = ( + message[0].contents[0].text if message and isinstance(message[0].contents[0], TextContent) else "no input" + ) + response_text = f"{self.response_text}: {input_text}" + + # Create response message for both streaming and non-streaming cases + response_message = ChatMessage(role=ChatRole.ASSISTANT, contents=[TextContent(text=response_text)]) + + # Emit update event. + streaming_update = AgentRunResponseUpdate( + contents=[TextContent(text=response_text)], role=ChatRole.ASSISTANT, message_id=str(uuid.uuid4()) + ) + await ctx.add_event(AgentRunUpdateEvent(executor_id=self.id, data=streaming_update)) + + # Pass message to next executor if any (for both streaming and non-streaming) + await ctx.send_message([response_message]) + + +class RequestingExecutor(Executor): + """Executor that sends RequestInfoMessage to trigger RequestInfoEvent.""" + + @handler + async def handle_message(self, _: list[ChatMessage], ctx: WorkflowContext[RequestInfoMessage]) -> None: + # Send a RequestInfoMessage to trigger the request info process + await ctx.send_message(RequestInfoMessage()) + + @handler + async def handle_request_response(self, _: Any, ctx: WorkflowContext[ChatMessage]) -> None: + # Handle the response and emit completion response + update = AgentRunResponseUpdate( + contents=[TextContent(text="Request completed successfully")], + role=ChatRole.ASSISTANT, + message_id=str(uuid.uuid4()), + ) + await ctx.add_event(AgentRunUpdateEvent(executor_id=self.id, data=update)) + + +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 + executor1 = SimpleExecutor(id="executor1", response_text="Step1", emit_streaming=False) + executor2 = SimpleExecutor(id="executor2", response_text="Step2", emit_streaming=False) + + workflow = WorkflowBuilder().set_start_executor(executor1).add_edge(executor1, executor2).build() + + agent = WorkflowAgent(workflow=workflow, name="Test Agent") + + # Execute workflow end-to-end + result = await agent.run("Hello World") + + # Verify we got responses from both executors + assert isinstance(result, AgentRunResponse) + assert len(result.messages) >= 2, f"Expected at least 2 messages, got {len(result.messages)}" + + # Find messages from each executor + step1_messages = [] + step2_messages = [] + + for message in result.messages: + first_content = message.contents[0] + if isinstance(first_content, TextContent): + text = first_content.text + if text.startswith("Step1:"): + step1_messages.append(message) + elif text.startswith("Step2:"): + step2_messages.append(message) + + # Verify both executors produced output + assert len(step1_messages) >= 1, "Should have received message from Step1 executor" + assert len(step2_messages) >= 1, "Should have received message from Step2 executor" + + # Verify the processing worked for both + step1_text = step1_messages[0].contents[0].text + step2_text = step2_messages[0].contents[0].text + 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 + executor1 = SimpleExecutor(id="stream1", response_text="Streaming1", emit_streaming=True) + executor2 = SimpleExecutor(id="stream2", response_text="Streaming2", emit_streaming=True) + + # Create workflow with just one executor + workflow = WorkflowBuilder().set_start_executor(executor1).add_edge(executor1, executor2).build() + + agent = WorkflowAgent(workflow=workflow, name="Streaming Test Agent") + + # Execute workflow streaming to capture streaming events + updates = [] + async for update in agent.run_streaming("Test input"): + updates.append(update) + + # Should have received at least one streaming update + assert len(updates) >= 2, f"Expected at least 2 updates, got {len(updates)}" + + # Verify we got a streaming update + assert updates[0].contents is not None + first_content = updates[0].contents[0] + second_content = updates[1].contents[0] + assert isinstance(first_content, TextContent) + assert "Streaming1: Test input" in first_content.text + 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) + requesting_executor = RequestingExecutor(id="requester") + request_info_executor = RequestInfoExecutor() + + workflow = ( + WorkflowBuilder() + .set_start_executor(requesting_executor) + .add_edge(requesting_executor, request_info_executor) + .build() + ) + + agent = WorkflowAgent(workflow=workflow, name="Request Test Agent") + + # Execute workflow streaming to get request info event + updates = [] + async for update in agent.run_streaming("Start request"): + updates.append(update) + # Should have received a function call for the request info + assert len(updates) > 0 + + # Find the function call update (RequestInfoEvent converted to function call) + function_call_update = None + for update in updates: + if update.contents and hasattr(update.contents[0], "name") and update.contents[0].name == "request_info": + function_call_update = update + break + + assert function_call_update is not None, "Should have received a request_info function call" + function_call = function_call_update.contents[0] + + # Verify the function call has expected structure + assert function_call.call_id is not None + assert function_call.name == "request_info" + assert isinstance(function_call.arguments, dict) + assert "request_id" in function_call.arguments + + # Verify the request is tracked in pending_requests + assert len(agent.pending_requests) == 1 + assert function_call.call_id in agent.pending_requests + + # Now provide a function result response to test continuation + response_message = ChatMessage( + role=ChatRole.USER, + contents=[FunctionResultContent(call_id=function_call.call_id, result="User provided answer")], + ) + + # Continue the workflow with the response + continuation_result = await agent.run(response_message) + + # Should complete successfully + assert isinstance(continuation_result, AgentRunResponse) + + # Verify cleanup - pending requests should be cleared after function response handling + assert len(agent.pending_requests) == 0 + + def test_workflow_as_agent_method(self) -> None: + """Test that Workflow.as_agent() creates a properly configured WorkflowAgent.""" + # Create a simple workflow + executor = SimpleExecutor(id="executor1", response_text="Response", emit_streaming=False) + workflow = WorkflowBuilder().set_start_executor(executor).build() + + # Test as_agent with a name + agent = workflow.as_agent(name="TestAgent") + + # Verify the agent is properly configured + assert isinstance(agent, WorkflowAgent) + assert agent.name == "TestAgent" + assert agent.workflow is workflow + assert agent.workflow.id == workflow.id + + # Test as_agent without a name (should use default) + agent_no_name = workflow.as_agent() + assert isinstance(agent_no_name, WorkflowAgent) + assert agent_no_name.workflow is workflow + + def test_workflow_as_agent_cannot_handle_agent_inputs(self) -> None: + """Test that Workflow.as_agent() raises an error if the start executor cannot handle agent inputs.""" + + class _Executor(Executor): + @handler + async def handle_bool(self, message: bool, context: WorkflowContext[Any]) -> None: + raise ValueError("Unsupported message type") + + # Create a simple workflow + executor = _Executor() + workflow = WorkflowBuilder().set_start_executor(executor).build() + + # Try to create an agent with unsupported input types + with pytest.raises(ValueError, match="Workflow's start executor cannot handle list\\[ChatMessage\\]"): + workflow.as_agent() + + +class TestWorkflowAgentMergeUpdates: + """Test cases specifically for the WorkflowAgent.merge_updates static method.""" + + def test_merge_updates_ordering_by_response_and_message_id(self): + """Test that merge_updates correctly orders messages by response_id groups and message_id chronologically.""" + # Create updates with different response_ids and message_ids in non-chronological order + updates = [ + # Response B, Message 2 (latest in resp B) + AgentRunResponseUpdate( + contents=[TextContent(text="RespB-Msg2")], + role=ChatRole.ASSISTANT, + response_id="resp-b", + message_id="msg-2", + created_at="2024-01-01T12:02:00Z", + ), + # Response A, Message 1 (earliest overall) + AgentRunResponseUpdate( + contents=[TextContent(text="RespA-Msg1")], + role=ChatRole.ASSISTANT, + response_id="resp-a", + message_id="msg-1", + created_at="2024-01-01T12:00:00Z", + ), + # Response B, Message 1 (earlier in resp B) + AgentRunResponseUpdate( + contents=[TextContent(text="RespB-Msg1")], + role=ChatRole.ASSISTANT, + response_id="resp-b", + message_id="msg-1", + created_at="2024-01-01T12:01:00Z", + ), + # Response A, Message 2 (later in resp A) + AgentRunResponseUpdate( + contents=[TextContent(text="RespA-Msg2")], + role=ChatRole.ASSISTANT, + response_id="resp-a", + message_id="msg-2", + created_at="2024-01-01T12:00:30Z", + ), + # Global dangling update (no response_id) - should go at end + AgentRunResponseUpdate( + contents=[TextContent(text="Global-Dangling")], + role=ChatRole.ASSISTANT, + response_id=None, + message_id="msg-global", + created_at="2024-01-01T11:59:00Z", # Earliest timestamp but should be last + ), + ] + + result = WorkflowAgent.merge_updates(updates, "final-response-id") + + # Verify correct response_id is set + assert result.response_id == "final-response-id" + + # Should have 5 messages total + assert len(result.messages) == 5 + + # Verify ordering: responses are processed by response_id groups, + # within each group messages are chronologically ordered, + # global dangling goes at the end + message_texts = [ + msg.contents[0].text if isinstance(msg.contents[0], TextContent) else "" for msg in result.messages + ] + + # The exact order depends on dict iteration order for response_ids, + # but within each response group, chronological order should be maintained + # and global dangling should be last + assert "Global-Dangling" in message_texts[-1] # Global dangling at end + + # Find positions of resp-a and resp-b messages + resp_a_positions = [i for i, text in enumerate(message_texts) if "RespA" in text] + resp_b_positions = [i for i, text in enumerate(message_texts) if "RespB" in text] + + # Within resp-a group: Msg1 (earlier) should come before Msg2 (later) + resp_a_texts = [message_texts[i] for i in resp_a_positions] + assert resp_a_texts.index("RespA-Msg1") < resp_a_texts.index("RespA-Msg2") + + # Within resp-b group: Msg1 (earlier) should come before Msg2 (later) + resp_b_texts = [message_texts[i] for i in resp_b_positions] + assert resp_b_texts.index("RespB-Msg1") < resp_b_texts.index("RespB-Msg2") + + # ENHANCED: Verify response group separation and ordering + # Messages from the same response_id should be grouped together (not interleaved) + + # Check resp-a group is contiguous (all positions are consecutive) + if len(resp_a_positions) > 1: + for i in range(1, len(resp_a_positions)): + assert resp_a_positions[i] == resp_a_positions[i - 1] + 1, ( + f"RespA messages are not contiguous: positions {resp_a_positions}" + ) + + # Check resp-b group is contiguous (all positions are consecutive) + if len(resp_b_positions) > 1: + for i in range(1, len(resp_b_positions)): + assert resp_b_positions[i] == resp_b_positions[i - 1] + 1, ( + f"RespB messages are not contiguous: positions {resp_b_positions}" + ) + + # Response groups are no longer required to be ordered by latest timestamp + # We only ensure messages within each group are chronologically ordered + # Verify global dangling message position (should be last, after all response groups) + global_dangling_pos = message_texts.index("Global-Dangling") + if resp_a_positions: + assert global_dangling_pos > max(resp_a_positions), "Global dangling should come after resp-a group" + if resp_b_positions: + assert global_dangling_pos > max(resp_b_positions), "Global dangling should come after resp-b group" + + def test_merge_updates_metadata_aggregation(self): + """Test that merge_updates correctly aggregates usage details, timestamps, and additional properties.""" + # Create updates with various metadata including usage details + updates = [ + AgentRunResponseUpdate( + contents=[ + TextContent(text="First"), + UsageContent( + details=UsageDetails(input_token_count=10, output_token_count=5, total_token_count=15) + ), + ], + role=ChatRole.ASSISTANT, + response_id="resp-1", + message_id="msg-1", + created_at="2024-01-01T12:00:00Z", + additional_properties={"source": "executor1", "priority": "high"}, + ), + AgentRunResponseUpdate( + contents=[ + TextContent(text="Second"), + UsageContent( + details=UsageDetails(input_token_count=20, output_token_count=8, total_token_count=28) + ), + ], + role=ChatRole.ASSISTANT, + response_id="resp-2", + message_id="msg-2", + created_at="2024-01-01T12:01:00Z", # Later timestamp + additional_properties={"source": "executor2", "category": "analysis"}, + ), + AgentRunResponseUpdate( + contents=[ + TextContent(text="Third"), + UsageContent(details=UsageDetails(input_token_count=5, output_token_count=3, total_token_count=8)), + ], + role=ChatRole.ASSISTANT, + response_id="resp-1", # Same response_id as first + message_id="msg-3", + created_at="2024-01-01T11:59:00Z", # Earlier timestamp + additional_properties={"details": "merged", "priority": "low"}, # Different priority value + ), + ] + + result = WorkflowAgent.merge_updates(updates, "aggregated-response") + + # Verify response_id is set correctly + assert result.response_id == "aggregated-response" + + # Verify latest timestamp is used (should be 12:01:00Z from second update) + assert result.created_at == "2024-01-01T12:01:00Z" + + # Verify messages are present + assert len(result.messages) == 3 + + # Verify usage details are aggregated correctly + # Should sum all usage details: (10+20+5) + (5+8+3) + (15+28+8) = 35+16+51 = 51 total tokens + expected_usage = UsageDetails(input_token_count=35, output_token_count=16, total_token_count=51) + assert result.usage_details == expected_usage + + # Verify additional properties are merged correctly + # Note: Within response groups, later updates' properties win conflicts, + # but across response groups, the dict.update() order determines which wins + expected_properties = { + "source": "executor2", # From resp-2 (latest source value) + "priority": "high", # From resp-1 first update (resp-1 processed before resp-2) + "category": "analysis", # From resp-2 (only place this appears) + # "details": "merged" is NOT in final result because resp-1's aggregated + # properties only include final merged result from its own updates + } + assert result.additional_properties == expected_properties diff --git a/python/samples/getting_started/workflow/step_10a_workflow_agent_reflection_pattern.py b/python/samples/getting_started/workflow/step_10a_workflow_agent_reflection_pattern.py new file mode 100644 index 0000000000..9b86ebc103 --- /dev/null +++ b/python/samples/getting_started/workflow/step_10a_workflow_agent_reflection_pattern.py @@ -0,0 +1,248 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +from dataclasses import dataclass +from uuid import uuid4 + +from agent_framework import AgentRunResponseUpdate, AIContents, ChatClient, ChatMessage, ChatRole +from agent_framework.openai import OpenAIChatClient +from agent_framework.workflow import AgentRunUpdateEvent, Executor, WorkflowBuilder, WorkflowContext, handler +from pydantic import BaseModel + +""" +The following sample demonstrates how to wrap a workflow as an agent using WorkflowAgent. + +This sample shows how to: +1. Create a workflow with a reflection pattern (Worker + Reviewer executors) +2. Wrap the workflow as an agent using the .as_agent() method +3. Stream responses from the workflow agent like a regular agent +4. Implement a review-retry mechanism where responses are iteratively improved + +The example implements a quality-controlled AI assistant where: +- Worker executor generates responses to user queries +- Reviewer executor evaluates the responses and provides feedback +- If not approved, the Worker incorporates feedback and regenerates the response +- The cycle continues until the response is approved +- Only approved responses are emitted to the external consumer + +Key concepts demonstrated: +- WorkflowAgent: Wraps a workflow to make it behave as an agent +- Bidirectional workflow with cycles (Worker ↔ Reviewer) +- AgentRunUpdateEvent: How workflows communicate with external consumers +- Structured output parsing for review feedback +- State management with pending requests tracking +""" + + +@dataclass +class ReviewRequest: + request_id: str + user_messages: list[ChatMessage] + agent_messages: list[ChatMessage] + + +@dataclass +class ReviewResponse: + request_id: str + feedback: str + approved: bool + + +class Reviewer(Executor): + """An executor that reviews messages and provides feedback.""" + + def __init__(self, chat_client: ChatClient) -> None: + super().__init__() + self._chat_client = chat_client + + @handler + async def review(self, request: ReviewRequest, ctx: WorkflowContext[ReviewResponse]) -> None: + print(f"🔍 Reviewer: Evaluating response for request {request.request_id[:8]}...") + + # Use the chat client to review the message and use structured output. + # NOTE: this can be modified to use an evaluation framework. + + class _Response(BaseModel): + feedback: str + approved: bool + + # Define the system prompt. + messages = [ + ChatMessage( + role=ChatRole.SYSTEM, + text="You are a reviewer for an AI agent, please provide feedback on the " + "following exchange between a user and the AI agent, " + "and indicate if the agent's responses are approved or not.\n" + "Use the following criteria for your evaluation:\n" + "- Relevance: Does the response address the user's query?\n" + "- Accuracy: Is the information provided correct?\n" + "- Clarity: Is the response easy to understand?\n" + "- Completeness: Does the response cover all aspects of the query?\n" + "Be critical in your evaluation and provide constructive feedback.\n" + "Do not approve until all criteria are met.", + ) + ] + + # Add user and agent messages to the chat history. + messages.extend(request.user_messages) + + # Add agent messages to the chat history. + messages.extend(request.agent_messages) + + # Add add one more instruction for the assistant to follow. + messages.append( + ChatMessage(role=ChatRole.USER, text="Please provide a review of the agent's responses to the user.") + ) + + print("🔍 Reviewer: Sending review request to LLM...") + # Get the response from the chat client. + response = await self._chat_client.get_response(messages=messages, response_format=_Response) + + # Parse the response. + parsed = _Response.model_validate_json(response.messages[-1].text) + + print(f"🔍 Reviewer: Review complete - Approved: {parsed.approved}") + print(f"🔍 Reviewer: Feedback: {parsed.feedback}") + + # Send the review response. + await ctx.send_message( + ReviewResponse(request_id=request.request_id, feedback=parsed.feedback, approved=parsed.approved) + ) + + +class Worker(Executor): + """An executor that performs tasks for the user.""" + + def __init__(self, chat_client: ChatClient) -> None: + super().__init__() + self._chat_client = chat_client + self._pending_requests: dict[str, tuple[ReviewRequest, list[ChatMessage]]] = {} + + @handler + async def handle_user_messages(self, user_messages: list[ChatMessage], ctx: WorkflowContext[ReviewRequest]) -> None: + print("🔧 Worker: Received user messages, generating response...") + + # Handle user messages and prepare a review request for the reviewer. + # Define the system prompt. + messages = [ChatMessage(role=ChatRole.SYSTEM, text="You are a helpful assistant.")] + + # Add user messages. + messages.extend(user_messages) + + print("🔧 Worker: Calling LLM to generate response...") + # Get the response from the chat client. + response = await self._chat_client.get_response(messages=messages) + print(f"🔧 Worker: Response generated: {response.messages[-1].text}") + + # Add agent messages. + messages.extend(response.messages) + + # Create the review request. + request = ReviewRequest(request_id=str(uuid4()), user_messages=user_messages, agent_messages=response.messages) + + print(f"🔧 Worker: Generated response, sending to reviewer (ID: {request.request_id[:8]})") + # Send the review request. + await ctx.send_message(request) + + # Add to pending requests. + self._pending_requests[request.request_id] = (request, messages) + + @handler + async def handle_review_response(self, review: ReviewResponse, ctx: WorkflowContext[ReviewRequest]) -> None: + print(f"🔧 Worker: Received review for request {review.request_id[:8]} - Approved: {review.approved}") + + # Handle the review response. Depending on the approval status, + # either emit the approved response as AgentRunUpdateEvent, or + # retry given the feedback. + if review.request_id not in self._pending_requests: + raise ValueError(f"Received review response for unknown request ID: {review.request_id}") + # Remove the request from pending requests. + request, messages = self._pending_requests.pop(review.request_id) + + if review.approved: + print("✅ Worker: Response approved! Emitting to external consumer...") + # If approved, emit the agent run response update to the workflow's + # external consumer. + contents: list[AIContents] = [] + for message in request.agent_messages: + contents.extend(message.contents) + # Emitting an AgentRunUpdateEvent in a workflow wrapped by a WorkflowAgent + # will send the AgentRunResponseUpdate to the WorkflowAgent's + # event stream. + await ctx.add_event( + AgentRunUpdateEvent(self.id, data=AgentRunResponseUpdate(contents=contents, role=ChatRole.ASSISTANT)) + ) + return + + print(f"❌ Worker: Response not approved. Feedback: {review.feedback}") + print("🔧 Worker: Incorporating feedback and regenerating response...") + + # Construct new messages with feedback. + messages.append(ChatMessage(role=ChatRole.SYSTEM, text=review.feedback)) + + # Add additional instruction to address the feedback. + messages.append( + ChatMessage( + role=ChatRole.SYSTEM, + text="Please incorporate the feedback above, and provide a response to user's next message.", + ) + ) + messages.extend(request.user_messages) + + # Get the new response from the chat client. + response = await self._chat_client.get_response(messages=messages) + print(f"🔧 Worker: New response generated after feedback: {response.messages[-1].text}") + + # Process the response. + messages.extend(response.messages) + + print(f"🔧 Worker: Generated improved response, sending for re-review (ID: {review.request_id[:8]})") + # Send an updated review request. + new_request = ReviewRequest( + request_id=review.request_id, user_messages=request.user_messages, agent_messages=response.messages + ) + await ctx.send_message(new_request) + + # Add to pending requests. + self._pending_requests[new_request.request_id] = (new_request, messages) + + +async def main() -> None: + print("🚀 Starting Workflow Agent Demo") + print("=" * 50) + + # Create executors. + print("📝 Creating chat client and executors...") + mini_chat_client = OpenAIChatClient(ai_model_id="gpt-4.1-nano") + chat_client = OpenAIChatClient(ai_model_id="gpt-4.1") + reviewer = Reviewer(chat_client=chat_client) + worker = Worker(chat_client=mini_chat_client) + + print("🏗️ Building workflow with Worker ↔ Reviewer cycle...") + # Create the workflow agent with an underlying reflection workflow. + agent = ( + WorkflowBuilder() + .add_edge(worker, reviewer) # <--- This edge allows the worker to send requests to the reviewer + .add_edge(reviewer, worker) # <--- This edge allows the reviewer to send feedback back to the worker + .set_start_executor(worker) + .build() + .as_agent() # Convert the workflow to an agent. + ) + + print("🎯 Running workflow agent with user query...") + print("Query: 'Write code for parallel reading 1 million files on disk and write to a sorted output file.'") + print("-" * 50) + + # Run the agent and stream events. + async for event in agent.run_streaming( + "Write code for parallel reading 1 million files on disk and write to a sorted output file." + ): + print(f"📤 Agent Response: {event}") + + print("=" * 50) + print("✅ Workflow completed!") + + +if __name__ == "__main__": + print("🎬 Initializing Workflow as Agent Sample...") + asyncio.run(main()) diff --git a/python/samples/getting_started/workflow/step_10b_workflow_agent_human_in_the_loop.py b/python/samples/getting_started/workflow/step_10b_workflow_agent_human_in_the_loop.py new file mode 100644 index 0000000000..73372ea0bf --- /dev/null +++ b/python/samples/getting_started/workflow/step_10b_workflow_agent_human_in_the_loop.py @@ -0,0 +1,145 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +from dataclasses import dataclass + +from agent_framework import ( + ChatMessage, + ChatRole, + FunctionCallContent, + FunctionResultContent, +) +from agent_framework.openai import OpenAIChatClient +from agent_framework.workflow import ( + Executor, + RequestInfoExecutor, + RequestInfoMessage, + RequestResponse, + WorkflowAgent, + WorkflowBuilder, + WorkflowContext, + handler, +) +from step_10a_workflow_agent_reflection_pattern import ReviewRequest, ReviewResponse, Worker + + +@dataclass +class HumanReviewRequest(RequestInfoMessage): + agent_request: ReviewRequest | None = None + + +class ReviewerWithHumanInTheLoop(Executor): + """An executor that raises to human manager for review when not confident.""" + + def __init__(self, worker_id: str, request_info_id: str) -> None: + super().__init__() + self._worker_id = worker_id + self._request_info_id = request_info_id + + @handler + async def review(self, request: ReviewRequest, ctx: WorkflowContext[ReviewResponse | HumanReviewRequest]) -> None: + print(f"🔍 Reviewer: Evaluating response for request {request.request_id[:8]}...") + + # NOTE: for simplicity, we always escalate to human manager. + # See step_10a_workflow_agent_reflection_pattern.py for implementation + # using an chat client. + + print("🔍 Reviewer: Escalate to human manager") + # Send to human manager + await ctx.send_message( + HumanReviewRequest(agent_request=request), + target_id=self._request_info_id, + ) + + @handler + async def accept_human_review( + self, response: RequestResponse[HumanReviewRequest, ReviewResponse], ctx: WorkflowContext[ReviewResponse] + ) -> None: + human_response = response.data + assert isinstance(human_response, ReviewResponse) + print(f"🔍 Reviewer: Accepting human review for request {human_response.request_id[:8]}...") + print(f"🔍 Reviewer: Human feedback: {human_response.feedback}") + print(f"🔍 Reviewer: Human approved: {human_response.approved}") + print("🔍 Reviewer: Forwarding human review back to worker...") + await ctx.send_message(human_response, target_id=self._worker_id) + + +async def main() -> None: + print("🚀 Starting Workflow Agent with Human-in-the-Loop Demo") + print("=" * 50) + + # Create executors. + print("📝 Creating chat client and executors...") + mini_chat_client = OpenAIChatClient(ai_model_id="gpt-4.1-nano") + worker = Worker(chat_client=mini_chat_client) + request_info_executor = RequestInfoExecutor() + reviewer = ReviewerWithHumanInTheLoop(worker_id=worker.id, request_info_id=request_info_executor.id) + + print("🏗️ Building workflow with Worker ↔ Reviewer cycle...") + # Create the workflow agent with an underlying reflection workflow. + agent = ( + WorkflowBuilder() + .add_edge(worker, reviewer) # <--- This edge allows the worker to send requests to the reviewer + .add_edge(reviewer, worker) # <--- This edge allows the reviewer to send feedback back to the worker + .add_edge( + reviewer, request_info_executor + ) # <--- This edge allows the reviewer to send human input requests through the request info executor + .add_edge( + request_info_executor, reviewer + ) # <--- This edge allows the human input to be forwarded back to the reviewer + .set_start_executor(worker) + .build() + .as_agent() # Convert the workflow to an agent. + ) + + print("🎯 Running workflow agent with user query...") + print("Query: 'Write code for parallel reading 1 million files on disk and write to a sorted output file.'") + print("-" * 50) + + # NOTE: you can also run the workflow directly, i.e., without the as_agent(). + # Then, you will need to handle RequestInfoEvent and send response to the workflow + # using send_response(). + + # Run the agent. + response = await agent.run( + "Write code for parallel reading 1 million Files on disk and write to a sorted output file." + ) + # + # Find human review function call. + # TODO(ekzhu): update this to FunctionApprovalRequestContent + # monitor: https://github.com/microsoft/agent-framework/issues/285 + human_review_function_call: FunctionCallContent | None = None + for message in response.messages: + for content in message.contents: + if isinstance(content, FunctionCallContent) and content.name == WorkflowAgent.REQUEST_INFO_FUNCTION_NAME: + human_review_function_call = content + + # Handle human review if needed. + if human_review_function_call: + # Use WorkflowAgent.RequestInfoFunctionArgs to parse the request. + if isinstance(human_review_function_call.arguments, str): + request = WorkflowAgent.RequestInfoFunctionArgs.model_validate_json(human_review_function_call.arguments) + else: + request = WorkflowAgent.RequestInfoFunctionArgs.model_validate(human_review_function_call.arguments) + # Mock a human approval. + human_response = ReviewResponse( + request_id=request.data["agent_request"]["request_id"], feedback="Approved", approved=True + ) + # Create the function call result to be sent back. + # TODO(ekzhu): update this to FunctionApprovalResponseContent + # monitor: https://github.com/microsoft/agent-framework/issues/285 + human_review_function_result = FunctionResultContent( + call_id=human_review_function_call.call_id, + result=human_response, + ) + # Send the human review result back to the agent. + response = await agent.run(ChatMessage(role=ChatRole.TOOL, contents=[human_review_function_result])) + print(f"📤 Agent Response: {response.messages[-1].text}") + + print("=" * 50) + print("✅ Workflow completed!") + + +if __name__ == "__main__": + print("🎬 Initializing Workflow as Agent Sample...") + asyncio.run(main())