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

[BREAKING] Python: Merge send_responses into run method (#3720)

Browse Source 
* Streamline workflow run api with send responses in one method

* Fixes

* Address copilot feedback
This commit is contained in:
Evan Mattson
2026-02-07 07:32:38 +09:00
committed by GitHub
Unverified
parent 15256bb616
commit a17f13598b
39 changed files with 561 additions and 335 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/core/agent_framework/_workflows/_agent.py
+10 -18
View File
@@ -340,25 +340,22 @@ class WorkflowAgent(BaseAgent):
Yields:
WorkflowEvent objects from the workflow execution.
"""
# Determine the execution mode based on state
# Determine the execution mode based on state.
# The streaming flag controls the workflow's internal streaming mode,
# which affects executor behavior (e.g. AgentExecutor emits different event
# types in streaming vs non-streaming mode).
if bool(self.pending_requests):
# This is a continuation - send function responses back
function_responses = self._process_pending_requests(input_messages)
if streaming:
async for event in self.workflow.send_responses_streaming(function_responses):
async for event in self.workflow.run(responses=function_responses, stream=True, **kwargs):
yield event
else:
workflow_result = await self.workflow.send_responses(function_responses)
for event in workflow_result:
for event in await self.workflow.run(responses=function_responses, **kwargs):
yield event
elif checkpoint_id is not None:
# Resume from checkpoint - don't prepend thread history since workflow state
# is being restored from the checkpoint
if streaming:
async for event in self.workflow.run(
message=None,
stream=True,
checkpoint_id=checkpoint_id,
checkpoint_storage=checkpoint_storage,
@@ -366,19 +363,15 @@ class WorkflowAgent(BaseAgent):
):
yield event
else:
workflow_result = await self.workflow.run(
message=None,
for event in await self.workflow.run(
checkpoint_id=checkpoint_id,
checkpoint_storage=checkpoint_storage,
**kwargs,
)
for event in workflow_result:
):
yield event
else:
# Initial run - build conversation from thread history
conversation_messages = await self._build_conversation_messages(thread, input_messages)
if streaming:
async for event in self.workflow.run(
message=conversation_messages,
@@ -388,12 +381,11 @@ class WorkflowAgent(BaseAgent):
):
yield event
else:
workflow_result = await self.workflow.run(
for event in await self.workflow.run(
message=conversation_messages,
checkpoint_storage=checkpoint_storage,
**kwargs,
)
for event in workflow_result:
):
yield event
# endregion Run Methods
python/packages/core/agent_framework/_workflows/_typing_utils.py
+48
View File
@@ -177,6 +177,54 @@ def is_instance_of(data: Any, target_type: type | UnionType | Any) -> bool:
return isinstance(data, target_type)
def try_coerce_to_type(data: Any, target_type: type | UnionType | Any) -> Any:
"""Try to coerce data to the target type.
Attempts lightweight type coercion for common cases where raw data
(e.g., from JSON deserialization) needs to be converted to the expected type.
Returns the coerced value if successful, or the original value if coercion
is not needed or not possible.
Args:
data: The data to coerce.
target_type: The type to coerce to.
Returns:
The coerced value, or the original value if coercion fails.
"""
# If already the right type, return as-is
if is_instance_of(data, target_type):
return data
# Can't coerce to non-concrete targets (Union, generic, etc.)
if not isinstance(target_type, type):
return data
# int -> float (JSON integers for float fields)
if isinstance(data, int) and target_type is float:
return float(data)
# dict -> dataclass
if isinstance(data, dict):
from dataclasses import is_dataclass
if is_dataclass(target_type):
try:
return target_type(**data)
except (TypeError, ValueError):
return data
# dict -> Pydantic model
if hasattr(target_type, "model_validate"):
try:
return target_type.model_validate(data)
except Exception:
return data
return data
def serialize_type(t: type) -> str:
"""Serialize a type to a string.
python/packages/core/agent_framework/_workflows/_workflow.py
+162 -153
View File
@@ -9,9 +9,10 @@ import json
import logging
import types
import uuid
from collections.abc import AsyncIterable, Awaitable, Callable
from collections.abc import AsyncIterable, Awaitable, Callable, Sequence
from typing import Any, Literal, overload
from .._types import ResponseStream
from ..observability import OtelAttr, capture_exception, create_workflow_span
from ._agent import WorkflowAgent
from ._checkpoint import CheckpointStorage
@@ -31,7 +32,7 @@ from ._model_utils import DictConvertible
from ._runner import Runner
from ._runner_context import RunnerContext
from ._state import State
from ._typing_utils import is_instance_of
from ._typing_utils import is_instance_of, try_coerce_to_type
logger = logging.getLogger(__name__)
@@ -144,7 +145,7 @@ class Workflow(DictConvertible):
2. Executor implements `response_handler()` to process the response
3. Requests are emitted as request_info events (WorkflowEvent with type='request_info') in the event stream
4. Workflow enters IDLE_WITH_PENDING_REQUESTS state
5. Caller handles requests and provides responses via the `send_responses` or `send_responses_streaming` methods
5. Caller handles requests and provides responses via `run(responses=...)` or `run(responses=..., stream=True)`
6. Responses are routed to the requesting executors and response handlers are invoked
## Checkpointing
@@ -450,186 +451,143 @@ class Workflow(DictConvertible):
message: Any | None = None,
*,
stream: Literal[True],
responses: dict[str, Any] | None = None,
checkpoint_id: str | None = None,
checkpoint_storage: CheckpointStorage | None = None,
**kwargs: Any,
) -> AsyncIterable[WorkflowEvent]: ...
) -> ResponseStream[WorkflowEvent, WorkflowRunResult]: ...
@overload
async def run(
def run(
self,
message: Any | None = None,
*,
stream: Literal[False] = ...,
responses: dict[str, Any] | None = None,
checkpoint_id: str | None = None,
checkpoint_storage: CheckpointStorage | None = None,
include_status_events: bool = False,
**kwargs: Any,
) -> WorkflowRunResult: ...
) -> Awaitable[WorkflowRunResult]: ...
def run(
self,
message: Any | None = None,
*,
stream: bool = False,
responses: dict[str, Any] | None = None,
checkpoint_id: str | None = None,
checkpoint_storage: CheckpointStorage | None = None,
include_status_events: bool = False,
**kwargs: Any,
) -> AsyncIterable[WorkflowEvent] | Awaitable[WorkflowRunResult]:
) -> ResponseStream[WorkflowEvent, WorkflowRunResult] | Awaitable[WorkflowRunResult]:
"""Run the workflow, optionally streaming events.
Unified interface supporting initial runs and checkpoint restoration.
Unified interface supporting initial runs, checkpoint restoration, and
sending responses to pending requests.
Args:
message: Initial message for the start executor. Required for new workflow runs,
should be None when resuming from checkpoint.
stream: If True, returns an async iterable of events. If False (default),
returns an awaitable WorkflowRunResult.
checkpoint_id: ID of checkpoint to restore from. If provided, the workflow resumes
from this checkpoint instead of starting fresh.
message: Initial message for the start executor. Required for new workflow runs.
Mutually exclusive with responses.
stream: If True, returns a ResponseStream of events with
``get_final_response()`` for the final WorkflowRunResult. If False
(default), returns an awaitable WorkflowRunResult.
responses: Responses to send for pending request info events, where keys are
request IDs and values are the corresponding response data. Mutually
exclusive with message. Can be combined with checkpoint_id to restore
a checkpoint and send responses in a single call.
checkpoint_id: ID of checkpoint to restore from. Can be used alone (resume
from checkpoint), with message (not allowed), or with responses
(restore then send responses).
checkpoint_storage: Runtime checkpoint storage.
include_status_events: Whether to include WorkflowStatusEvent instances (non-streaming only).
include_status_events: Whether to include status events (non-streaming only).
**kwargs: Additional keyword arguments to pass through to agent invocations.
Returns:
When stream=True: An AsyncIterable[WorkflowEvent] for streaming events.
When stream=True: A ResponseStream[WorkflowEvent, WorkflowRunResult] for
streaming events. Iterate for events, call get_final_response() for result.
When stream=False: An Awaitable[WorkflowRunResult] with all events.
Raises:
ValueError: If both message and checkpoint_id are provided, or if neither is provided.
ValueError: If parameter combination is invalid.
"""
if stream:
return self._run_streaming(
# Validate parameters and set running flag eagerly (before any async work)
self._validate_run_params(message, responses, checkpoint_id)
self._ensure_not_running()
response_stream = ResponseStream[WorkflowEvent, WorkflowRunResult](
self._run_core(
message=message,
responses=responses,
checkpoint_id=checkpoint_id,
checkpoint_storage=checkpoint_storage,
streaming=stream,
**kwargs,
)
return self._run_non_streaming(
message=message,
checkpoint_id=checkpoint_id,
checkpoint_storage=checkpoint_storage,
include_status_events=include_status_events,
**kwargs,
),
finalizer=functools.partial(self._finalize_events, include_status_events=include_status_events),
cleanup_hooks=[
functools.partial(self._run_cleanup, checkpoint_storage),
],
)
async def _run_streaming(
if stream:
return response_stream
return response_stream.get_final_response()
async def _run_core(
self,
message: Any | None = None,
*,
responses: dict[str, Any] | None = None,
checkpoint_id: str | None = None,
checkpoint_storage: CheckpointStorage | None = None,
streaming: bool = False,
**kwargs: Any,
) -> AsyncIterable[WorkflowEvent]:
"""Internal streaming implementation."""
# Validate mutually exclusive parameters BEFORE setting running flag
if message is not None and checkpoint_id is not None:
raise ValueError("Cannot provide both 'message' and 'checkpoint_id'. Use one or the other.")
if message is None and checkpoint_id is None:
raise ValueError("Must provide either 'message' (new run) or 'checkpoint_id' (resume).")
self._ensure_not_running()
# Enable runtime checkpointing if storage provided
# Two cases:
# 1. checkpoint_storage + checkpoint_id: Load checkpoint from this storage and resume
# 2. checkpoint_storage without checkpoint_id: Enable checkpointing for this run
if checkpoint_storage is not None:
self._runner.context.set_runtime_checkpoint_storage(checkpoint_storage)
try:
# Reset context only for new runs (not checkpoint restoration)
reset_context = message is not None and checkpoint_id is None
async for event in self._run_workflow_with_tracing(
initial_executor_fn=functools.partial(
self._execute_with_message_or_checkpoint, message, checkpoint_id, checkpoint_storage
),
reset_context=reset_context,
streaming=True,
run_kwargs=kwargs if kwargs else None,
):
if event.type == "output" and not self._should_yield_output_event(event):
continue
yield event
finally:
if checkpoint_storage is not None:
self._runner.context.clear_runtime_checkpoint_storage()
self._reset_running_flag()
async def send_responses_streaming(self, responses: dict[str, Any]) -> AsyncIterable[WorkflowEvent]:
"""Send responses back to the workflow and stream the events generated by the workflow.
Args:
responses: The responses to be sent back to the workflow, where keys are request IDs
and values are the corresponding response data.
"""Single core execution path for both streaming and non-streaming modes.
Yields:
WorkflowEvent: The events generated during the workflow execution after sending the responses.
WorkflowEvent: The events generated during the workflow execution.
"""
self._ensure_not_running()
try:
async for event in self._run_workflow_with_tracing(
initial_executor_fn=functools.partial(self._send_responses_internal, responses),
reset_context=False, # Don't reset context when sending responses
streaming=True,
):
if event.type == "output" and not self._should_yield_output_event(event):
continue
yield event
finally:
self._reset_running_flag()
async def _run_non_streaming(
self,
message: Any | None = None,
*,
checkpoint_id: str | None = None,
checkpoint_storage: CheckpointStorage | None = None,
include_status_events: bool = False,
**kwargs: Any,
) -> WorkflowRunResult:
"""Internal non-streaming implementation."""
# Validate mutually exclusive parameters BEFORE setting running flag
if message is not None and checkpoint_id is not None:
raise ValueError("Cannot provide both 'message' and 'checkpoint_id'. Use one or the other.")
if message is None and checkpoint_id is None:
raise ValueError("Must provide either 'message' (new run) or 'checkpoint_id' (resume).")
self._ensure_not_running()
# Enable runtime checkpointing if storage provided
if checkpoint_storage is not None:
self._runner.context.set_runtime_checkpoint_storage(checkpoint_storage)
try:
# Reset context only for new runs (not checkpoint restoration)
reset_context = message is not None and checkpoint_id is None
initial_executor_fn, reset_context = self._resolve_execution_mode(
message, responses, checkpoint_id, checkpoint_storage
)
raw_events = [
event
async for event in self._run_workflow_with_tracing(
initial_executor_fn=functools.partial(
self._execute_with_message_or_checkpoint, message, checkpoint_id, checkpoint_storage
),
reset_context=reset_context,
run_kwargs=kwargs if kwargs else None,
)
]
finally:
if checkpoint_storage is not None:
self._runner.context.clear_runtime_checkpoint_storage()
self._reset_running_flag()
async for event in self._run_workflow_with_tracing(
initial_executor_fn=initial_executor_fn,
reset_context=reset_context,
streaming=streaming,
run_kwargs=kwargs if kwargs else None,
):
if event.type == "output" and not self._should_yield_output_event(event):
continue
yield event
# Filter events for non-streaming mode
filtered: list[WorkflowEvent[Any]] = []
status_events: list[WorkflowEvent[Any]] = []
async def _run_cleanup(self, checkpoint_storage: CheckpointStorage | None) -> None:
"""Cleanup hook called after stream consumption."""
if checkpoint_storage is not None:
self._runner.context.clear_runtime_checkpoint_storage()
self._reset_running_flag()
for ev in raw_events:
# Omit started events from non-streaming (telemetry-only)
@staticmethod
def _finalize_events(
events: Sequence[WorkflowEvent],
*,
include_status_events: bool = False,
) -> WorkflowRunResult:
"""Convert collected workflow events into a WorkflowRunResult.
Filters out internal events for non-streaming callers.
"""
filtered: list[WorkflowEvent] = []
status_events: list[WorkflowEvent] = []
for ev in events:
# Omit started events from result (telemetry-only)
if ev.type == "started":
continue
# Track status; include inline only if explicitly requested
@@ -638,41 +596,88 @@ class Workflow(DictConvertible):
if include_status_events:
filtered.append(ev)
continue
if ev.type == "output" and not self._should_yield_output_event(ev):
continue
filtered.append(ev)
return WorkflowRunResult(filtered, status_events)
async def send_responses(self, responses: dict[str, Any]) -> WorkflowRunResult:
"""Send responses back to the workflow.
@staticmethod
def _validate_run_params(
message: Any | None,
responses: dict[str, Any] | None,
checkpoint_id: str | None,
) -> None:
"""Validate parameter combinations for run().
Args:
responses: A dictionary where keys are request IDs and values are the corresponding response data.
Rules:
- message and responses are mutually exclusive
- message and checkpoint_id are mutually exclusive
- At least one of message, responses, or checkpoint_id must be provided
- responses + checkpoint_id is allowed (restore then send)
"""
if message is not None and responses is not None:
raise ValueError("Cannot provide both 'message' and 'responses'. Use one or the other.")
if message is not None and checkpoint_id is not None:
raise ValueError("Cannot provide both 'message' and 'checkpoint_id'. Use one or the other.")
if message is None and responses is None and checkpoint_id is None:
raise ValueError(
"Must provide at least one of: 'message' (new run), 'responses' (send responses), "
"or 'checkpoint_id' (resume from checkpoint)."
)
def _resolve_execution_mode(
self,
message: Any | None,
responses: dict[str, Any] | None,
checkpoint_id: str | None,
checkpoint_storage: CheckpointStorage | None,
) -> tuple[Callable[[], Awaitable[None]], bool]:
"""Determine the initial executor function and reset_context flag based on parameters.
Returns:
A WorkflowRunResult instance containing a list of events generated during the workflow execution.
A tuple of (initial_executor_fn, reset_context).
"""
self._ensure_not_running()
try:
events = [
event
async for event in self._run_workflow_with_tracing(
initial_executor_fn=functools.partial(self._send_responses_internal, responses),
reset_context=False, # Don't reset context when sending responses
if responses is not None:
if checkpoint_id is not None:
# Combined: restore checkpoint then send responses
initial_executor_fn = functools.partial(
self._restore_and_send_responses, checkpoint_id, checkpoint_storage, responses
)
]
status_events = [e for e in events if e.type == "status"]
filtered_events: list[WorkflowEvent[Any]] = []
for e in events:
if e.type == "output" and not self._should_yield_output_event(e):
continue
if e.type in ("status", "started"):
continue
filtered_events.append(e)
return WorkflowRunResult(filtered_events, status_events)
finally:
self._reset_running_flag()
else:
# Send responses only (requires pending requests in workflow state)
initial_executor_fn = functools.partial(self._send_responses_internal, responses)
return initial_executor_fn, False
# Regular run or checkpoint restoration
initial_executor_fn = functools.partial(
self._execute_with_message_or_checkpoint, message, checkpoint_id, checkpoint_storage
)
reset_context = message is not None and checkpoint_id is None
return initial_executor_fn, reset_context
async def _restore_and_send_responses(
self,
checkpoint_id: str,
checkpoint_storage: CheckpointStorage | None,
responses: dict[str, Any],
) -> None:
"""Restore from a checkpoint then send responses to pending requests.
Args:
checkpoint_id: ID of checkpoint to restore from.
checkpoint_storage: Runtime checkpoint storage.
responses: Responses to send after restoration.
"""
has_checkpointing = self._runner.context.has_checkpointing()
if not has_checkpointing and checkpoint_storage is None:
raise ValueError(
"Cannot restore from checkpoint: either provide checkpoint_storage parameter "
"or build workflow with WorkflowBuilder.with_checkpointing(checkpoint_storage)."
)
await self._runner.restore_from_checkpoint(checkpoint_id, checkpoint_storage)
await self._send_responses_internal(responses)
async def _send_responses_internal(self, responses: dict[str, Any]) -> None:
"""Internal method to validate and send responses to the executors."""
@@ -680,20 +685,24 @@ class Workflow(DictConvertible):
if not pending_requests:
raise RuntimeError("No pending requests found in workflow context.")
# Validate responses against pending requests
# Validate and coerce responses against pending requests
coerced_responses: dict[str, Any] = {}
for request_id, response in responses.items():
if request_id not in pending_requests:
raise ValueError(f"Response provided for unknown request ID: {request_id}")
pending_request = pending_requests[request_id]
# Try to coerce raw values (e.g., dicts from JSON) to the expected type
response = try_coerce_to_type(response, pending_request.response_type)
if not is_instance_of(response, pending_request.response_type):
raise ValueError(
f"Response type mismatch for request ID {request_id}: "
f"expected {pending_request.response_type}, got {type(response)}"
)
coerced_responses[request_id] = response
await asyncio.gather(*[
self._runner_context.send_request_info_response(request_id, response)
for request_id, response in responses.items()
for request_id, response in coerced_responses.items()
])
def _get_executor_by_id(self, executor_id: str) -> Executor:
python/packages/core/agent_framework/_workflows/_workflow_executor.py
+1 -1
View File
@@ -653,7 +653,7 @@ class WorkflowExecutor(Executor):
try:
# Resume the sub-workflow with all collected responses
result = await self.workflow.send_responses(responses_to_send)
result = await self.workflow.run(responses=responses_to_send)
# Remove handled requests from result. The result may contain the original
# RequestInfoEvents that were already handled. This is due to checkpointing
# and rehydration of the workflow that re-adds the RequestInfoEvents to the
python/packages/core/tests/workflow/test_agent_executor_tool_calls.py
+8 -8
View File
@@ -267,9 +267,9 @@ async def test_agent_executor_tool_call_with_approval() -> None:
assert approval_request.data.function_call.arguments == '{"query": "test"}'
# Act
events = await workflow.send_responses({
approval_request.request_id: approval_request.data.to_function_approval_response(True)
})
events = await workflow.run(
responses={approval_request.request_id: approval_request.data.to_function_approval_response(True)}
)
# Assert
final_response = events.get_outputs()
@@ -303,9 +303,9 @@ async def test_agent_executor_tool_call_with_approval_streaming() -> None:
# Act
output: str | None = None
async for event in workflow.send_responses_streaming({
approval_request.request_id: approval_request.data.to_function_approval_response(True)
}):
async for event in workflow.run(
stream=True, responses={approval_request.request_id: approval_request.data.to_function_approval_response(True)}
):
if event.type == "output":
output = event.data
@@ -346,7 +346,7 @@ async def test_agent_executor_parallel_tool_call_with_approval() -> None:
approval_request.request_id: approval_request.data.to_function_approval_response(True) # type: ignore
for approval_request in events.get_request_info_events()
}
events = await workflow.send_responses(responses)
events = await workflow.run(responses=responses)
# Assert
final_response = events.get_outputs()
@@ -385,7 +385,7 @@ async def test_agent_executor_parallel_tool_call_with_approval_streaming() -> No
}
output: str | None = None
async for event in workflow.send_responses_streaming(responses):
async for event in workflow.run(stream=True, responses=responses):
if event.type == "output":
output = event.data
python/packages/core/tests/workflow/test_request_info_and_response.py
+11 -8
View File
@@ -192,7 +192,7 @@ class TestRequestInfoAndResponse:
# Send response and continue workflow
completed = False
async for event in workflow.send_responses_streaming({request_info_event.request_id: True}):
async for event in workflow.run(stream=True, responses={request_info_event.request_id: True}):
if event.type == "status" and event.state == WorkflowRunState.IDLE:
completed = True
@@ -219,7 +219,7 @@ class TestRequestInfoAndResponse:
# Send response with calculated result
calculated_result = 31.0
completed = False
async for event in workflow.send_responses_streaming({request_info_event.request_id: calculated_result}):
async for event in workflow.run(stream=True, responses={request_info_event.request_id: calculated_result}):
if event.type == "status" and event.state == WorkflowRunState.IDLE:
completed = True
@@ -254,7 +254,7 @@ class TestRequestInfoAndResponse:
# Send responses for both requests
responses = {approval_event.request_id: True, calc_event.request_id: 50.0}
completed = False
async for event in workflow.send_responses_streaming(responses):
async for event in workflow.run(stream=True, responses=responses):
if event.type == "status" and event.state == WorkflowRunState.IDLE:
completed = True
@@ -276,7 +276,7 @@ class TestRequestInfoAndResponse:
# Deny the request
completed = False
async for event in workflow.send_responses_streaming({request_info_event.request_id: False}):
async for event in workflow.run(stream=True, responses={request_info_event.request_id: False}):
if event.type == "status" and event.state == WorkflowRunState.IDLE:
completed = True
@@ -303,7 +303,7 @@ class TestRequestInfoAndResponse:
# Continue with response
completed = False
async for event in workflow.send_responses_streaming({request_info_event.request_id: True}):
async for event in workflow.run(stream=True, responses={request_info_event.request_id: True}):
if event.type == "status" and event.state == WorkflowRunState.IDLE:
completed = True
@@ -395,9 +395,12 @@ class TestRequestInfoAndResponse:
# Step 6: Provide response to the restored request and complete the workflow
final_completed = False
async for event in restored_workflow.send_responses_streaming({
request_info_event.request_id: True # Approve the request
}):
async for event in restored_workflow.run(
stream=True,
responses={
request_info_event.request_id: True # Approve the request
},
):
if event.type == "status" and event.state == WorkflowRunState.IDLE:
final_completed = True
python/packages/core/tests/workflow/test_sub_workflow.py
+12 -8
View File
@@ -201,9 +201,11 @@ async def test_basic_sub_workflow() -> None:
assert request_events[0].data.domain == "example.com"
# Send response through the main workflow
await main_workflow.send_responses({
request_events[0].request_id: True # Domain is approved
})
await main_workflow.run(
responses={
request_events[0].request_id: True # Domain is approved
}
)
# Check result
assert parent.result is not None
@@ -245,9 +247,11 @@ async def test_sub_workflow_with_interception():
assert request_events[0].data.domain == "unknown.com"
# Send external response
await main_workflow.send_responses({
request_events[0].request_id: False # Domain not approved
})
await main_workflow.run(
responses={
request_events[0].request_id: False # Domain not approved
}
)
assert parent.result is not None
assert parent.result.email == "user@unknown.com"
assert parent.result.is_valid is False
@@ -447,7 +451,7 @@ async def test_concurrent_sub_workflow_execution() -> None:
# Send responses for all requests (approve all domains)
responses = {event.request_id: True for event in request_events}
await main_workflow.send_responses(responses)
await main_workflow.run(responses=responses)
# All results should be collected
assert len(processor.results) == len(emails)
@@ -613,7 +617,7 @@ async def test_sub_workflow_checkpoint_restore_no_duplicate_requests() -> None:
assert resumed_first_request_id == first_request_id
request_events: list[WorkflowEvent] = []
async for event in workflow2.send_responses_streaming({resumed_first_request_id: "first_answer"}):
async for event in workflow2.run(stream=True, responses={resumed_first_request_id: "first_answer"}):
if event.type == "request_info":
request_events.append(event)
python/packages/core/tests/workflow/test_typing_utils.py
+70
View File
@@ -13,6 +13,7 @@ from agent_framework._workflows._typing_utils import (
normalize_type_to_list,
resolve_type_annotation,
serialize_type,
try_coerce_to_type,
)
# region: normalize_type_to_list tests
@@ -420,3 +421,72 @@ def test_type_compatibility_complex() -> None:
# Incompatible nested structure
incompatible_target = list[dict[Union[str, bytes], int]]
assert not is_type_compatible(source, incompatible_target)
# region: try_coerce_to_type tests
def test_coerce_already_correct_type() -> None:
"""Values already matching the target type are returned as-is."""
assert try_coerce_to_type(42, int) == 42
assert try_coerce_to_type("hello", str) == "hello"
assert try_coerce_to_type(True, bool) is True
def test_coerce_int_to_float() -> None:
"""JSON integers should be coercible to float."""
result = try_coerce_to_type(1, float)
assert result == 1.0
assert isinstance(result, float)
def test_coerce_dict_to_dataclass() -> None:
"""Dicts (from JSON) should be coercible to dataclasses."""
@dataclass
class Point:
x: int
y: int
result = try_coerce_to_type({"x": 1, "y": 2}, Point)
assert isinstance(result, Point)
assert result.x == 1
assert result.y == 2
def test_coerce_dict_to_dataclass_bad_keys_returns_original() -> None:
"""Dicts with wrong keys should return the original dict, not raise."""
@dataclass
class Point:
x: int
y: int
original = {"a": 1, "b": 2}
result = try_coerce_to_type(original, Point)
assert result is original
def test_coerce_non_concrete_target_returns_original() -> None:
"""Union and other non-concrete types should return the original value."""
result = try_coerce_to_type(42, int | str)
assert result == 42
result = try_coerce_to_type({"x": 1}, Union[str, int])
assert result == {"x": 1}
def test_coerce_unrelated_types_returns_original() -> None:
"""Coercion between unrelated types should return the original value."""
assert try_coerce_to_type("hello", int) == "hello"
assert try_coerce_to_type(3.14, str) == 3.14
assert try_coerce_to_type([1, 2], dict) == [1, 2]
def test_coerce_any_returns_original() -> None:
"""Any target type should accept any value without coercion."""
assert try_coerce_to_type(42, Any) == 42
assert try_coerce_to_type({"k": "v"}, Any) == {"k": "v"}
# endregion: try_coerce_to_type tests
python/packages/core/tests/workflow/test_workflow.py
+9 -9
View File
@@ -383,7 +383,7 @@ async def test_workflow_run_stream_from_checkpoint_with_external_storage(
try:
events: list[WorkflowEvent] = []
async for event in workflow_without_checkpointing.run(
checkpoint_id=checkpoint_id, checkpoint_storage=storage
checkpoint_id=checkpoint_id, checkpoint_storage=storage, stream=True
):
events.append(event)
if len(events) >= 2: # Limit to avoid infinite loops
@@ -952,11 +952,11 @@ async def test_workflow_run_parameter_validation(simple_executor: Executor) -> N
pass
# Invalid: none of message or checkpoint_id
with pytest.raises(ValueError, match="Must provide either"):
with pytest.raises(ValueError, match="Must provide at least one of"):
await workflow.run()
# Invalid: none of message or checkpoint_id (streaming)
with pytest.raises(ValueError, match="Must provide either"):
with pytest.raises(ValueError, match="Must provide at least one of"):
async for _ in workflow.run(stream=True):
pass
@@ -1174,8 +1174,8 @@ async def test_output_executors_filtering_with_fan_in() -> None:
assert outputs[0] == 40
async def test_output_executors_filtering_with_send_responses() -> None:
"""Test output filtering works correctly with send_responses method."""
async def test_output_executors_filtering_with_run_responses() -> None:
"""Test output filtering works correctly with run(responses=...) method."""
executor = MockExecutorRequestApproval(id="approval_executor")
workflow = WorkflowBuilder().set_start_executor(executor).with_output_from([executor]).build()
@@ -1189,7 +1189,7 @@ async def test_output_executors_filtering_with_send_responses() -> None:
# Send approval response
responses = {request_events[0].request_id: ApprovalMessage(approved=True)}
response_result = await workflow.send_responses(responses)
response_result = await workflow.run(responses=responses)
outputs = response_result.get_outputs()
# Output should be yielded since approval_executor is in output_executors
@@ -1197,8 +1197,8 @@ async def test_output_executors_filtering_with_send_responses() -> None:
assert outputs[0] == 42
async def test_output_executors_filtering_with_send_responses_streaming() -> None:
"""Test output filtering works correctly with send_responses_streaming method."""
async def test_output_executors_filtering_with_run_responses_streaming() -> None:
"""Test output filtering works correctly with run(responses=..., stream=True) method."""
executor = MockExecutorRequestApproval(id="approval_executor")
workflow = WorkflowBuilder().set_start_executor(executor).build()
@@ -1218,7 +1218,7 @@ async def test_output_executors_filtering_with_send_responses_streaming() -> Non
# Send approval response via streaming
responses = {request_events[0].request_id: ApprovalMessage(approved=True)}
output_events: list[WorkflowEvent] = []
async for event in workflow.send_responses_streaming(responses):
async for event in workflow.run(responses=responses, stream=True):
if event.type == "output":
output_events.append(event)
python/packages/declarative/agent_framework_declarative/_workflows/_executors_agents.py
+1 -1
View File
@@ -755,7 +755,7 @@ class InvokeAzureAgentExecutor(DeclarativeActionExecutor):
When externalLoop.when is configured and evaluates to true after agent response,
this method emits an ExternalInputRequest via ctx.request_info() and returns.
The workflow will yield, and when the caller provides a response via
send_responses_streaming(), the handle_external_input_response handler
run(responses=..., stream=True), the handle_external_input_response handler
will continue the loop.
"""
state = await self._ensure_state_initialized(ctx, trigger)
python/packages/devui/agent_framework_devui/_executor.py
+4 -52
View File
@@ -451,8 +451,6 @@ class AgentFrameworkExecutor:
logger.info(f"Resuming workflow with HIL responses for {len(hil_responses)} request(s)")
# Unwrap primitive responses if they're wrapped in {response: value} format
from ._utils import parse_input_for_type
unwrapped_responses = {}
for request_id, response_value in hil_responses.items():
if isinstance(response_value, dict) and "response" in response_value:
@@ -461,62 +459,16 @@ class AgentFrameworkExecutor:
hil_responses = unwrapped_responses
# NOTE: Two-step approach for stateless HTTP (framework limitation):
# 1. Restore checkpoint to load pending requests into workflow's in-memory state
# 2. Then send responses using send_responses_streaming
# Future: Framework should support run(stream=True, checkpoint_id, responses) in single call
# (checkpoint_id is guaranteed to exist due to earlier validation)
logger.debug(f"Restoring checkpoint {checkpoint_id} then sending HIL responses")
logger.debug(f"Restoring checkpoint {checkpoint_id} and sending HIL responses")
try:
# Step 1: Restore checkpoint to populate workflow's in-memory pending requests
restored = False
async for _event in workflow.run(
async for event in workflow.run(
stream=True,
responses=hil_responses,
checkpoint_id=checkpoint_id,
checkpoint_storage=checkpoint_storage,
):
restored = True
break # Stop immediately after restoration, don't process events
if not restored:
raise RuntimeError("Checkpoint restoration did not yield any events")
# Reset running flags so we can call send_responses_streaming
if hasattr(workflow, "_is_running"):
workflow._is_running = False
if hasattr(workflow, "_runner") and hasattr(workflow._runner, "_running"):
workflow._runner._running = False
# Extract response types from restored workflow and convert responses to proper types
try:
if hasattr(workflow, "_runner") and hasattr(workflow._runner, "context"):
runner_context = workflow._runner.context
pending_requests_dict = await runner_context.get_pending_request_info_events()
converted_responses = {}
for request_id, response_value in hil_responses.items():
if request_id in pending_requests_dict:
pending_request = pending_requests_dict[request_id]
if hasattr(pending_request, "response_type"):
response_type = pending_request.response_type
try:
response_value = parse_input_for_type(response_value, response_type)
logger.debug(
f"Converted HIL response for {request_id} to {type(response_value)}"
)
except Exception as e:
logger.warning(f"Failed to convert HIL response for {request_id}: {e}")
converted_responses[request_id] = response_value
hil_responses = converted_responses
except Exception as e:
logger.warning(f"Could not convert HIL responses to proper types: {e}")
async for event in workflow.send_responses_streaming(hil_responses):
# Enrich new request_info events (type='request_info')
# that may come from subsequent HIL requests
# Enrich new request_info events that may come from subsequent HIL requests
if event.type == "request_info":
self._enrich_request_info_event_with_response_schema(event, workflow)
python/packages/orchestrations/agent_framework_orchestrations/_magentic.py
+1 -1
View File
@@ -1524,7 +1524,7 @@ class MagenticBuilder:
if request.kind == MagenticHumanInterventionKind.PLAN_REVIEW:
# Review plan and respond
reply = MagenticHumanInterventionReply(decision=MagenticHumanInterventionDecision.APPROVE)
await workflow.send_responses({event.request_id: reply})
await workflow.run(responses={event.request_id: reply})
See Also:
- :class:`MagenticHumanInterventionRequest`: Event emitted for review
python/packages/orchestrations/tests/test_group_chat.py
+3 -3
View File
@@ -784,9 +784,9 @@ async def test_group_chat_with_request_info_filtering():
# Continue the workflow with a response
outputs: list[WorkflowEvent] = []
async for event in workflow.send_responses_streaming({
request_event.request_id: AgentRequestInfoResponse.approve()
}):
async for event in workflow.run(
stream=True, responses={request_event.request_id: AgentRequestInfoResponse.approve()}
):
if event.type == "output":
outputs.append(event)
python/packages/orchestrations/tests/test_handoff.py
+8 -6
View File
@@ -255,9 +255,9 @@ async def test_handoff_async_termination_condition() -> None:
assert requests
events = await _drain(
workflow.send_responses_streaming({
requests[-1].request_id: [ChatMessage(role="user", text="Second user message")]
})
workflow.run(
stream=True, responses={requests[-1].request_id: [ChatMessage(role="user", text="Second user message")]}
)
)
outputs = [ev for ev in events if ev.type == "output"]
assert len(outputs) == 1
@@ -508,7 +508,7 @@ async def test_handoff_with_participant_factories():
# Follow-up message
events = await _drain(
workflow.send_responses_streaming({requests[-1].request_id: [ChatMessage(role="user", text="More details")]})
workflow.run(stream=True, responses={requests[-1].request_id: [ChatMessage(role="user", text="More details")]})
)
outputs = [ev for ev in events if ev.type == "output"]
assert outputs
@@ -582,7 +582,9 @@ async def test_handoff_with_participant_factories_and_add_handoff():
# Second user message - specialist_a hands off to specialist_b
events = await _drain(
workflow.send_responses_streaming({requests[-1].request_id: [ChatMessage(role="user", text="Need escalation")]})
workflow.run(
stream=True, responses={requests[-1].request_id: [ChatMessage(role="user", text="Need escalation")]}
)
)
requests = [ev for ev in events if ev.type == "request_info"]
assert requests
@@ -617,7 +619,7 @@ async def test_handoff_participant_factories_with_checkpointing():
assert requests
events = await _drain(
workflow.send_responses_streaming({requests[-1].request_id: [ChatMessage(role="user", text="follow up")]})
workflow.run(stream=True, responses={requests[-1].request_id: [ChatMessage(role="user", text="follow up")]})
)
outputs = [ev for ev in events if ev.type == "output"]
assert outputs, "Should have workflow output after termination condition is met"
python/packages/orchestrations/tests/test_magentic.py
+7 -6
View File
@@ -251,7 +251,7 @@ async def test_magentic_workflow_plan_review_approval_to_completion():
completed = False
output: list[ChatMessage] | None = None
async for ev in wf.send_responses_streaming(responses={req_event.request_id: req_event.data.approve()}):
async for ev in wf.run(stream=True, responses={req_event.request_id: req_event.data.approve()}):
if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
completed = True
elif ev.type == "output":
@@ -297,16 +297,17 @@ async def test_magentic_plan_review_with_revise():
# Send a revise response
saw_second_review = False
completed = False
async for ev in wf.send_responses_streaming(
responses={req_event.request_id: req_event.data.revise("Looks good; consider Z")}
async for ev in wf.run(
stream=True, responses={req_event.request_id: req_event.data.revise("Looks good; consider Z")}
):
if ev.type == "request_info" and ev.request_type is MagenticPlanReviewRequest:
saw_second_review = True
req_event = ev
# Approve the second review
async for ev in wf.send_responses_streaming(
responses={req_event.request_id: req_event.data.approve()} # type: ignore[union-attr]
async for ev in wf.run(
stream=True,
responses={req_event.request_id: req_event.data.approve()}, # type: ignore[union-attr]
):
if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
completed = True
@@ -397,7 +398,7 @@ async def test_magentic_checkpoint_resume_round_trip():
assert isinstance(req_event.data, MagenticPlanReviewRequest)
responses = {req_event.request_id: req_event.data.approve()}
async for event in wf_resume.send_responses_streaming(responses=responses):
async for event in wf_resume.run(stream=True, responses=responses):
if event.type == "output":
completed = event
assert completed is not None
python/samples/autogen-migration/orchestrations/03_swarm.py
+1 -1
View File
@@ -193,7 +193,7 @@ async def run_agent_framework() -> None:
current_executor = None
stream_line_open = False
async for event in workflow.send_responses_streaming(responses):
async for event in workflow.run(stream=True, responses=responses):
if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
# Print executor name header when switching to a new agent
if current_executor != event.executor_id:
python/samples/getting_started/orchestrations/handoff_participant_factory.py
+3 -3
View File
@@ -208,9 +208,9 @@ async def _run_workflow(workflow: Workflow, user_inputs: list[str]) -> None:
responses = {req.request_id: HandoffAgentUserRequest.terminate() for req in pending_requests}
# Send responses and get new events
# We use send_responses_streaming() to get events as they occur, allowing us to
# display agent responses in real-time and handle new requests as they arrive
workflow_result = await workflow.send_responses(responses)
# We use run(responses=...) to get events, allowing us to
# display agent responses and handle new requests as they arrive
workflow_result = await workflow.run(responses=responses)
pending_requests = _handle_events(workflow_result)
python/samples/getting_started/orchestrations/handoff_simple.py
+2 -2
View File
@@ -255,9 +255,9 @@ async def main() -> None:
}
# Send responses and get new events
# We use send_responses() to get events from the workflow, allowing us to
# We use run(responses=...) to get events from the workflow, allowing us to
# display agent responses and handle new requests as they arrive
events = await workflow.send_responses(responses)
events = await workflow.run(responses=responses)
pending_requests = _handle_events(events)
"""
python/samples/getting_started/orchestrations/handoff_with_code_interpreter_file.py
+1 -1
View File
@@ -191,7 +191,7 @@ async def main() -> None:
print(f"\nUser: {user_input}")
responses = {request.request_id: HandoffAgentUserRequest.create_response(user_input)}
events = await _drain(workflow.send_responses_streaming(responses))
events = await _drain(workflow.run(stream=True, responses=responses))
requests, file_ids = _handle_events(events)
all_file_ids.extend(file_ids)
input_index += 1
python/samples/getting_started/orchestrations/magentic_checkpoint.py
+2 -2
View File
@@ -29,7 +29,7 @@ Concepts highlighted here:
must keep stable IDs so the checkpoint state aligns when we rebuild the graph.
2. **Executor snapshotting** - checkpoints capture the pending plan-review request
map, at superstep boundaries.
3. **Resume with responses** - `Workflow.send_responses_streaming` accepts a
3. **Resume with responses** - `Workflow.run(responses=...)` accepts a
`responses` mapping so we can inject the stored human reply during restoration.
Prerequisites:
@@ -157,7 +157,7 @@ async def main() -> None:
# Supply the approval and continue to run to completion.
final_event: WorkflowEvent | None = None
async for event in resumed_workflow.send_responses_streaming({request_info_event.request_id: approval}):
async for event in resumed_workflow.run(stream=True, responses={request_info_event.request_id: approval}):
if event.type == "output":
final_event = event
python/samples/getting_started/orchestrations/magentic_human_plan_review.py
+2 -2
View File
@@ -139,14 +139,14 @@ async def main() -> None:
print("=" * 60)
# Initiate the first run of the workflow.
# Runs are not isolated; state is preserved across multiple calls to run or send_responses_streaming.
# Runs are not isolated; state is preserved across multiple calls to run.
stream = workflow.run(task, stream=True)
pending_responses = await process_event_stream(stream)
while pending_responses is not None:
# Run the workflow until there is no more human feedback to provide,
# in which case this workflow completes.
stream = workflow.send_responses_streaming(pending_responses)
stream = workflow.run(stream=True, responses=pending_responses)
pending_responses = await process_event_stream(stream)
python/samples/getting_started/workflows/agents/azure_chat_agents_tool_calls_with_feedback.py
+1 -1
View File
@@ -281,7 +281,7 @@ async def main() -> None:
)
initial_run = False
elif pending_responses is not None:
stream = workflow.send_responses_streaming(pending_responses)
stream = workflow.run(stream=True, responses=pending_responses)
pending_responses = None
else:
break
python/samples/getting_started/workflows/checkpoint/checkpoint_with_human_in_the_loop.py
+1 -1
View File
@@ -249,7 +249,7 @@ async def run_interactive_session(
while True:
if responses:
event_stream = workflow.send_responses_streaming(responses)
event_stream = workflow.run(stream=True, responses=responses)
requests.clear()
responses = None
else:
python/samples/getting_started/workflows/checkpoint/handoff_with_tool_approval_checkpoint_resume.py
+19 -18
View File
@@ -23,22 +23,24 @@ from azure.identity import AzureCliCredential
"""
Sample: Handoff Workflow with Tool Approvals + Checkpoint Resume
Demonstrates the two-step pattern for resuming a handoff workflow from a checkpoint
while handling both HandoffAgentUserRequest prompts and function approval request Content
for tool calls (e.g., submit_refund).
Demonstrates resuming a handoff workflow from a checkpoint while handling both
HandoffAgentUserRequest prompts and function approval request Content for tool calls
(e.g., submit_refund).
Scenario:
1. User starts a conversation with the workflow.
2. Agents may emit user input requests or tool approval requests.
3. Workflow writes a checkpoint capturing pending requests and pauses.
4. Process can exit/restart.
5. On resume: Load the checkpoint, surface pending approvals/user prompts, and provide responses.
5. On resume: Restore checkpoint, inspect pending requests, then provide responses.
6. Workflow continues from the saved state.
Pattern:
- Step 1: workflow.run(checkpoint_id=..., stream=True) to restore checkpoint and pending requests.
- Step 2: workflow.send_responses_streaming(responses) to supply human replies and approvals.
- Two-step approach is required because send_responses_streaming does not accept checkpoint_id.
- workflow.run(checkpoint_id=..., stream=True) to restore checkpoint and discover pending requests.
- workflow.run(stream=True, responses=responses) to supply human replies and approvals.
(Two steps are needed here because the sample must inspect request types before building responses.
When response payloads are already known, use the single-call form:
workflow.run(stream=True, checkpoint_id=..., responses=responses).)
Prerequisites:
- Azure CLI authentication (az login).
@@ -228,13 +230,13 @@ async def resume_with_responses(
approve_tools: bool | None = None,
) -> tuple[list[WorkflowEvent], str | None]:
"""
Two-step resume pattern (answers customer questions and tool approvals):
Resume from checkpoint and send responses.
Step 1: Restore checkpoint to load pending requests into workflow state
Step 2: Send user responses using send_responses_streaming
Step 1: Restore checkpoint to discover pending request types.
Step 2: Build typed responses and send via workflow.run(responses=...).
This is the current pattern required because send_responses_streaming
doesn't accept a checkpoint_id parameter.
When response payloads are already known, these can be combined into a single
workflow.run(stream=True, checkpoint_id=..., responses=...) call.
"""
print(f"\n{'=' * 60}")
print("RESUMING WORKFLOW WITH HUMAN INPUT")
@@ -253,10 +255,9 @@ async def resume_with_responses(
checkpoints.sort(key=lambda cp: cp.timestamp, reverse=True)
latest_checkpoint = checkpoints[0]
print(f"Step 1: Restoring checkpoint {latest_checkpoint.checkpoint_id}")
print(f"Restoring checkpoint {latest_checkpoint.checkpoint_id}")
# Step 1: Restore the checkpoint to load pending requests into memory
# The checkpoint restoration re-emits pending request_info events
# First, restore checkpoint to discover pending requests
restored_requests: list[WorkflowEvent] = []
async for event in workflow.run(checkpoint_id=latest_checkpoint.checkpoint_id, stream=True): # type: ignore[attr-defined]
if event.type == "request_info":
@@ -274,11 +275,11 @@ async def resume_with_responses(
user_response=user_response,
approve_tools=approve_tools,
)
print(f"Step 2: Sending responses for {len(responses)} request(s)")
print(f"Sending responses for {len(responses)} request(s)")
new_pending_requests: list[WorkflowEvent] = []
async for event in workflow.send_responses_streaming(responses):
async for event in workflow.run(stream=True, responses=responses):
if event.type == "status":
print(f"[Status] {event.state}")
@@ -309,7 +310,7 @@ async def main() -> None:
This sample shows:
1. Starting a workflow and getting a HandoffAgentUserRequest
2. Pausing (checkpoint is saved automatically)
3. Resuming from checkpoint with a user response or tool approval (two-step pattern)
3. Resuming from checkpoint with a user response or tool approval
4. Continuing the conversation until completion
"""
python/samples/getting_started/workflows/checkpoint/sub_workflow_checkpoint.py
+1 -1
View File
@@ -380,7 +380,7 @@ async def main() -> None:
approval_response = "approve"
output_event: WorkflowEvent | None = None
async for event in workflow2.send_responses_streaming({request_info_event.request_id: approval_response}):
async for event in workflow2.run(stream=True, responses={request_info_event.request_id: approval_response}):
if event.type == "output":
output_event = event
python/samples/getting_started/workflows/composition/sub_workflow_parallel_requests.py
+1 -1
View File
@@ -347,7 +347,7 @@ async def main() -> None:
else:
print(f"Unknown request info event data type: {type(event.data)}")
run_result = await main_workflow.send_responses(responses)
run_result = await main_workflow.run(responses=responses)
outputs = run_result.get_outputs()
if outputs:
python/samples/getting_started/workflows/declarative/customer_support/main.py
+1 -1
View File
@@ -251,7 +251,7 @@ async def main() -> None:
# Continue workflow with user response
print(f"\n{YELLOW}WORKFLOW:{RESET} Restore\n")
response = AgentExternalInputResponse(user_input=user_input)
stream = workflow.send_responses_streaming({pending_request_id: response})
stream = workflow.run(stream=True, responses={pending_request_id: response})
pending_request_id = None
else:
# Start workflow
python/samples/getting_started/workflows/declarative/function_tools/main.py
+1 -1
View File
@@ -90,7 +90,7 @@ async def main():
while True:
if pending_request_id:
response = ExternalInputResponse(user_input=user_input)
stream = workflow.send_responses_streaming({pending_request_id: response})
stream = workflow.run(stream=True, responses={pending_request_id: response})
else:
stream = workflow.run({"userInput": user_input}, stream=True)
python/samples/getting_started/workflows/human-in-the-loop/agents_with_HITL.py
+2 -2
View File
@@ -199,7 +199,7 @@ async def main() -> None:
)
# Initiate the first run of the workflow.
# Runs are not isolated; state is preserved across multiple calls to run or send_responses_streaming.
# Runs are not isolated; state is preserved across multiple calls to run.
stream = workflow.run(
"Create a short launch blurb for the LumenX desk lamp. Emphasize adjustability and warm lighting.",
stream=True,
@@ -209,7 +209,7 @@ async def main() -> None:
while pending_responses is not None:
# Run the workflow until there is no more human feedback to provide,
# in which case this workflow completes.
stream = workflow.send_responses_streaming(pending_responses)
stream = workflow.run(stream=True, responses=pending_responses)
pending_responses = await process_event_stream(stream)
print("\nWorkflow complete.")
python/samples/getting_started/workflows/human-in-the-loop/agents_with_approval_requests.py
+2 -2
View File
@@ -249,7 +249,7 @@ async def main() -> None:
)
# Initiate the first run of the workflow.
# Runs are not isolated; state is preserved across multiple calls to run or send_responses_streaming.
# Runs are not isolated; state is preserved across multiple calls to run.
events = await workflow.run(incoming_email)
request_info_events = events.get_request_info_events()
@@ -276,7 +276,7 @@ async def main() -> None:
print("Performing automatic approval for demo purposes...")
responses[request_info_event.request_id] = data.to_function_approval_response(approved=True)
events = await workflow.send_responses(responses)
events = await workflow.run(responses=responses)
request_info_events = events.get_request_info_events()
# The output should only come from conclude_workflow executor and it's a single string
python/samples/getting_started/workflows/human-in-the-loop/concurrent_request_info.py
+2 -2
View File
@@ -183,14 +183,14 @@ async def main() -> None:
)
# Initiate the first run of the workflow.
# Runs are not isolated; state is preserved across multiple calls to run or send_responses_streaming.
# Runs are not isolated; state is preserved across multiple calls to run.
stream = workflow.run("Analyze the impact of large language models on software development.", stream=True)
pending_responses = await process_event_stream(stream)
while pending_responses is not None:
# Run the workflow until there is no more human feedback to provide,
# in which case this workflow completes.
stream = workflow.send_responses_streaming(pending_responses)
stream = workflow.run(stream=True, responses=pending_responses)
pending_responses = await process_event_stream(stream)
python/samples/getting_started/workflows/human-in-the-loop/group_chat_request_info.py
+2 -2
View File
@@ -147,7 +147,7 @@ async def main() -> None:
)
# Initiate the first run of the workflow.
# Runs are not isolated; state is preserved across multiple calls to run or send_responses_streaming.
# Runs are not isolated; state is preserved across multiple calls to run.
stream = workflow.run(
"Discuss how our team should approach adopting AI tools for productivity. "
"Consider benefits, risks, and implementation strategies.",
@@ -158,7 +158,7 @@ async def main() -> None:
while pending_responses is not None:
# Run the workflow until there is no more human feedback to provide,
# in which case this workflow completes.
stream = workflow.send_responses_streaming(pending_responses)
stream = workflow.run(stream=True, responses=pending_responses)
pending_responses = await process_event_stream(stream)
python/samples/getting_started/workflows/human-in-the-loop/guessing_game_with_human_input.py
+5 -5
View File
@@ -29,7 +29,7 @@ the workflow completes when idle with no pending work.
Purpose:
Show how to integrate a human step in the middle of an LLM workflow by using
`request_info` and `send_responses_streaming`.
`request_info` and `run(responses=..., stream=True)`.
Demonstrate:
- Alternating turns between an AgentExecutor and a human, driven by events.
@@ -42,11 +42,11 @@ Prerequisites:
- Basic familiarity with WorkflowBuilder, executors, edges, events, and streaming runs.
"""
# How human-in-the-loop is achieved via `request_info` and `send_responses_streaming`:
# How human-in-the-loop is achieved via `request_info` and `run(responses=..., stream=True)`:
# - An executor (TurnManager) calls `ctx.request_info` with a payload (HumanFeedbackRequest).
# - The workflow run pauses and emits a with the payload and the request_id.
# - The application captures the event, prompts the user, and collects replies.
# - The application calls `send_responses_streaming` with a map of request_ids to replies.
# - The application calls `run(stream=True, responses=...)` with a map of request_ids to replies.
# - The workflow resumes, and the response is delivered to the executor method decorated with @response_handler.
# - The executor can then continue the workflow, e.g., by sending a new message to the agent.
@@ -205,14 +205,14 @@ async def main() -> None:
).build()
# Initiate the first run of the workflow.
# Runs are not isolated; state is preserved across multiple calls to run or send_responses_streaming.
# Runs are not isolated; state is preserved across multiple calls to run.
stream = workflow.run("start", stream=True)
pending_responses = await process_event_stream(stream)
while pending_responses is not None:
# Run the workflow until there is no more human feedback to provide,
# in which case this workflow completes.
stream = workflow.send_responses_streaming(pending_responses)
stream = workflow.run(stream=True, responses=pending_responses)
pending_responses = await process_event_stream(stream)
"""
python/samples/getting_started/workflows/human-in-the-loop/sequential_request_info.py
+4 -4
View File
@@ -13,8 +13,8 @@ using the standard request_info pattern for consistency.
Demonstrate:
- Configuring request info with `.with_request_info()`
- Handling with AgentInputRequest data
- Injecting responses back into the workflow via send_responses_streaming
- Handling request_info events with AgentInputRequest data
- Injecting responses back into the workflow via run(responses=..., stream=True)
Prerequisites:
- Azure OpenAI configured for AzureOpenAIChatClient with required environment variables
@@ -122,14 +122,14 @@ async def main() -> None:
)
# Initiate the first run of the workflow.
# Runs are not isolated; state is preserved across multiple calls to run or send_responses_streaming.
# Runs are not isolated; state is preserved across multiple calls to run.
stream = workflow.run("Write a brief introduction to artificial intelligence.", stream=True)
pending_responses = await process_event_stream(stream)
while pending_responses is not None:
# Run the workflow until there is no more human feedback to provide,
# in which case this workflow completes.
stream = workflow.send_responses_streaming(pending_responses)
stream = workflow.run(stream=True, responses=pending_responses)
pending_responses = await process_event_stream(stream)
python/samples/getting_started/workflows/orchestration/magentic_human_plan_review.py
+144
View File
@@ -0,0 +1,144 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import json
from typing import cast
from agent_framework import (
AgentRunUpdateEvent,
ChatAgent,
ChatMessage,
MagenticBuilder,
MagenticPlanReviewRequest,
WorkflowEvent,
)
from agent_framework.openai import OpenAIChatClient
"""
Sample: Magentic Orchestration with Human Plan Review
This sample demonstrates how humans can review and provide feedback on plans
generated by the Magentic workflow orchestrator. When plan review is enabled,
the workflow requests human approval or revision before executing each plan.
Key concepts:
- with_plan_review(): Enables human review of generated plans
- MagenticPlanReviewRequest: The event type for plan review requests
- Human can choose to: approve the plan or provide revision feedback
Plan review options:
- approve(): Accept the proposed plan and continue execution
- revise(feedback): Provide textual feedback to modify the plan
Prerequisites:
- OpenAI credentials configured for `OpenAIChatClient`.
"""
async def main() -> None:
researcher_agent = ChatAgent(
name="ResearcherAgent",
description="Specialist in research and information gathering",
instructions="You are a Researcher. You find information and gather facts.",
chat_client=OpenAIChatClient(model_id="gpt-4o"),
)
analyst_agent = ChatAgent(
name="AnalystAgent",
description="Data analyst who processes and summarizes research findings",
instructions="You are an Analyst. You analyze findings and create summaries.",
chat_client=OpenAIChatClient(model_id="gpt-4o"),
)
manager_agent = ChatAgent(
name="MagenticManager",
description="Orchestrator that coordinates the workflow",
instructions="You coordinate a team to complete tasks efficiently.",
chat_client=OpenAIChatClient(model_id="gpt-4o"),
)
print("\nBuilding Magentic Workflow with Human Plan Review...")
workflow = (
MagenticBuilder()
.participants([researcher_agent, analyst_agent])
.with_manager(
agent=manager_agent,
max_round_count=10,
max_stall_count=1,
max_reset_count=2,
)
.with_plan_review() # Request human input for plan review
.build()
)
task = "Research sustainable aviation fuel technology and summarize the findings."
print(f"\nTask: {task}")
print("\nStarting workflow execution...")
print("=" * 60)
pending_request: WorkflowEvent | None = None
pending_responses: dict[str, object] | None = None
output_event: WorkflowEvent | None = None
while not output_event:
if pending_responses is not None:
stream = workflow.run(stream=True, responses=pending_responses)
else:
stream = workflow.run(task, stream=True)
last_message_id: str | None = None
async for event in stream:
if isinstance(event, AgentRunUpdateEvent):
message_id = event.data.message_id
if message_id != last_message_id:
if last_message_id is not None:
print("\n")
print(f"- {event.executor_id}:", end=" ", flush=True)
last_message_id = message_id
print(event.data, end="", flush=True)
elif event.type == "request_info" and event.request_type is MagenticPlanReviewRequest:
pending_request = event
elif event.type == "output":
output_event = event
pending_responses = None
# Handle plan review request if any
if pending_request is not None:
event_data = cast(MagenticPlanReviewRequest, pending_request.data)
print("\n\n[Magentic Plan Review Request]")
if event_data.current_progress is not None:
print("Current Progress Ledger:")
print(json.dumps(event_data.current_progress.to_dict(), indent=2))
print()
print(f"Proposed Plan:\n{event_data.plan.text}\n")
print("Please provide your feedback (press Enter to approve):")
reply = await asyncio.get_event_loop().run_in_executor(None, input, "> ")
if reply.strip() == "":
print("Plan approved.\n")
pending_responses = {pending_request.request_id: event_data.approve()}
else:
print("Plan revised by human.\n")
pending_responses = {pending_request.request_id: event_data.revise(reply)}
pending_request = None
print("\n" + "=" * 60)
print("WORKFLOW COMPLETED")
print("=" * 60)
print("Final Output:")
# The output of the Magentic workflow is a list of ChatMessages with only one final message
# generated by the orchestrator.
output_messages = cast(list[ChatMessage], output_event.data)
if output_messages:
output = output_messages[-1].text
print(output)
if __name__ == "__main__":
asyncio.run(main())
python/samples/getting_started/workflows/tool-approval/concurrent_builder_tool_approval.py
+2 -2
View File
@@ -155,7 +155,7 @@ async def main() -> None:
print("-" * 60)
# Initiate the first run of the workflow.
# Runs are not isolated; state is preserved across multiple calls to run or send_responses_streaming.
# Runs are not isolated; state is preserved across multiple calls to run.
stream = workflow.run(
"Manage my portfolio. Use a max of 5000 dollars to adjust my position using "
"your best judgment based on market sentiment. No need to confirm trades with me.",
@@ -166,7 +166,7 @@ async def main() -> None:
while pending_responses is not None:
# Run the workflow until there is no more human feedback to provide,
# in which case this workflow completes.
stream = workflow.send_responses_streaming(pending_responses)
stream = workflow.run(stream=True, responses=pending_responses)
pending_responses = await process_event_stream(stream)
"""
python/samples/getting_started/workflows/tool-approval/group_chat_builder_tool_approval.py
+2 -2
View File
@@ -165,7 +165,7 @@ async def main() -> None:
print("-" * 60)
# Initiate the first run of the workflow.
# Runs are not isolated; state is preserved across multiple calls to run or send_responses_streaming.
# Runs are not isolated; state is preserved across multiple calls to run.
stream = workflow.run(
"We need to deploy version 2.4.0 to production. Please coordinate the deployment.", stream=True
)
@@ -174,7 +174,7 @@ async def main() -> None:
while pending_responses is not None:
# Run the workflow until there is no more human feedback to provide,
# in which case this workflow completes.
stream = workflow.send_responses_streaming(pending_responses)
stream = workflow.run(stream=True, responses=pending_responses)
pending_responses = await process_event_stream(stream)
"""
python/samples/getting_started/workflows/tool-approval/sequential_builder_tool_approval.py
+4 -4
View File
@@ -34,8 +34,8 @@ requiring any additional builder configuration.
Demonstrate:
- Using @tool(approval_mode="always_require") for sensitive operations.
- Handling with function_approval_request Content in sequential workflows.
- Resuming workflow execution after approval via send_responses_streaming.
- Handling request_info events with function_approval_request Content in sequential workflows.
- Resuming workflow execution after approval via run(responses=..., stream=True).
Prerequisites:
- OpenAI or Azure OpenAI configured with the required environment variables.
@@ -118,7 +118,7 @@ async def main() -> None:
print("-" * 60)
# Initiate the first run of the workflow.
# Runs are not isolated; state is preserved across multiple calls to run or send_responses_streaming.
# Runs are not isolated; state is preserved across multiple calls to run.
stream = workflow.run(
"Check the schema and then update all orders with status 'pending' to 'processing'", stream=True
)
@@ -127,7 +127,7 @@ async def main() -> None:
while pending_responses is not None:
# Run the workflow until there is no more human feedback to provide,
# in which case this workflow completes.
stream = workflow.send_responses_streaming(pending_responses)
stream = workflow.run(stream=True, responses=pending_responses)
pending_responses = await process_event_stream(stream)
"""
python/samples/semantic-kernel-migration/orchestrations/handoff.py
+1 -1
View File
@@ -252,7 +252,7 @@ async def run_agent_framework_example(initial_task: str, scripted_responses: Seq
except StopIteration:
user_reply = "Thanks, that's all."
responses = {request.request_id: user_reply for request in pending}
final_events = await _drain_events(workflow.send_responses_streaming(responses))
final_events = await _drain_events(workflow.run(stream=True, responses=responses))
pending = _collect_handoff_requests(final_events)
conversation = _extract_final_conversation(final_events)