agent/agent-framework
1
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: Replace RequestInfoExecutor with request_info API and @response_handler (#1466)

Browse Source 
* Prototype: Add request_info API and @response_handler

* Add original_request as a parameter to the response handler

* Prototype: request interception in sub workflows

* Prototype: request interception in sub workflows 2

* WIP: Make checkpointing work

* checkpointing with sub workflow

* Fix function executor

* Allow sub-workflow to output directly

* Remove ReqeustInfoExecutor and related classes; Debugging checkpoint_with_human_in_the_loop

* Fix Handoff and sample

* fix pending requests in checkpoint

* Fix unit tests

* Fix formatting

* Resolve comments

* Address comment

* Add checkpoint tests

* Add tests

* misc

* fix mypy

* fix mypy

* Use request type as part of the key

* Log warning if there is not response handler for a request

* Update Internal edge group comments

* REcord message type in executor processing span

* Update sample

* Improve tests
This commit is contained in:
Tao Chen
2025-10-29 16:31:23 -07:00
committed by GitHub
Unverified
parent f6eadd412e
commit 943d92674e
54 changed files with 7532 additions and 6684 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/samples/_run_all_samples.py
+98 -29
View File
@@ -56,9 +56,9 @@ def run_sample(
sample_path: Path,
use_uv: bool = True,
python_root: Path | None = None,
) -> tuple[bool, str, str]:
) -> tuple[bool, str, str, str]:
"""
Run a single sample file using subprocess and return (success, output, error_info).
Run a single sample file using subprocess and return (success, output, error_info, error_type).
Args:
sample_path: Path to the sample file
@@ -66,7 +66,8 @@ def run_sample(
python_root: Root directory for uv run
Returns:
Tuple of (success, output, error_info)
Tuple of (success, output, error_info, error_type)
error_type can be: "timeout", "input_hang", "execution_error", "exception"
"""
if use_uv and python_root:
cmd = ["uv", "run", "python", str(sample_path)]
@@ -75,29 +76,69 @@ def run_sample(
cmd = [sys.executable, sample_path.name]
cwd = sample_path.parent
# Set environment variables to handle Unicode properly
env = os.environ.copy()
env["PYTHONIOENCODING"] = "utf-8" # Force Python to use UTF-8 for I/O
env["PYTHONUTF8"] = "1" # Enable UTF-8 mode in Python 3.7+
try:
result = subprocess.run(
cmd,
cwd=cwd,
capture_output=True,
text=True,
timeout=60, # 60 second timeout
# Use Popen for better timeout handling with stdin for samples that may wait for input
# Popen gives us more control over process lifecycle compared to subprocess.run()
process = subprocess.Popen(
cmd, # Command to execute as a list [program, arg1, arg2, ...]
cwd=cwd, # Working directory for the subprocess
stdout=subprocess.PIPE, # Capture stdout so we can read the output
stderr=subprocess.PIPE, # Capture stderr so we can read error messages
stdin=subprocess.PIPE, # Create a pipe for stdin so we can send input
text=True, # Handle input/output as text strings (not bytes)
encoding="utf-8", # Use UTF-8 encoding to handle Unicode characters like emojis
errors="replace", # Replace problematic characters instead of failing
env=env, # Pass environment variables for proper Unicode handling
)
if result.returncode == 0:
output = result.stdout.strip() if result.stdout.strip() else "No output"
return True, output, ""
try:
# communicate() sends input to stdin and waits for process to complete
# input="" sends an empty string to stdin, which causes input() calls to
# immediately receive EOFError (End Of File) since there's no data to read.
# This prevents the process from hanging indefinitely waiting for user input.
stdout, stderr = process.communicate(input="", timeout=60)
except subprocess.TimeoutExpired:
# If the process doesn't complete within the timeout period, we need to
# forcibly terminate it. This is especially important for processes that
# ignore EOFError and continue to hang on input() calls.
error_info = f"Exit code: {result.returncode}"
if result.stderr.strip():
error_info += f"\nSTDERR: {result.stderr}"
# First attempt: Send SIGKILL (immediate termination) on Unix or TerminateProcess on Windows
process.kill()
try:
# Give the process a few seconds to clean up after being killed
stdout, stderr = process.communicate(timeout=5)
except subprocess.TimeoutExpired:
# If the process is still alive after kill(), use terminate() as a last resort
# terminate() sends SIGTERM (graceful termination request) which may work
# when kill() doesn't on some systems
process.terminate()
stdout, stderr = "", "Process forcibly terminated"
return False, "", f"TIMEOUT: {sample_path.name} (exceeded 60 seconds)", "timeout"
return False, result.stdout.strip() if result.stdout.strip() else "", error_info
if process.returncode == 0:
output = stdout.strip() if stdout.strip() else "No output"
return True, output, "", "success"
except subprocess.TimeoutExpired:
return False, "", f"TIMEOUT: {sample_path.name} (exceeded 60 seconds)"
error_info = f"Exit code: {process.returncode}"
if stderr.strip():
error_info += f"\nSTDERR: {stderr}"
# Check if this looks like an input/interaction related error
error_type = "execution_error"
stderr_safe = stderr.encode("utf-8", errors="replace").decode("utf-8") if stderr else ""
if "EOFError" in stderr_safe or "input" in stderr_safe.lower() or "stdin" in stderr_safe.lower():
error_type = "input_hang"
elif "UnicodeEncodeError" in stderr_safe and ("charmap" in stderr_safe or "codec can't encode" in stderr_safe):
error_type = "input_hang" # Unicode errors often indicate interactive samples with emojis
return False, stdout.strip() if stdout.strip() else "", error_info, error_type
except Exception as e:
return False, "", f"ERROR: {sample_path.name} - Exception: {str(e)}"
return False, "", f"ERROR: {sample_path.name} - Exception: {str(e)}", "exception"
def parse_arguments() -> argparse.Namespace:
@@ -161,7 +202,7 @@ def main() -> None:
print(f"Found {len(sample_files)} Python sample files")
# Run samples concurrently
results: list[tuple[Path, bool, str, str]] = []
results: list[tuple[Path, bool, str, str, str]] = []
with ThreadPoolExecutor(max_workers=args.max_workers) as executor:
# Submit all tasks
@@ -174,53 +215,81 @@ def main() -> None:
for future in as_completed(future_to_sample):
sample_path = future_to_sample[future]
try:
success, output, error_info = future.result()
results.append((sample_path, success, output, error_info))
success, output, error_info, error_type = future.result()
results.append((sample_path, success, output, error_info, error_type))
# Print progress - show relative path from samples directory
relative_path = sample_path.relative_to(samples_dir)
if success:
print(f"✅ {relative_path}")
else:
print(f"❌ {relative_path} - {error_info.split(':', 1)[0]}")
# Show error type in progress display
error_display = f"{error_type.upper()}" if error_type != "execution_error" else "ERROR"
print(f"❌ {relative_path} - {error_display}")
except Exception as e:
error_info = f"Future exception: {str(e)}"
results.append((sample_path, False, "", error_info))
results.append((sample_path, False, "", error_info, "exception"))
relative_path = sample_path.relative_to(samples_dir)
print(f"❌ {relative_path} - {error_info}")
print(f"❌ {relative_path} - EXCEPTION")
# Sort results by original file order for consistent reporting
sample_to_index = {path: i for i, path in enumerate(sample_files)}
results.sort(key=lambda x: sample_to_index[x[0]])
successful_runs = sum(1 for _, success, _, _ in results if success)
successful_runs = sum(1 for _, success, _, _, _ in results if success)
failed_runs = len(results) - successful_runs
# Categorize failures by type
timeout_failures = [r for r in results if not r[1] and r[4] == "timeout"]
input_hang_failures = [r for r in results if not r[1] and r[4] == "input_hang"]
execution_errors = [r for r in results if not r[1] and r[4] == "execution_error"]
exceptions = [r for r in results if not r[1] and r[4] == "exception"]
# Print detailed results
print(f"\n{'=' * 80}")
print("DETAILED RESULTS:")
print(f"{'=' * 80}")
for sample_path, success, output, error_info in results:
for sample_path, success, output, error_info, error_type in results:
relative_path = sample_path.relative_to(samples_dir)
if success:
print(f"✅ {relative_path}")
if output and output != "No output":
print(f" Output preview: {output[:100]}{'...' if len(output) > 100 else ''}")
else:
print(f"❌ {relative_path}")
# Display error with type indicator
if error_type == "timeout":
print(f"⏱️ {relative_path} - TIMEOUT (likely waiting for input)")
elif error_type == "input_hang":
print(f"⌨️ {relative_path} - INPUT ERROR (interactive sample)")
elif error_type == "exception":
print(f"💥 {relative_path} - EXCEPTION")
else:
print(f"❌ {relative_path} - EXECUTION ERROR")
print(f" Error: {error_info}")
# Print summary
# Print categorized summary
print(f"\n{'=' * 80}")
if failed_runs == 0:
print("🎉 ALL SAMPLES COMPLETED SUCCESSFULLY!")
else:
print(f"❌ {failed_runs} SAMPLE(S) FAILED!")
print(f"Successful runs: {successful_runs}")
print(f"Failed runs: {failed_runs}")
if failed_runs > 0:
print("\nFailure breakdown:")
if len(timeout_failures) > 0:
print(f" ⏱️ Timeouts (likely interactive): {len(timeout_failures)}")
if len(input_hang_failures) > 0:
print(f" ⌨️ Input errors (interactive): {len(input_hang_failures)}")
if len(execution_errors) > 0:
print(f" ❌ Execution errors: {len(execution_errors)}")
if len(exceptions) > 0:
print(f" 💥 Exceptions: {len(exceptions)}")
if args.subdir:
print(f"Subdirectory filter: {args.subdir}")
python/samples/getting_started/workflows/README.md
+2 -2
View File
@@ -37,7 +37,7 @@ Once comfortable with these, explore the rest of the samples below.
| Azure Chat Agents (Streaming) | [agents/azure_chat_agents_streaming.py](./agents/azure_chat_agents_streaming.py) | Add Azure Chat agents as edges and handle streaming events |
| Azure AI Chat Agents (Streaming) | [agents/azure_ai_agents_streaming.py](./agents/azure_ai_agents_streaming.py) | Add Azure AI agents as edges and handle streaming events |
| Azure Chat Agents (Function Bridge) | [agents/azure_chat_agents_function_bridge.py](./agents/azure_chat_agents_function_bridge.py) | Chain two agents with a function executor that injects external context |
| Azure Chat Agents (Tools + HITL) | [agents/azure_chat_agents_tool_calls_with_feedback.py](./agents/azure_chat_agents_tool_calls_with_feedback.py) | Tool-enabled writer/editor pipeline with human feedback gating via RequestInfoExecutor |
| Azure Chat Agents (Tools + HITL) | [agents/azure_chat_agents_tool_calls_with_feedback.py](./agents/azure_chat_agents_tool_calls_with_feedback.py) | Tool-enabled writer/editor pipeline with human feedback gating |
| Custom Agent Executors | [agents/custom_agent_executors.py](./agents/custom_agent_executors.py) | Create executors to handle agent run methods |
| Sequential Workflow as Agent | [agents/sequential_workflow_as_agent.py](./agents/sequential_workflow_as_agent.py) | Build a sequential workflow orchestrating agents, then expose it as a reusable agent |
| Concurrent Workflow as Agent | [agents/concurrent_workflow_as_agent.py](./agents/concurrent_workflow_as_agent.py) | Build a concurrent fan-out/fan-in workflow, then expose it as a reusable agent |
@@ -58,7 +58,7 @@ Once comfortable with these, explore the rest of the samples below.
| Sample | File | Concepts |
|---|---|---|
| Sub-Workflow (Basics) | [composition/sub_workflow_basics.py](./composition/sub_workflow_basics.py) | Wrap a workflow as an executor and orchestrate sub-workflows |
| Sub-Workflow: Request Interception | [composition/sub_workflow_request_interception.py](./composition/sub_workflow_request_interception.py) | Intercept and forward sub-workflow requests using @handler for RequestInfoMessage subclasses |
| Sub-Workflow: Request Interception | [composition/sub_workflow_request_interception.py](./composition/sub_workflow_request_interception.py) | Intercept and forward sub-workflow requests using @handler for SubWorkflowRequestMessage |
| Sub-Workflow: Parallel Requests | [composition/sub_workflow_parallel_requests.py](./composition/sub_workflow_parallel_requests.py) | Multiple specialized interceptors handling different request types from same sub-workflow |
### control-flow
python/samples/getting_started/workflows/agents/azure_chat_agents_tool_calls_with_feedback.py
+112 -73
View File
@@ -8,32 +8,32 @@ from typing import Annotated
from agent_framework import (
AgentExecutorRequest,
AgentExecutorResponse,
AgentRunResponse,
AgentRunUpdateEvent,
ChatMessage,
Executor,
FunctionCallContent,
FunctionResultContent,
RequestInfoEvent,
RequestInfoExecutor,
RequestInfoMessage,
RequestResponse,
Role,
ToolMode,
WorkflowBuilder,
WorkflowContext,
WorkflowOutputEvent,
handler,
response_handler,
)
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential
from pydantic import Field
from typing_extensions import Never
"""
Sample: Tool-enabled agents with human feedback
Pipeline layout:
writer_agent (uses Azure OpenAI tools) -> DraftFeedbackCoordinator -> RequestInfoExecutor
-> DraftFeedbackCoordinator -> final_editor_agent
writer_agent (uses Azure OpenAI tools) -> Coordinator -> writer_agent
-> Coordinator -> final_editor_agent -> Coordinator -> output
The writer agent calls tools to gather product facts before drafting copy. A custom executor
packages the draft and emits a RequestInfoEvent so a human can comment, then replays the human
@@ -41,7 +41,7 @@ guidance back into the conversation before the final editor agent produces the p
Demonstrates:
- Attaching Python function tools to an agent inside a workflow.
- Capturing the writer's output and routing it through RequestInfoExecutor for human review.
- Capturing the writer's output for human review.
- Streaming AgentRunUpdateEvent updates alongside human-in-the-loop pauses.
Prerequisites:
@@ -82,27 +82,37 @@ def get_brand_voice_profile(
@dataclass
class DraftFeedbackRequest(RequestInfoMessage):
"""Payload sent to RequestInfoExecutor for human review."""
class DraftFeedbackRequest:
"""Payload sent for human review."""
prompt: str = ""
draft_text: str = ""
conversation: list[ChatMessage] = field(default_factory=list) # type: ignore[reportUnknownVariableType]
class DraftFeedbackCoordinator(Executor):
class Coordinator(Executor):
"""Bridge between the writer agent, human feedback, and final editor."""
def __init__(self, *, id: str = "draft_feedback_coordinator") -> None:
def __init__(self, id: str, writer_id: str, final_editor_id: str) -> None:
super().__init__(id)
self.writer_id = writer_id
self.final_editor_id = final_editor_id
@handler
async def on_writer_response(
self,
draft: AgentExecutorResponse,
ctx: WorkflowContext[DraftFeedbackRequest],
ctx: WorkflowContext[Never, AgentRunResponse],
) -> None:
# Preserve the full conversation so the final editor can see tool traces and the initial prompt.
"""Handle responses from the other two agents in the workflow."""
if draft.executor_id == self.final_editor_id:
# Final editor response; yield output directly.
await ctx.yield_output(draft.agent_run_response)
return
# Writer agent response; request human feedback.
# Preserve the full conversation so the final editor
# can see tool traces and the initial prompt.
conversation: list[ChatMessage]
if draft.full_conversation is not None:
conversation = list(draft.full_conversation)
@@ -117,18 +127,34 @@ class DraftFeedbackCoordinator(Executor):
"(tone tweaks, must-have detail, target audience, etc.). "
"Keep it under 30 words."
)
await ctx.send_message(DraftFeedbackRequest(prompt=prompt, draft_text=draft_text, conversation=conversation))
await ctx.request_info(
DraftFeedbackRequest(prompt=prompt, draft_text=draft_text, conversation=conversation),
DraftFeedbackRequest,
str,
)
@handler
@response_handler
async def on_human_feedback(
self,
feedback: RequestResponse[DraftFeedbackRequest, str],
original_request: DraftFeedbackRequest,
feedback: str,
ctx: WorkflowContext[AgentExecutorRequest],
) -> None:
note = (feedback.data or "").strip()
request = feedback.original_request
note = feedback.strip()
if note.lower() == "approve":
# Human approved the draft as-is; forward it unchanged.
await ctx.send_message(
AgentExecutorRequest(
messages=original_request.conversation
+ [ChatMessage(Role.USER, text="The draft is approved as-is.")],
should_respond=True,
),
target_id=self.final_editor_id,
)
return
conversation: list[ChatMessage] = list(request.conversation)
# Human provided feedback; prompt the writer to revise.
conversation: list[ChatMessage] = list(original_request.conversation)
instruction = (
"A human reviewer shared the following guidance:\n"
f"{note or 'No specific guidance provided.'}\n\n"
@@ -136,11 +162,57 @@ class DraftFeedbackCoordinator(Executor):
"Keep the response under 120 words and reflect any requested tone adjustments."
)
conversation.append(ChatMessage(Role.USER, text=instruction))
await ctx.send_message(AgentExecutorRequest(messages=conversation, should_respond=True))
await ctx.send_message(
AgentExecutorRequest(messages=conversation, should_respond=True), target_id=self.writer_id
)
def display_agent_run_update(event: AgentRunUpdateEvent, last_executor: str | None) -> None:
"""Display an AgentRunUpdateEvent in a readable format."""
printed_tool_calls: set[str] = set()
printed_tool_results: set[str] = set()
executor_id = event.executor_id
update = event.data
# Extract and print any new tool calls or results from the update.
function_calls = [c for c in update.contents if isinstance(c, FunctionCallContent)] # type: ignore[union-attr]
function_results = [c for c in update.contents if isinstance(c, FunctionResultContent)] # type: ignore[union-attr]
if executor_id != last_executor:
if last_executor is not None:
print()
print(f"{executor_id}:", end=" ", flush=True)
last_executor = executor_id
# Print any new tool calls before the text update.
for call in function_calls:
if call.call_id in printed_tool_calls:
continue
printed_tool_calls.add(call.call_id)
args = call.arguments
args_preview = json.dumps(args, ensure_ascii=False) if isinstance(args, dict) else (args or "").strip()
print(
f"\n{executor_id} [tool-call] {call.name}({args_preview})",
flush=True,
)
print(f"{executor_id}:", end=" ", flush=True)
# Print any new tool results before the text update.
for result in function_results:
if result.call_id in printed_tool_results:
continue
printed_tool_results.add(result.call_id)
result_text = result.result
if not isinstance(result_text, str):
result_text = json.dumps(result_text, ensure_ascii=False)
print(
f"\n{executor_id} [tool-result] {result.call_id}: {result_text}",
flush=True,
)
print(f"{executor_id}:", end=" ", flush=True)
# Finally, print the text update.
print(update, end="", flush=True)
async def main() -> None:
"""Run the workflow and bridge human feedback between two agents."""
# Create agents with tools and instructions.
chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())
writer_agent = chat_client.create_agent(
@@ -157,33 +229,39 @@ async def main() -> None:
final_editor_agent = chat_client.create_agent(
name="final_editor_agent",
instructions=(
"You are an editor who polishes marketing copy using human guidance. "
"Respect factual details from the prior messages while applying the feedback."
"You are an editor who polishes marketing copy after human approval. "
"Correct any legal or factual issues. Return the final version even if no changes are made. "
),
)
feedback_coordinator = DraftFeedbackCoordinator()
request_info_executor = RequestInfoExecutor(id="human_feedback")
coordinator = Coordinator(
id="coordinator",
writer_id="writer_agent",
final_editor_id="final_editor_agent",
)
# Build the workflow.
workflow = (
WorkflowBuilder()
.set_start_executor(writer_agent)
.add_edge(writer_agent, feedback_coordinator)
.add_edge(feedback_coordinator, request_info_executor)
.add_edge(request_info_executor, feedback_coordinator)
.add_edge(feedback_coordinator, final_editor_agent)
.add_edge(writer_agent, coordinator)
.add_edge(coordinator, writer_agent)
.add_edge(final_editor_agent, coordinator)
.add_edge(coordinator, final_editor_agent)
.build()
)
# Switch to turn on agent run update display.
# By default this is off to reduce clutter during human input.
display_agent_run_update_switch = False
print(
"Interactive mode. When prompted, provide a short feedback note for the editor (type 'exit' to quit).",
"Interactive mode. When prompted, provide a short feedback note for the editor.",
flush=True,
)
pending_responses: dict[str, str] | None = None
completed = False
printed_tool_calls: set[str] = set()
printed_tool_results: set[str] = set()
while not completed:
last_executor: str | None = None
@@ -198,48 +276,9 @@ async def main() -> None:
requests: list[tuple[str, DraftFeedbackRequest]] = []
async for event in stream:
if isinstance(event, AgentRunUpdateEvent):
executor_id = event.executor_id
update = event.data
# Extract and print any new tool calls or results from the update.
function_calls = [c for c in update.contents if isinstance(c, FunctionCallContent)] # type: ignore[union-attr]
function_results = [c for c in update.contents if isinstance(c, FunctionResultContent)] # type: ignore[union-attr]
if executor_id != last_executor:
if last_executor is not None:
print()
print(f"{executor_id}:", end=" ", flush=True)
last_executor = executor_id
# Print any new tool calls before the text update.
for call in function_calls:
if call.call_id in printed_tool_calls:
continue
printed_tool_calls.add(call.call_id)
args = call.arguments
if isinstance(args, dict):
args_preview = json.dumps(args, ensure_ascii=False)
else:
args_preview = (args or "").strip()
print(
f"\n{executor_id} [tool-call] {call.name}({args_preview})",
flush=True,
)
print(f"{executor_id}:", end=" ", flush=True)
# Print any new tool results before the text update.
for result in function_results:
if result.call_id in printed_tool_results:
continue
printed_tool_results.add(result.call_id)
result_text = result.result
if not isinstance(result_text, str):
result_text = json.dumps(result_text, ensure_ascii=False)
print(
f"\n{executor_id} [tool-result] {result.call_id}: {result_text}",
flush=True,
)
print(f"{executor_id}:", end=" ", flush=True)
# Finally, print the text update.
print(update, end="", flush=True)
elif isinstance(event, RequestInfoEvent) and isinstance(event.data, DraftFeedbackRequest):
if isinstance(event, AgentRunUpdateEvent) and display_agent_run_update_switch:
display_agent_run_update(event, last_executor)
if isinstance(event, RequestInfoEvent) and isinstance(event.data, DraftFeedbackRequest):
# Stash the request so we can prompt the human after the stream completes.
requests.append((event.request_id, event.data))
last_executor = None
@@ -256,7 +295,7 @@ async def main() -> None:
for request_id, request in requests:
print("\n----- Writer draft -----")
print(request.draft_text.strip())
print("\nProvide guidance for the editor (or press Enter to accept the draft).")
print("\nProvide guidance for the editor (or 'approve' to accept the draft).")
answer = input("Human feedback: ").strip() # noqa: ASYNC250
if answer.lower() == "exit":
print("Exiting...")
python/samples/getting_started/workflows/agents/workflow_as_agent_human_in_the_loop.py
+23 -29
View File
@@ -7,6 +7,9 @@ from dataclasses import dataclass
from pathlib import Path
from typing import Any
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential
# Ensure local getting_started package can be imported when running as a script.
_SAMPLES_ROOT = Path(__file__).resolve().parents[3]
if str(_SAMPLES_ROOT) not in sys.path:
@@ -17,16 +20,13 @@ from agent_framework import ( # noqa: E402
Executor,
FunctionCallContent,
FunctionResultContent,
RequestInfoExecutor,
RequestInfoMessage,
RequestResponse,
Role,
WorkflowAgent,
WorkflowBuilder,
WorkflowContext,
handler,
response_handler,
)
from agent_framework.openai import OpenAIChatClient # noqa: E402
from getting_started.workflows.agents.workflow_as_agent_reflection_pattern import ( # noqa: E402
ReviewRequest,
ReviewResponse,
@@ -40,20 +40,20 @@ Purpose:
This sample demonstrates how to build a workflow agent that escalates uncertain
decisions to a human manager. A Worker generates results, while a Reviewer
evaluates them. When the Reviewer is not confident, it escalates the decision
to a human via RequestInfoExecutor, receives the human response, and then
forwards that response back to the Worker. The workflow completes when idle.
to a human, receives the human response, and then forwards that response back
to the Worker. The workflow completes when idle.
Prerequisites:
- OpenAI account configured and accessible for OpenAIChatClient.
- Familiarity with WorkflowBuilder, Executor, and WorkflowContext from agent_framework.
- Understanding of request-response message handling (RequestInfoMessage, RequestResponse).
- Understanding of request-response message handling in executors.
- (Optional) Review of reflection and escalation patterns, such as those in
workflow_as_agent_reflection.py.
"""
@dataclass
class HumanReviewRequest(RequestInfoMessage):
class HumanReviewRequest:
"""A request message type for escalation to a human reviewer."""
agent_request: ReviewRequest | None = None
@@ -62,14 +62,13 @@ class HumanReviewRequest(RequestInfoMessage):
class ReviewerWithHumanInTheLoop(Executor):
"""Executor that always escalates reviews to a human manager."""
def __init__(self, worker_id: str, request_info_id: str, reviewer_id: str | None = None) -> None:
def __init__(self, worker_id: str, reviewer_id: str | None = None) -> None:
unique_id = reviewer_id or f"{worker_id}-reviewer"
super().__init__(id=unique_id)
self._worker_id = worker_id
self._request_info_id = request_info_id
@handler
async def review(self, request: ReviewRequest, ctx: WorkflowContext[ReviewResponse | HumanReviewRequest]) -> None:
async def review(self, request: ReviewRequest, ctx: WorkflowContext) -> None:
# In this simplified example, we always escalate to a human manager.
# See workflow_as_agent_reflection.py for an implementation
# using an automated agent to make the review decision.
@@ -77,23 +76,21 @@ class ReviewerWithHumanInTheLoop(Executor):
print("Reviewer: Escalating to human manager...")
# Forward the request to a human manager by sending a HumanReviewRequest.
await ctx.send_message(
HumanReviewRequest(agent_request=request),
target_id=self._request_info_id,
)
await ctx.request_info(HumanReviewRequest(agent_request=request), HumanReviewRequest, ReviewResponse)
@handler
@response_handler
async def accept_human_review(
self, response: RequestResponse[HumanReviewRequest, ReviewResponse], ctx: WorkflowContext[ReviewResponse]
self,
original_request: ReviewRequest,
response: ReviewResponse,
ctx: WorkflowContext[ReviewResponse],
) -> None:
# Accept the human review response and forward it back to the Worker.
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(f"Reviewer: Accepting human review for request {response.request_id[:8]}...")
print(f"Reviewer: Human feedback: {response.feedback}")
print(f"Reviewer: Human approved: {response.approved}")
print("Reviewer: Forwarding human review back to worker...")
await ctx.send_message(human_response, target_id=self._worker_id)
await ctx.send_message(response, target_id=self._worker_id)
async def main() -> None:
@@ -102,20 +99,17 @@ async def main() -> None:
# Create executors for the workflow.
print("Creating chat client and executors...")
mini_chat_client = OpenAIChatClient(model_id="gpt-4.1-nano")
mini_chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())
worker = Worker(id="sub-worker", chat_client=mini_chat_client)
request_info_executor = RequestInfoExecutor(id="request_info")
reviewer = ReviewerWithHumanInTheLoop(worker_id=worker.id, request_info_id=request_info_executor.id)
reviewer = ReviewerWithHumanInTheLoop(worker_id=worker.id)
print("Building workflow with Worker ↔ Reviewer cycle...")
print("Building workflow with Worker-Reviewer cycle...")
# Build a workflow with bidirectional communication between Worker and Reviewer,
# and escalation paths for human review.
agent = (
WorkflowBuilder()
.add_edge(worker, reviewer) # Worker sends requests to Reviewer
.add_edge(reviewer, worker) # Reviewer sends feedback to Worker
.add_edge(reviewer, request_info_executor) # Reviewer requests human input
.add_edge(request_info_executor, reviewer) # Human input forwarded back to Reviewer
.set_start_executor(worker)
.build()
.as_agent() # Convert workflow into an agent interface
python/samples/getting_started/workflows/checkpoint/checkpoint_with_human_in_the_loop.py
+95 -234
View File
@@ -1,11 +1,13 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
from collections.abc import AsyncIterable
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any
# NOTE: the Azure client imports above are real dependencies. When running this
# sample outside of Azure-enabled environments you may wish to swap in the
# `agent_framework.builtin` chat client or mock the writer executor. We keep the
# concrete import here so readers can see an end-to-end configuration.
from agent_framework import (
AgentExecutor,
AgentExecutorRequest,
@@ -14,30 +16,20 @@ from agent_framework import (
Executor,
FileCheckpointStorage,
RequestInfoEvent,
RequestInfoExecutor,
RequestInfoMessage,
RequestResponse,
Role,
Workflow,
WorkflowBuilder,
WorkflowCheckpoint,
WorkflowContext,
WorkflowOutputEvent,
WorkflowRunState,
WorkflowStatusEvent,
get_checkpoint_summary,
handler,
response_handler,
)
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential
# NOTE: the Azure client imports above are real dependencies. When running this
# sample outside of Azure-enabled environments you may wish to swap in the
# `agent_framework.builtin` chat client or mock the writer executor. We keep the
# concrete import here so readers can see an end-to-end configuration.
if TYPE_CHECKING:
from agent_framework import Workflow
from agent_framework._workflows._checkpoint import WorkflowCheckpoint
"""
Sample: Checkpoint + human-in-the-loop quickstart.
@@ -45,17 +37,14 @@ This getting-started sample keeps the moving pieces to a minimum:
1. A brief is turned into a consistent prompt for an AI copywriter.
2. The copywriter (an `AgentExecutor`) drafts release notes.
3. A reviewer gateway routes every draft through `RequestInfoExecutor` so a human
can approve or request tweaks.
3. A reviewer gateway sends a request for approval for every draft.
4. The workflow records checkpoints between each superstep so you can stop the
program, restart later, and optionally pre-supply human answers on resume.
Key concepts demonstrated
-------------------------
- Minimal executor pipeline with checkpoint persistence.
- Human-in-the-loop pause/resume by pairing `RequestInfoExecutor` with
checkpoint restoration.
- Supplying responses at restore time (`run_stream_from_checkpoint(..., responses=...)`).
- Human-in-the-loop pause/resume with checkpoint restoration.
Typical pause/resume flow
-------------------------
@@ -110,8 +99,8 @@ class BriefPreparer(Executor):
@dataclass
class HumanApprovalRequest(RequestInfoMessage):
"""Message sent to the human reviewer via RequestInfoExecutor."""
class HumanApprovalRequest:
"""Request sent to the human reviewer."""
# These fields are intentionally simple because they are serialised into
# checkpoints. Keeping them primitive types guarantees the new
@@ -124,52 +113,42 @@ class HumanApprovalRequest(RequestInfoMessage):
class ReviewGateway(Executor):
"""Routes agent drafts to humans and optionally back for revisions."""
def __init__(self, id: str, reviewer_id: str, writer_id: str, finalize_id: str) -> None:
def __init__(self, id: str, writer_id: str) -> None:
super().__init__(id=id)
self._reviewer_id = reviewer_id
self._writer_id = writer_id
self._finalize_id = finalize_id
@handler
async def on_agent_response(
self,
response: AgentExecutorResponse,
ctx: WorkflowContext[HumanApprovalRequest, str],
) -> None:
# Capture the agent output so we can surface it to the reviewer and
# persist iterations. The `RequestInfoExecutor` relies on this state to
# rehydrate when checkpoints are restored.
async def on_agent_response(self, response: AgentExecutorResponse, ctx: WorkflowContext) -> None:
# Capture the agent output so we can surface it to the reviewer and persist iterations.
draft = response.agent_run_response.text or ""
iteration = int((await ctx.get_executor_state() or {}).get("iteration", 0)) + 1
await ctx.set_executor_state({"iteration": iteration, "last_draft": draft})
# Emit a human approval request. Because this flows through
# RequestInfoExecutor it will pause the workflow until an answer is
# supplied either interactively or via pre-supplied responses.
await ctx.send_message(
# Emit a human approval request.
await ctx.request_info(
HumanApprovalRequest(
prompt="Review the draft. Reply 'approve' or provide edit instructions.",
draft=draft,
iteration=iteration,
),
target_id=self._reviewer_id,
HumanApprovalRequest,
str,
)
@handler
@response_handler
async def on_human_feedback(
self,
feedback: RequestResponse[HumanApprovalRequest, str],
original_request: HumanApprovalRequest,
feedback: str,
ctx: WorkflowContext[AgentExecutorRequest | str, str],
) -> None:
# The RequestResponse wrapper gives us both the human data and the
# original request message, even when resuming from checkpoints.
reply = (feedback.data or "").strip()
# The `original_request` is the request we sent earlier that is now being answered.
reply = feedback.strip()
state = await ctx.get_executor_state() or {}
draft = state.get("last_draft") or (feedback.original_request.draft if feedback.original_request else "")
draft = state.get("last_draft") or (original_request.draft or "")
if reply.lower() == "approve":
# When the human signs off we can short-circuit the workflow and
# send the approved draft to the final executor.
await ctx.send_message(draft, target_id=self._finalize_id)
# Workflow is completed when the human approves.
await ctx.yield_output(draft)
return
# Any other response loops us back to the writer with fresh guidance.
@@ -187,63 +166,34 @@ class ReviewGateway(Executor):
)
class FinaliseExecutor(Executor):
"""Publishes the approved text."""
@handler
async def publish(self, text: str, ctx: WorkflowContext[Any, str]) -> None:
# Store the output so diagnostics or a UI could fetch the final copy.
await ctx.set_executor_state({"published_text": text})
# Yield the final output so the workflow completes cleanly.
await ctx.yield_output(text)
def create_workflow(*, checkpoint_storage: FileCheckpointStorage | None = None) -> "Workflow":
def create_workflow(checkpoint_storage: FileCheckpointStorage) -> Workflow:
"""Assemble the workflow graph used by both the initial run and resume."""
# The Azure client is created once so our agent executor can issue calls to
# the hosted model. The agent id is stable across runs which keeps
# checkpoints deterministic.
# The Azure client is created once so our agent executor can issue calls to the hosted
# model. The agent id is stable across runs which keeps checkpoints deterministic.
chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())
writer = AgentExecutor(
chat_client.create_agent(
instructions="Write concise, warm release notes that sound human and helpful.",
),
id="writer",
)
# RequestInfoExecutor is the lynchpin for human-in-the-loop: every draft is
# routed through it so checkpoints can pause while waiting for responses.
review = RequestInfoExecutor(id="request_info")
finalise = FinaliseExecutor(id="finalise")
gateway = ReviewGateway(
id="review_gateway",
reviewer_id=review.id,
writer_id=writer.id,
finalize_id=finalise.id,
)
agent = chat_client.create_agent(instructions="Write concise, warm release notes that sound human and helpful.")
writer = AgentExecutor(agent, id="writer")
gateway = ReviewGateway(id="review_gateway", writer_id=writer.id)
prepare = BriefPreparer(id="prepare_brief", agent_id=writer.id)
# Wire the workflow DAG. Edges mirror the numbered steps described in the
# module docstring. Because `WorkflowBuilder` is declarative, reading these
# edges is often the quickest way to understand execution order.
builder = (
workflow_builder = (
WorkflowBuilder(max_iterations=6)
.set_start_executor(prepare)
.add_edge(prepare, writer)
.add_edge(writer, gateway)
.add_edge(gateway, review)
.add_edge(review, gateway) # human resumes loop
.add_edge(gateway, writer) # revisions
.add_edge(gateway, finalise)
.add_edge(gateway, writer) # revisions loop
.with_checkpointing(checkpoint_storage=checkpoint_storage)
)
# Opt-in to persistence when the caller provides storage. The workflow
# object itself is identical whether or not checkpointing is enabled.
if checkpoint_storage:
builder = builder.with_checkpointing(checkpoint_storage=checkpoint_storage)
return builder.build()
return workflow_builder.build()
def _render_checkpoint_summary(checkpoints: list["WorkflowCheckpoint"]) -> None:
def render_checkpoint_summary(checkpoints: list["WorkflowCheckpoint"]) -> None:
"""Pretty-print saved checkpoints with the new framework summaries."""
print("\nCheckpoint summary:")
@@ -251,166 +201,83 @@ def _render_checkpoint_summary(checkpoints: list["WorkflowCheckpoint"]) -> None:
# Compose a single line per checkpoint so the user can scan the output
# and pick the resume point that still has outstanding human work.
line = (
f"- {summary.checkpoint_id} | iter={summary.iteration_count} "
f"- {summary.checkpoint_id} | timestamp={summary.timestamp} | iter={summary.iteration_count} "
f"| targets={summary.targets} | states={summary.executor_ids}"
)
if summary.status:
line += f" | status={summary.status}"
if summary.draft_preview:
line += f" | draft_preview={summary.draft_preview}"
if summary.pending_requests:
line += f" | pending_request_id={summary.pending_requests[0].request_id}"
if summary.pending_request_info_events:
line += f" | pending_request_id={summary.pending_request_info_events[0].request_id}"
print(line)
def _print_events(events: list[Any]) -> tuple[str | None, list[tuple[str, HumanApprovalRequest]]]:
"""Echo workflow events to the console and collect outstanding requests."""
completed_output: str | None = None
requests: list[tuple[str, HumanApprovalRequest]] = []
for event in events:
print(f"Event: {event}")
if isinstance(event, WorkflowOutputEvent):
completed_output = event.data
if isinstance(event, RequestInfoEvent) and isinstance(event.data, HumanApprovalRequest):
# Capture pending human approvals so the caller can ask the user for
# input after the current batch of events is processed.
requests.append((event.request_id, event.data))
elif isinstance(event, WorkflowStatusEvent) and event.state in {
WorkflowRunState.IN_PROGRESS_PENDING_REQUESTS,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
}:
print(f"Workflow state: {event.state.name}")
return completed_output, requests
def _prompt_for_responses(requests: list[tuple[str, HumanApprovalRequest]]) -> dict[str, str] | None:
def prompt_for_responses(requests: dict[str, HumanApprovalRequest]) -> dict[str, str]:
"""Interactive CLI prompt for any live RequestInfo requests."""
if not requests:
return None
answers: dict[str, str] = {}
for request_id, request in requests:
# Keep the prompt conversational so testers can use the script without
# memorising the workflow APIs.
responses: dict[str, str] = {}
for request_id, request in requests.items():
print("\n=== Human approval needed ===")
print(f"request_id: {request_id}")
if request.iteration:
print(f"Iteration: {request.iteration}")
print(f"Iteration: {request.iteration}")
print(request.prompt)
print("Draft: \n---\n" + request.draft + "\n---")
answer = input("Type 'approve' or enter revision guidance (or 'exit' to quit): ").strip() # noqa: ASYNC250
if answer.lower() == "exit":
response = input("Type 'approve' or enter revision guidance (or 'exit' to quit): ").strip()
if response.lower() == "exit":
raise SystemExit("Stopped by user.")
answers[request_id] = answer
return answers
responses[request_id] = response
return responses
def _maybe_pre_supply_responses(cp: "WorkflowCheckpoint") -> dict[str, str] | None:
"""Offer to collect responses before resuming a checkpoint."""
pending = get_checkpoint_summary(cp).pending_requests
if not pending:
return None
print(
"This checkpoint still has pending human input. Provide the responses now so the resume step "
"applies them immediately and does not re-emit the original RequestInfo event."
)
choice = input("Pre-supply responses for this checkpoint? [y/N]: ").strip().lower() # noqa: ASYNC250
if choice not in {"y", "yes"}:
return None
answers: dict[str, str] = {}
for item in pending:
iteration = item.iteration or 0
print(f"\nPending draft (iteration {iteration} | request_id={item.request_id}):")
draft_text = (item.draft or "").strip()
if draft_text:
# The shortened preview in the summary may truncate text; here we
# show the full draft so the reviewer can make an informed choice.
print("Draft:\n---\n" + draft_text + "\n---")
else:
print("Draft: [not captured in checkpoint payload - refer to your notes/log]")
prompt_text = (item.prompt or "Review the draft").strip()
print(prompt_text)
answer = input("Response ('approve' or guidance, 'exit' to abort): ").strip() # noqa: ASYNC250
if answer.lower() == "exit":
raise SystemExit("Resume aborted by user.")
answers[item.request_id] = answer
return answers
async def _consume(stream: AsyncIterable[Any]) -> list[Any]:
"""Materialise an async event stream into a list."""
return [event async for event in stream]
async def run_interactive_session(workflow: "Workflow", initial_message: str) -> str | None:
async def run_interactive_session(
workflow: Workflow,
initial_message: str | None = None,
checkpoint_id: str | None = None,
) -> str:
"""Run the workflow until it either finishes or pauses for human input."""
pending_responses: dict[str, str] | None = None
requests: dict[str, HumanApprovalRequest] = {}
responses: dict[str, str] | None = None
completed_output: str | None = None
first = True
while completed_output is None:
if first:
# Kick off the workflow with the initial brief. The returned events
# include RequestInfo events when the agent produces a draft.
events = await _consume(workflow.run_stream(initial_message))
first = False
elif pending_responses:
# Feed any answers the user just typed back into the workflow.
events = await _consume(workflow.send_responses_streaming(pending_responses))
while True:
if responses:
event_stream = workflow.send_responses_streaming(responses)
requests.clear()
responses = None
else:
if initial_message:
print(f"\nStarting workflow with brief: {initial_message}\n")
event_stream = workflow.run_stream(initial_message)
elif checkpoint_id:
print("\nStarting workflow from checkpoint...\n")
event_stream = workflow.run_stream_from_checkpoint(checkpoint_id)
else:
raise ValueError("Either initial_message or checkpoint_id must be provided")
async for event in event_stream:
if isinstance(event, WorkflowStatusEvent):
print(event)
if isinstance(event, WorkflowOutputEvent):
completed_output = event.data
if isinstance(event, RequestInfoEvent):
if isinstance(event.data, HumanApprovalRequest):
requests[event.request_id] = event.data
else:
raise ValueError("Unexpected request data type")
if completed_output:
break
completed_output, requests = _print_events(events)
if completed_output is None:
pending_responses = _prompt_for_responses(requests)
if requests:
responses = prompt_for_responses(requests)
continue
raise RuntimeError("Workflow stopped without completing or requesting input")
return completed_output
async def resume_from_checkpoint(
workflow: "Workflow",
checkpoint_id: str,
storage: FileCheckpointStorage,
pre_supplied: dict[str, str] | None,
) -> None:
"""Resume a stored checkpoint and continue until completion or another pause."""
print(f"\nResuming from checkpoint: {checkpoint_id}")
events = await _consume(
workflow.run_stream_from_checkpoint(
checkpoint_id,
checkpoint_storage=storage,
responses=pre_supplied,
)
)
completed_output, requests = _print_events(events)
if pre_supplied and not requests and completed_output is None:
# When the checkpoint only needed the provided answers we let the user
# know the workflow is waiting for the next superstep (usually another
# agent response).
print("Pre-supplied responses applied automatically; workflow is now waiting for the next step.")
pending = _prompt_for_responses(requests)
while completed_output is None and pending:
events = await _consume(workflow.send_responses_streaming(pending))
completed_output, requests = _print_events(events)
if completed_output is None:
pending = _prompt_for_responses(requests)
else:
break
if completed_output:
print(f"Workflow completed with: {completed_output}")
async def main() -> None:
"""Entry point used by both the initial run and subsequent resumes."""
@@ -428,11 +295,8 @@ async def main() -> None:
)
print("Running workflow (human approval required)...")
completed = await run_interactive_session(workflow, initial_message=brief)
if completed:
print(f"Initial run completed with final copy: {completed}")
else:
print("Initial run paused for human input.")
result = await run_interactive_session(workflow, initial_message=brief)
print(f"Workflow completed with: {result}")
checkpoints = await storage.list_checkpoints()
if not checkpoints:
@@ -441,7 +305,7 @@ async def main() -> None:
# Show the user what is available before we prompt for the index. The
# summary helper keeps this output consistent with other tooling.
_render_checkpoint_summary(checkpoints)
render_checkpoint_summary(checkpoints)
sorted_cps = sorted(checkpoints, key=lambda c: c.timestamp)
print("\nAvailable checkpoints:")
@@ -472,14 +336,11 @@ async def main() -> None:
print("Selected checkpoint already reflects a completed workflow; nothing to resume.")
return
# If the user wants, capture their decisions now so the resume call can
# push them into the workflow and avoid re-prompting.
pre_responses = _maybe_pre_supply_responses(chosen)
resumed_workflow = create_workflow()
new_workflow = create_workflow(checkpoint_storage=storage)
# Resume with a fresh workflow instance. The checkpoint carries the
# persistent state while this object holds the runtime wiring.
await resume_from_checkpoint(resumed_workflow, chosen.checkpoint_id, storage, pre_responses)
result = await run_interactive_session(new_workflow, checkpoint_id=chosen.checkpoint_id)
print(f"Workflow completed with: {result}")
if __name__ == "__main__":
python/samples/getting_started/workflows/checkpoint/sub_workflow_checkpoint.py
+105 -66
View File
@@ -3,6 +3,7 @@
import asyncio
import contextlib
import json
import uuid
from dataclasses import dataclass, field, replace
from datetime import datetime, timedelta
from pathlib import Path
@@ -11,9 +12,8 @@ from agent_framework import (
Executor,
FileCheckpointStorage,
RequestInfoEvent,
RequestInfoExecutor,
RequestInfoMessage,
RequestResponse,
SubWorkflowRequestMessage,
SubWorkflowResponseMessage,
Workflow,
WorkflowBuilder,
WorkflowContext,
@@ -22,6 +22,7 @@ from agent_framework import (
WorkflowRunState,
WorkflowStatusEvent,
handler,
response_handler,
)
CHECKPOINT_DIR = Path(__file__).with_suffix("").parent / "tmp" / "sub_workflow_checkpoints"
@@ -30,7 +31,7 @@ CHECKPOINT_DIR = Path(__file__).with_suffix("").parent / "tmp" / "sub_workflow_c
Sample: Checkpointing for workflows that embed sub-workflows.
This sample shows how a parent workflow that wraps a sub-workflow can:
- run until the sub-workflow emits a human approval request via RequestInfoExecutor
- run until the sub-workflow emits a human approval request
- persist a checkpoint that captures the pending request (including complex payloads)
- resume later, supplying the human decision directly at restore time
@@ -78,9 +79,10 @@ class FinalDraft:
@dataclass
class ReviewRequest(RequestInfoMessage):
"""Human approval request surfaced via RequestInfoExecutor."""
class ReviewRequest:
"""Human approval request surfaced via `request_info`."""
id: str = str(uuid.uuid4())
topic: str = ""
iteration: int = 1
draft_excerpt: str = ""
@@ -88,6 +90,14 @@ class ReviewRequest(RequestInfoMessage):
reviewer_guidance: list[str] = field(default_factory=list) # type: ignore
@dataclass
class ReviewDecision:
"""The review decision to be sent to downstream executors along with the original request."""
decision: str
original_request: ReviewRequest
# ---------------------------------------------------------------------------
# Sub-workflow executors
# ---------------------------------------------------------------------------
@@ -122,7 +132,8 @@ class DraftReviewRouter(Executor):
super().__init__(id="draft_review")
@handler
async def request_review(self, draft: DraftPackage, ctx: WorkflowContext[ReviewRequest]) -> None:
async def request_review(self, draft: DraftPackage, ctx: WorkflowContext) -> None:
"""Request a review upon receiving a draft."""
excerpt = draft.content.splitlines()[0]
request = ReviewRequest(
topic=draft.topic,
@@ -134,15 +145,17 @@ class DraftReviewRouter(Executor):
"Confirm CTA is action-oriented",
],
)
await ctx.send_message(request, target_id="sub_review_requests")
await ctx.request_info(request, ReviewRequest, str)
@handler
@response_handler
async def forward_decision(
self,
decision: RequestResponse[ReviewRequest, str],
ctx: WorkflowContext[RequestResponse[ReviewRequest, str]],
original_request: ReviewRequest,
decision: str,
ctx: WorkflowContext[ReviewDecision],
) -> None:
await ctx.send_message(decision, target_id="draft_finaliser")
"""Route the decision to the next executor."""
await ctx.send_message(ReviewDecision(decision=decision, original_request=original_request))
class DraftFinaliser(Executor):
@@ -154,11 +167,11 @@ class DraftFinaliser(Executor):
@handler
async def on_review_decision(
self,
decision: RequestResponse[ReviewRequest, str],
review_decision: ReviewDecision,
ctx: WorkflowContext[DraftTask, FinalDraft],
) -> None:
reply = (decision.data or "").strip().lower()
original = decision.original_request
reply = review_decision.decision.strip().lower()
original = review_decision.original_request
topic = original.topic if original else "unknown topic"
iteration = original.iteration if original else 1
@@ -192,12 +205,11 @@ class LaunchCoordinator(Executor):
def __init__(self) -> None:
super().__init__(id="launch_coordinator")
self._final: FinalDraft | None = None
@handler
async def kick_off(self, topic: str, ctx: WorkflowContext[DraftTask]) -> None:
task = DraftTask(topic=topic, due=_utc_now() + timedelta(hours=2))
await ctx.send_message(task, target_id="launch_subworkflow")
await ctx.send_message(task)
@handler
async def collect_final(self, draft: FinalDraft, ctx: WorkflowContext[None, FinalDraft]) -> None:
@@ -209,8 +221,6 @@ class LaunchCoordinator(Executor):
normalised = replace(draft, approved_at=parsed)
approved_at = parsed
self._final = normalised
approved_display = approved_at.isoformat() if hasattr(approved_at, "isoformat") else str(approved_at)
print("\n>>> Parent workflow received approved draft:")
@@ -221,9 +231,50 @@ class LaunchCoordinator(Executor):
await ctx.yield_output(normalised)
@property
def final_result(self) -> FinalDraft | None:
return self._final
@handler
async def handler_sub_workflow_request(
self,
request: SubWorkflowRequestMessage,
ctx: WorkflowContext,
) -> None:
"""Handle requests from the sub-workflow.
Note that the message type must be SubWorkflowRequestMessage to intercept the request.
"""
if not isinstance(request.source_event.data, ReviewRequest):
raise TypeError(f"Expected 'ReviewRequest', got {type(request.source_event.data)}")
# Record the request to response matching
review_request = request.source_event.data
executor_state = await ctx.get_executor_state() or {}
executor_state[review_request.id] = request
await ctx.set_executor_state(executor_state)
# Send the request without modification
await ctx.request_info(review_request, ReviewRequest, str)
@response_handler
async def handle_request_response(
self,
original_request: ReviewRequest,
response: str,
ctx: WorkflowContext[SubWorkflowResponseMessage],
) -> None:
"""Process the response and send it back to the sub-workflow.
Note that the response must be sent back using SubWorkflowResponseMessage to route
the response back to the sub-workflow.
"""
executor_state = await ctx.get_executor_state() or {}
request_message = executor_state.pop(original_request.id, None)
# Save the executor state back to the context
await ctx.set_executor_state(executor_state)
if request_message is None:
raise ValueError("No matching pending request found for the resource response")
await ctx.send_message(request_message.create_response(response))
# ---------------------------------------------------------------------------
@@ -234,17 +285,13 @@ class LaunchCoordinator(Executor):
def build_sub_workflow() -> WorkflowExecutor:
writer = DraftWriter()
router = DraftReviewRouter()
request_info = RequestInfoExecutor(id="sub_review_requests")
finaliser = DraftFinaliser()
sub_workflow = (
WorkflowBuilder()
.set_start_executor(writer)
.add_edge(writer, router)
.add_edge(router, request_info)
.add_edge(request_info, router, condition=lambda msg: isinstance(msg, RequestResponse))
.add_edge(router, finaliser, condition=lambda msg: isinstance(msg, RequestResponse))
.add_edge(request_info, finaliser)
.add_edge(router, finaliser)
.add_edge(finaliser, writer) # permits revision loops
.build()
)
@@ -252,28 +299,19 @@ def build_sub_workflow() -> WorkflowExecutor:
return WorkflowExecutor(sub_workflow, id="launch_subworkflow")
def build_parent_workflow(storage: FileCheckpointStorage) -> tuple[LaunchCoordinator, Workflow]:
def build_parent_workflow(storage: FileCheckpointStorage) -> Workflow:
coordinator = LaunchCoordinator()
sub_executor = build_sub_workflow()
parent_request_info = RequestInfoExecutor(id="parent_review_gateway")
workflow = (
return (
WorkflowBuilder()
.set_start_executor(coordinator)
.add_edge(coordinator, sub_executor)
.add_edge(sub_executor, coordinator, condition=lambda msg: isinstance(msg, FinalDraft))
.add_edge(
sub_executor,
parent_request_info,
condition=lambda msg: isinstance(msg, RequestInfoMessage),
)
.add_edge(parent_request_info, sub_executor)
.add_edge(sub_executor, coordinator)
.with_checkpointing(storage)
.build()
)
return coordinator, workflow
async def main() -> None:
CHECKPOINT_DIR.mkdir(parents=True, exist_ok=True)
@@ -282,9 +320,10 @@ async def main() -> None:
storage = FileCheckpointStorage(CHECKPOINT_DIR)
_, workflow = build_parent_workflow(storage)
workflow = build_parent_workflow(storage)
print("\n=== Stage 1: run until sub-workflow requests human review ===")
request_id: str | None = None
async for event in workflow.run_stream("Contoso Gadget Launch"):
if isinstance(event, RequestInfoEvent) and request_id is None:
@@ -294,52 +333,52 @@ async def main() -> None:
break
if request_id is None:
print("Sub-workflow completed without requesting review.")
return
raise RuntimeError("Sub-workflow completed without requesting review.")
checkpoints = await storage.list_checkpoints(workflow.id)
if not checkpoints:
print("No checkpoints written.")
return
raise RuntimeError("No checkpoints found.")
# Print the checkpoint to show pending requests
# We didn't handle the request above so the request is still pending the last checkpoint
checkpoints.sort(key=lambda cp: cp.timestamp)
resume_checkpoint = checkpoints[-1]
print(f"Using checkpoint {resume_checkpoint.checkpoint_id} at iteration {resume_checkpoint.iteration_count}")
checkpoint_path = storage.storage_path / f"{resume_checkpoint.checkpoint_id}.json"
if checkpoint_path.exists():
snapshot = json.loads(checkpoint_path.read_text())
exec_states = snapshot.get("executor_states", {})
sub_pending = exec_states.get("sub_review_requests", {}).get("request_events", {})
parent_pending = exec_states.get("parent_review_gateway", {}).get("request_events", {})
print(f"Pending review requests (sub executor snapshot): {list(sub_pending.keys())}")
print(f"Pending review requests (parent executor snapshot): {list(parent_pending.keys())}")
checkpoint_content_dict = json.loads(checkpoint_path.read_text())
print(f"Pending review requests: {checkpoint_content_dict.get('pending_request_info_events', {})}")
print("\n=== Stage 2: resume from checkpoint ===")
print("\n=== Stage 2: resume from checkpoint and approve draft ===")
# Rebuild fresh instances to mimic a separate process resuming
coordinator2, workflow2 = build_parent_workflow(storage)
workflow2 = build_parent_workflow(storage)
approval_response = "approve"
final_event: WorkflowOutputEvent | None = None
request_info_event: RequestInfoEvent | None = None
async for event in workflow2.run_stream_from_checkpoint(
resume_checkpoint.checkpoint_id,
responses={request_id: approval_response},
):
if isinstance(event, RequestInfoEvent):
request_info_event = event
if request_info_event is None:
raise RuntimeError("No request_info_event captured.")
print("\n=== Stage 3: approve draft ==")
approval_response = "approve"
output_event: WorkflowOutputEvent | None = None
async for event in workflow2.send_responses_streaming({request_info_event.request_id: approval_response}):
if isinstance(event, WorkflowOutputEvent):
final_event = event
output_event = event
if final_event is None:
print("Workflow did not complete after resume.")
return
if output_event is None:
raise RuntimeError("Workflow did not complete after resume.")
final = final_event.data
output = output_event.data
print("\n=== Final Draft (from resumed run) ===")
print(final)
if coordinator2.final_result is None:
print("Coordinator did not capture final result via handler.")
else:
print("Coordinator stored final draft successfully.")
print(output)
""""
Sample Output:
python/samples/getting_started/workflows/composition/sub_workflow_parallel_requests.py
+277 -354
View File
@@ -1,87 +1,59 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import uuid
from dataclasses import dataclass
from typing import Any
from typing import Literal
from agent_framework import (
Executor,
RequestInfoExecutor,
RequestInfoMessage,
RequestResponse,
RequestInfoEvent,
SubWorkflowRequestMessage,
SubWorkflowResponseMessage,
Workflow,
WorkflowBuilder,
WorkflowContext,
WorkflowExecutor,
handler,
response_handler,
)
from typing_extensions import Never
"""
Sample: Sub-workflow with parallel request handling by specialized interceptors
This sample demonstrates how to handle multiple parallel requests from a sub-workflow to
different executors in the main workflow.
This sample demonstrates how different parent executors can handle different types of requests
from the same sub-workflow using regular @handler methods for RequestInfoMessage subclasses.
Prerequisite:
- Understanding of sub-workflows.
- Understanding of requests and responses.
Prerequisites:
- No external services required (external handling simulated via `RequestInfoExecutor`).
This pattern is useful when a sub-workflow needs to interact with multiple external systems
or services.
Key architectural principles:
1. Specialized interceptors: Each parent executor handles only specific request types
2. Type-based routing: ResourceCache handles ResourceRequest, PolicyEngine handles PolicyCheckRequest
3. Automatic type filtering: Each interceptor only receives requests with matching types
4. Fallback forwarding: Unhandled requests are forwarded to external services
This sample implements a resource request distribution system where:
1. A sub-workflow generates requests for computing resources and policy checks.
2. The main workflow has executors that handle resource allocation and policy checking.
3. Responses are routed back to the sub-workflow, which collects and processes them.
The example simulates a resource allocation system where:
- Sub-workflow makes mixed requests for resources (CPU, memory) and policy checks
- ResourceCache executor intercepts ResourceRequest messages, serves from cache or forwards
- PolicyEngine executor intercepts PolicyCheckRequest messages, applies rules or forwards
- Each interceptor uses typed @handler methods for automatic filtering
The sub-workflow sends two types of requests:
- ResourceRequest: Requests for computing resources (e.g., CPU, memory).
- PolicyRequest: Requests to check resource allocation policies.
Flow visualization:
Coordinator
|
| Mixed list[resource + policy requests]
v
[ Sub-workflow: WorkflowExecutor(ResourceRequester) ]
|
| Emits different RequestInfoMessage types:
| - ResourceRequest
| - PolicyCheckRequest
v
Parent workflow routes to specialized handlers:
| |
| ResourceCache.handle_resource_request | PolicyEngine.handle_policy_request
| (@handler ResourceRequest) | (@handler PolicyCheckRequest)
v v
Cache hit/miss decision Policy allow/deny decision
| |
| RequestResponse OR forward | RequestResponse OR forward
v v
Back to sub-workflow <----------> External RequestInfoExecutor
|
v
External responses route back
The main workflow contains:
- ResourceAllocator: Simulates a system that allocates computing resources.
- PolicyEngine: Simulates a policy engine that approves or denies resource requests.
"""
# 1. Define domain-specific request/response types
@dataclass
class ResourceRequest(RequestInfoMessage):
class ComputingResourceRequest:
"""Request for computing resources."""
resource_type: str = "cpu" # cpu, memory, disk, etc.
amount: int = 1
priority: str = "normal" # low, normal, high
@dataclass
class PolicyCheckRequest(RequestInfoMessage):
"""Request to check resource allocation policy."""
resource_type: str = ""
amount: int = 0
policy_type: str = "quota" # quota, compliance, security
request_type: Literal["resource", "policy"]
resource_type: Literal["cpu", "memory", "disk", "gpu"]
amount: int
priority: Literal["low", "normal", "high"] | None = None
policy_type: Literal["quota", "security"] | None = None
@dataclass
@@ -102,340 +74,291 @@ class PolicyResponse:
@dataclass
class RequestFinished:
pass
class ResourceRequest:
"""Request for computing resources."""
resource_type: Literal["cpu", "memory", "disk", "gpu"]
amount: int
priority: Literal["low", "normal", "high"]
id: str = str(uuid.uuid4())
# 2. Implement the sub-workflow executor - makes resource and policy requests
class ResourceRequester(Executor):
"""Simple executor that requests resources and checks policies."""
@dataclass
class PolicyRequest:
"""Request to check resource allocation policy."""
def __init__(self):
super().__init__(id="resource_requester")
self._request_count = 0
policy_type: Literal["quota", "security"]
resource_type: Literal["cpu", "memory", "disk", "gpu"]
amount: int
id: str = str(uuid.uuid4())
def build_resource_request_distribution_workflow() -> Workflow:
class RequestDistribution(Executor):
"""Distributes computing resource requests to appropriate executors."""
@handler
async def distribute_requests(
self,
requests: list[ComputingResourceRequest],
ctx: WorkflowContext[ResourceRequest | PolicyRequest | int],
) -> None:
for req in requests:
if req.request_type == "resource":
if req.priority is None:
raise ValueError("Priority must be set for resource requests")
await ctx.send_message(ResourceRequest(req.resource_type, req.amount, req.priority))
elif req.request_type == "policy":
if req.policy_type is None:
raise ValueError("Policy type must be set for policy requests")
await ctx.send_message(PolicyRequest(req.policy_type, req.resource_type, req.amount))
else:
raise ValueError(f"Unknown request type: {req.request_type}")
# Notify the collector about the number of requests sent
await ctx.send_message(len(requests))
class ResourceRequester(Executor):
"""Handles resource allocation requests."""
@handler
async def run(self, request: ResourceRequest, ctx: WorkflowContext) -> None:
await ctx.request_info(request, ResourceRequest, ResourceResponse)
@response_handler
async def handle_response(
self, original_request: ResourceRequest, response: ResourceResponse, ctx: WorkflowContext[ResourceResponse]
) -> None:
print(f"Resource allocated: {response.allocated} {response.resource_type} from {response.source}")
await ctx.send_message(response)
class PolicyChecker(Executor):
"""Handles policy check requests."""
@handler
async def run(self, request: PolicyRequest, ctx: WorkflowContext) -> None:
await ctx.request_info(request, PolicyRequest, PolicyResponse)
@response_handler
async def handle_response(
self, original_request: PolicyRequest, response: PolicyResponse, ctx: WorkflowContext[PolicyResponse]
) -> None:
print(f"Policy check result: {response.approved} - {response.reason}")
await ctx.send_message(response)
class ResultCollector(Executor):
"""Collects and processes all responses."""
def __init__(self, id: str) -> None:
super().__init__(id)
self._request_count = 0
self._responses: list[ResourceResponse | PolicyResponse] = []
@handler
async def set_request_count(self, count: int, ctx: WorkflowContext) -> None:
if count <= 0:
raise ValueError("Request count must be positive")
self._request_count = count
@handler
async def collect(self, response: ResourceResponse | PolicyResponse, ctx: WorkflowContext[Never, str]) -> None:
self._responses.append(response)
print(f"Collected {len(self._responses)}/{self._request_count} responses")
if len(self._responses) == self._request_count:
# All responses received, process them
await ctx.yield_output(f"All {self._request_count} requests processed.")
elif len(self._responses) > self._request_count:
raise ValueError("Received more responses than expected")
orchestrator = RequestDistribution("orchestrator")
resource_requester = ResourceRequester("resource_requester")
policy_checker = PolicyChecker("policy_checker")
result_collector = ResultCollector("result_collector")
return (
WorkflowBuilder()
.set_start_executor(orchestrator)
.add_edge(orchestrator, resource_requester)
.add_edge(orchestrator, policy_checker)
.add_edge(resource_requester, result_collector)
.add_edge(policy_checker, result_collector)
.add_edge(orchestrator, result_collector) # For request count
.build()
)
class ResourceAllocator(Executor):
"""Simulates a system that allocates computing resources."""
def __init__(self, id: str) -> None:
super().__init__(id)
self._cache: dict[str, int] = {"cpu": 10, "memory": 50, "disk": 100}
# Record pending requests to match responses
self._pending_requests: dict[str, RequestInfoEvent] = {}
async def _handle_resource_request(self, request: ResourceRequest) -> ResourceResponse | None:
"""Allocates resources based on request and available cache."""
available = self._cache.get(request.resource_type, 0)
if available >= request.amount:
self._cache[request.resource_type] -= request.amount
return ResourceResponse(request.resource_type, request.amount, "cache")
return None
@handler
async def request_resources(
self,
requests: list[dict[str, Any]],
ctx: WorkflowContext[ResourceRequest | PolicyCheckRequest],
async def handle_subworkflow_request(
self, request: SubWorkflowRequestMessage, ctx: WorkflowContext[SubWorkflowResponseMessage]
) -> None:
"""Process a list of resource requests."""
print(f"🏭 Sub-workflow processing {len(requests)} requests")
self._request_count += len(requests)
for req_data in requests:
req_type = req_data.get("request_type", "resource")
request: ResourceRequest | PolicyCheckRequest
if req_type == "resource":
print(f" 📦 Requesting resource: {req_data.get('type', 'cpu')} x{req_data.get('amount', 1)}")
request = ResourceRequest(
resource_type=req_data.get("type", "cpu"),
amount=req_data.get("amount", 1),
priority=req_data.get("priority", "normal"),
)
# Send to parent workflow for interception - not to target_id
await ctx.send_message(request)
elif req_type == "policy":
print(
f" 🛡️ Checking policy: {req_data.get('type', 'cpu')} x{req_data.get('amount', 1)} "
f"({req_data.get('policy_type', 'quota')})"
)
request = PolicyCheckRequest(
resource_type=req_data.get("type", "cpu"),
amount=req_data.get("amount", 1),
policy_type=req_data.get("policy_type", "quota"),
)
# Send to parent workflow for interception - not to target_id
await ctx.send_message(request)
@handler
async def handle_resource_response(
self,
response: RequestResponse[ResourceRequest, ResourceResponse],
ctx: WorkflowContext[Never, RequestFinished],
) -> None:
"""Handle resource allocation response."""
if response.data:
source_icon = "🏪" if response.data.source == "cache" else "🌐"
print(
f"📦 {source_icon} Sub-workflow received: {response.data.allocated} {response.data.resource_type} "
f"from {response.data.source}"
)
if self._collect_results():
# Yield completion result to the parent workflow.
await ctx.yield_output(RequestFinished())
@handler
async def handle_policy_response(
self,
response: RequestResponse[PolicyCheckRequest, PolicyResponse],
ctx: WorkflowContext[Never, RequestFinished],
) -> None:
"""Handle policy check response."""
if response.data:
status_icon = "✅" if response.data.approved else "❌"
print(
f"🛡️ {status_icon} Sub-workflow received policy response: "
f"{response.data.approved} - {response.data.reason}"
)
if self._collect_results():
# Yield completion result to the parent workflow.
await ctx.yield_output(RequestFinished())
def _collect_results(self) -> bool:
"""Collect and summarize results."""
self._request_count -= 1
print(f"📊 Sub-workflow completed request ({self._request_count} remaining)")
return self._request_count == 0
# 3. Implement the Resource Cache - Uses typed handler for ResourceRequest
class ResourceCache(Executor):
"""Interceptor that handles RESOURCE requests from cache using typed routing."""
# Use class attributes to avoid Pydantic assignment restrictions
cache: dict[str, int] = {"cpu": 10, "memory": 50, "disk": 100}
results: list[ResourceResponse] = []
def __init__(self):
super().__init__(id="resource_cache")
# Instance initialization only; state kept in class attributes as above
@handler
async def handle_resource_request(
self, request: ResourceRequest, ctx: WorkflowContext[RequestResponse[ResourceRequest, Any] | ResourceRequest]
) -> None:
"""Handle RESOURCE requests from sub-workflows and check cache first."""
resource_request = request
print(f"🏪 CACHE interceptor checking: {resource_request.amount} {resource_request.resource_type}")
available = self.cache.get(resource_request.resource_type, 0)
if available >= resource_request.amount:
# We can satisfy from cache
self.cache[resource_request.resource_type] -= resource_request.amount
response_data = ResourceResponse(
resource_type=resource_request.resource_type, allocated=resource_request.amount, source="cache"
)
print(f" ✅ Cache satisfied: {resource_request.amount} {resource_request.resource_type}")
self.results.append(response_data)
# Send response back to sub-workflow
response = RequestResponse(data=response_data, original_request=request, request_id=request.request_id)
await ctx.send_message(response, target_id=request.source_executor_id)
else:
# Cache miss - forward to external
print(f" ❌ Cache miss: need {resource_request.amount}, have {available} {resource_request.resource_type}")
await ctx.send_message(request)
@handler
async def collect_result(
self, response: RequestResponse[ResourceRequest, ResourceResponse], ctx: WorkflowContext
) -> None:
"""Collect results from external requests that were forwarded."""
if response.data and response.data.source != "cache": # Don't double-count our own results
self.results.append(response.data)
print(
f"🏪 🌐 Cache received external response: {response.data.allocated} {response.data.resource_type} "
f"from {response.data.source}"
)
# 4. Implement the Policy Engine - Uses typed handler for PolicyCheckRequest
class PolicyEngine(Executor):
"""Interceptor that handles POLICY requests using typed routing."""
# Use class attributes for simple sample state
quota: dict[str, int] = {
"cpu": 5, # Only allow up to 5 CPU units
"memory": 20, # Only allow up to 20 memory units
"disk": 1000, # Liberal disk policy
}
results: list[PolicyResponse] = []
def __init__(self):
super().__init__(id="policy_engine")
# Instance initialization only; state kept in class attributes as above
@handler
async def handle_policy_request(
self,
request: PolicyCheckRequest,
ctx: WorkflowContext[RequestResponse[PolicyCheckRequest, Any] | PolicyCheckRequest],
) -> None:
"""Handle POLICY requests from sub-workflows and apply rules."""
policy_request = request
print(
f"🛡️ POLICY interceptor checking: {policy_request.amount} {policy_request.resource_type}, policy={policy_request.policy_type}"
)
quota_limit = self.quota.get(policy_request.resource_type, 0)
if policy_request.policy_type == "quota":
if policy_request.amount <= quota_limit:
response_data = PolicyResponse(approved=True, reason=f"Within quota ({quota_limit})")
print(f" ✅ Policy approved: {policy_request.amount} <= {quota_limit}")
self.results.append(response_data)
# Send response back to sub-workflow
response = RequestResponse(data=response_data, original_request=request, request_id=request.request_id)
await ctx.send_message(response, target_id=request.source_executor_id)
return
# Exceeds quota - forward to external for review
print(f" ❌ Policy exceeds quota: {policy_request.amount} > {quota_limit}, forwarding to external")
await ctx.send_message(request)
"""Handles requests from sub-workflows."""
source_event: RequestInfoEvent = request.source_event
if not isinstance(source_event.data, ResourceRequest):
return
# Unknown policy type - forward to external
print(f" ❓ Unknown policy type: {policy_request.policy_type}, forwarding")
await ctx.send_message(request)
request_payload: ResourceRequest = source_event.data
response = await self._handle_resource_request(request_payload)
if response:
await ctx.send_message(request.create_response(response))
else:
# Request cannot be fulfilled via cache, forward the request to external
self._pending_requests[request_payload.id] = source_event
await ctx.request_info(request_payload, ResourceRequest, ResourceResponse)
@handler
async def collect_policy_result(
self, response: RequestResponse[PolicyCheckRequest, PolicyResponse], ctx: WorkflowContext
@response_handler
async def handle_external_response(
self,
original_request: ResourceRequest,
response: ResourceResponse,
ctx: WorkflowContext[SubWorkflowResponseMessage],
) -> None:
"""Collect policy results from external requests that were forwarded."""
if response.data:
self.results.append(response.data)
print(f"🛡️ 🌐 Policy received external response: {response.data.approved} - {response.data.reason}")
"""Handles responses from external systems and routes them to the sub-workflow."""
print(f"External resource allocated: {response.allocated} {response.resource_type} from {response.source}")
source_event = self._pending_requests.pop(original_request.id, None)
if source_event is None:
raise ValueError("No matching pending request found for the resource response")
await ctx.send_message(SubWorkflowResponseMessage(data=response, source_event=source_event))
class Coordinator(Executor):
def __init__(self):
super().__init__(id="coordinator")
class PolicyEngine(Executor):
"""Simulates a policy engine that approves or denies resource requests."""
def __init__(self, id: str) -> None:
super().__init__(id)
self._quota: dict[str, int] = {
"cpu": 5, # Only allow up to 5 CPU units
"memory": 20, # Only allow up to 20 memory units
"disk": 1000, # Liberal disk policy
}
# Record pending requests to match responses
self._pending_requests: dict[str, RequestInfoEvent] = {}
@handler
async def start(self, requests: list[dict[str, Any]], ctx: WorkflowContext[list[dict[str, Any]]]) -> None:
"""Start the resource allocation process."""
await ctx.send_message(requests, target_id="resource_workflow")
async def handle_subworkflow_request(
self, request: SubWorkflowRequestMessage, ctx: WorkflowContext[SubWorkflowResponseMessage]
) -> None:
"""Handles requests from sub-workflows."""
source_event: RequestInfoEvent = request.source_event
if not isinstance(source_event.data, PolicyRequest):
return
@handler
async def handle_completion(self, completion: RequestFinished, ctx: WorkflowContext) -> None:
"""Handle sub-workflow completion.
request_payload: PolicyRequest = source_event.data
# Simple policy logic for demonstration
if request_payload.policy_type == "quota":
allowed_amount = self._quota.get(request_payload.resource_type, 0)
if request_payload.amount <= allowed_amount:
response = PolicyResponse(True, "Within quota limits")
else:
response = PolicyResponse(False, "Exceeds quota limits")
await ctx.send_message(request.create_response(response))
else:
# For other policy types, forward to external system
self._pending_requests[request_payload.id] = source_event
await ctx.request_info(request_payload, PolicyRequest, PolicyResponse)
It comes from the sub-workflow yielded output.
"""
print("🎯 Main workflow received completion.")
@response_handler
async def handle_external_response(
self,
original_request: PolicyRequest,
response: PolicyResponse,
ctx: WorkflowContext[SubWorkflowResponseMessage],
) -> None:
"""Handles responses from external systems and routes them to the sub-workflow."""
print(f"External policy check result: {response.approved} - {response.reason}")
source_event = self._pending_requests.pop(original_request.id, None)
if source_event is None:
raise ValueError("No matching pending request found for the policy response")
await ctx.send_message(SubWorkflowResponseMessage(data=response, source_event=source_event))
async def main() -> None:
"""Demonstrate parallel request interception patterns."""
print("🚀 Starting Sub-Workflow Parallel Request Interception Demo...")
print("=" * 60)
# Create executors in the main workflow
sub_workflow = build_resource_request_distribution_workflow()
resource_allocator = ResourceAllocator("resource_allocator")
policy_engine = PolicyEngine("policy_engine")
# 5. Create the sub-workflow
resource_requester = ResourceRequester()
sub_request_info = RequestInfoExecutor(id="sub_request_info")
sub_workflow = (
WorkflowBuilder()
.set_start_executor(resource_requester)
.add_edge(resource_requester, sub_request_info)
.add_edge(sub_request_info, resource_requester)
.build()
# Create the WorkflowExecutor for the sub-workflow
# Setting allow_direct_output=True to let the sub-workflow output directly.
# This is because the sub-workflow is the both the entry point and the exit
# point of the main workflow.
sub_workflow_executor = WorkflowExecutor(
sub_workflow,
"sub_workflow_executor",
allow_direct_output=True,
)
# 6. Create parent workflow with PROPER interceptor pattern
cache = ResourceCache() # Intercepts ResourceRequest
policy = PolicyEngine() # Intercepts PolicyCheckRequest (different type!)
workflow_executor = WorkflowExecutor(sub_workflow, id="resource_workflow")
main_request_info = RequestInfoExecutor(id="main_request_info")
# Create a simple coordinator that starts the process
coordinator = Coordinator()
# TYPED ROUTING: Each executor handles specific typed RequestInfoMessage messages
# Build the main workflow
main_workflow = (
WorkflowBuilder()
.set_start_executor(coordinator)
.add_edge(coordinator, workflow_executor) # Start sub-workflow
.add_edge(workflow_executor, coordinator) # Sub-workflow completion back to coordinator
.add_edge(workflow_executor, cache) # WorkflowExecutor sends ResourceRequest to cache
.add_edge(workflow_executor, policy) # WorkflowExecutor sends PolicyCheckRequest to policy
.add_edge(cache, workflow_executor) # Cache sends RequestResponse back
.add_edge(policy, workflow_executor) # Policy sends RequestResponse back
.add_edge(cache, main_request_info) # Cache forwards ResourceRequest to external
.add_edge(policy, main_request_info) # Policy forwards PolicyCheckRequest to external
.add_edge(main_request_info, workflow_executor) # External responses back to sub-workflow
.set_start_executor(sub_workflow_executor)
.add_edge(sub_workflow_executor, resource_allocator)
.add_edge(resource_allocator, sub_workflow_executor)
.add_edge(sub_workflow_executor, policy_engine)
.add_edge(policy_engine, sub_workflow_executor)
.build()
)
# 7. Test with various requests (mixed resource and policy)
# Test requests
test_requests = [
{"request_type": "resource", "type": "cpu", "amount": 2, "priority": "normal"}, # Cache hit
{"request_type": "policy", "type": "cpu", "amount": 3, "policy_type": "quota"}, # Policy hit
{"request_type": "resource", "type": "memory", "amount": 15, "priority": "normal"}, # Cache hit
{"request_type": "policy", "type": "memory", "amount": 100, "policy_type": "quota"}, # Policy miss -> external
{"request_type": "resource", "type": "gpu", "amount": 1, "priority": "high"}, # Cache miss -> external
{"request_type": "policy", "type": "disk", "amount": 500, "policy_type": "quota"}, # Policy hit
{"request_type": "policy", "type": "cpu", "amount": 1, "policy_type": "security"}, # Unknown policy -> external
ComputingResourceRequest("resource", "cpu", 2, priority="normal"), # cache hit
ComputingResourceRequest("policy", "cpu", 3, policy_type="quota"), # policy hit
ComputingResourceRequest("resource", "memory", 15, priority="normal"), # cache hit
ComputingResourceRequest("policy", "memory", 100, policy_type="quota"), # policy miss -> external
ComputingResourceRequest("resource", "gpu", 1, priority="high"), # cache miss -> external
ComputingResourceRequest("policy", "disk", 500, policy_type="quota"), # policy hit
ComputingResourceRequest("policy", "cpu", 1, policy_type="security"), # unknown policy -> external
]
print(f"🧪 Testing with {len(test_requests)} mixed requests:")
for i, req in enumerate(test_requests, 1):
req_icon = "📦" if req["request_type"] == "resource" else "🛡️"
print(
f" {i}. {req_icon} {req['type']} x{req['amount']} "
f"({req.get('priority', req.get('policy_type', 'default'))})"
)
print("=" * 70)
# Run the workflow
print(f"🧪 Testing with {len(test_requests)} mixed requests.")
print("🚀 Starting main workflow...")
run_result = await main_workflow.run(test_requests)
# 8. Run the workflow
print("🎬 Running workflow...")
events = await main_workflow.run(test_requests)
# Handle request info events
request_info_events = run_result.get_request_info_events()
if request_info_events:
print(f"\n🔍 Handling {len(request_info_events)} request info events...\n")
# 9. Handle any external requests that couldn't be intercepted
request_events = events.get_request_info_events()
if request_events:
print(f"\n🌐 Handling {len(request_events)} external request(s)...")
external_responses: dict[str, Any] = {}
for event in request_events:
responses: dict[str, ResourceResponse | PolicyResponse] = {}
for event in request_info_events:
if isinstance(event.data, ResourceRequest):
# Handle ResourceRequest - create ResourceResponse
# Simulate external resource allocation
resource_response = ResourceResponse(
resource_type=event.data.resource_type, allocated=event.data.amount, source="external_provider"
)
external_responses[event.request_id] = resource_response
print(f" 🏭 External provider: {resource_response.allocated} {resource_response.resource_type}")
elif isinstance(event.data, PolicyCheckRequest):
# Handle PolicyCheckRequest - create PolicyResponse
policy_response = PolicyResponse(approved=True, reason="External policy service approved")
external_responses[event.request_id] = policy_response
print(f" 🔒 External policy: {'✅ APPROVED' if policy_response.approved else '❌ DENIED'}")
responses[event.request_id] = resource_response
elif isinstance(event.data, PolicyRequest):
# Simulate external policy check
response = PolicyResponse(True, "External system approved")
responses[event.request_id] = response
else:
print(f"Unknown request info event data type: {type(event.data)}")
await main_workflow.send_responses(external_responses)
run_result = await main_workflow.send_responses(responses)
outputs = run_result.get_outputs()
if outputs:
print("\nWorkflow completed with outputs:")
for output in outputs:
print(f"- {output}")
else:
print("\n🎯 All requests were intercepted internally!")
# 10. Show results and analysis
print("\n" + "=" * 70)
print("📊 RESULTS ANALYSIS")
print("=" * 70)
print(f"\n🏪 Cache Results ({len(cache.results)} handled):")
for result in cache.results:
print(f" ✅ {result.allocated} {result.resource_type} from {result.source}")
print(f"\n🛡️ Policy Results ({len(policy.results)} handled):")
for result in policy.results:
status_icon = "✅" if result.approved else "❌"
print(f" {status_icon} Approved: {result.approved} - {result.reason}")
print("\n💾 Final Cache State:")
for resource, amount in cache.cache.items():
print(f" 📦 {resource}: {amount} remaining")
print("\n📈 Summary:")
print(f" 🎯 Total requests: {len(test_requests)}")
print(f" 🏪 Resource requests handled: {len(cache.results)}")
print(f" 🛡️ Policy requests handled: {len(policy.results)}")
print(f" 🌐 External requests: {len(request_events) if request_events else 0}")
print("\n" + "=" * 70)
raise RuntimeError("Workflow did not produce an output.")
if __name__ == "__main__":
python/samples/getting_started/workflows/composition/sub_workflow_request_interception.py
+244 -226
View File
@@ -5,289 +5,307 @@ from dataclasses import dataclass
from agent_framework import (
Executor,
RequestInfoExecutor,
RequestInfoMessage,
RequestResponse,
SubWorkflowRequestMessage,
SubWorkflowResponseMessage,
Workflow,
WorkflowBuilder,
WorkflowContext,
WorkflowExecutor,
WorkflowOutputEvent,
handler,
response_handler,
)
from typing_extensions import Never
"""
Sample: Sub-Workflows with Request Interception
This sample demonstrates how to handle request from the sub-workflow in the main workflow.
This sample shows how to:
1. Create workflows that execute other workflows as sub-workflows
2. Intercept requests from sub-workflows using an executor with @handler for RequestInfoMessage subclasses
3. Conditionally handle or forward requests using RequestResponse messages
4. Handle external requests that are forwarded by the parent workflow
5. Proper request/response correlation for concurrent processing
Prerequisite:
- Understanding of sub-workflows.
- Understanding of requests and responses.
The example simulates an email validation system where:
- Sub-workflows validate multiple email addresses concurrently
- Parent workflows can intercept domain check requests for optimization
- Known domains (example.com, company.com) are approved locally
- Unknown domains (unknown.org) are forwarded to external services
- Request correlation ensures each email gets the correct domain check response
- External domain check requests are processed and responses routed back correctly
This pattern is useful when you want to reuse a workflow that makes requests to an external system,
but you want to intercept those requests in the main workflow and handle them without further propagation
to the external system.
Key concepts demonstrated:
- WorkflowExecutor: Wraps a workflow to make it behave as an executor
- RequestInfoMessage handler: @handler method to intercept sub-workflow requests
- Request correlation: Using request_id and source_executor_id to match responses with original requests
- Concurrent processing: Multiple emails processed simultaneously without interference
- External request routing: RequestInfoExecutor handles forwarded external requests
- Sub-workflow isolation: Sub-workflows work normally without knowing they're nested
- Sub-workflows complete by yielding outputs when validation is finished
Prerequisites:
- No external services required (external calls are simulated via `RequestInfoExecutor`).
Simple flow visualization:
Parent Orchestrator (handles DomainCheckRequest)
|
| EmailValidationRequest(email) x3 (concurrent)
v
[ Sub-workflow: WorkflowExecutor(EmailValidator) ]
|
| DomainCheckRequest(domain) with request_id and source_executor_id
v
Interception? yes -> handled locally with RequestResponse(data=True)
no -> forwarded to RequestInfoExecutor -> external service
|
v
Response routed back to sub-workflow using source_executor_id
This sample implements a smart email delivery system that validates email addresses before sending emails.
1. We will start by creating a workflow that validates email addresses in a sequential manner. The validation
consists of three steps: sanitization, format validation, and domain validation. The domain validation
step will involve checking if the email domain is valid by making a request to an external system.
2. Then we will create a main workflow that uses the email validation workflow as a sub-workflow. The main
workflow will intercept the domain validation requests from the sub-workflow and handle them internally
without propagating them to an external system.
3. Once the email address is validated, the main workflow will proceed to send the email if the address is valid,
or block the email if the address is invalid.
"""
# 1. Define domain-specific message types
@dataclass
class EmailValidationRequest:
"""Request to validate an email address."""
class SanitizedEmailResult:
"""Result of email sanitization and validation.
email: str
The properties get built up as the email address goes through
the validation steps in the workflow.
"""
@dataclass
class DomainCheckRequest(RequestInfoMessage):
"""Request to check if a domain is approved."""
domain: str = ""
@dataclass
class ValidationResult:
"""Result of email validation."""
email: str
original: str
sanitized: str
is_valid: bool
reason: str
# 2. Implement the sub-workflow executor (completely standard)
class EmailValidator(Executor):
"""Validates email addresses - doesn't know it's in a sub-workflow."""
def build_email_address_validation_workflow() -> Workflow:
"""Build an email address validation workflow.
def __init__(self) -> None:
"""Initialize the EmailValidator executor."""
super().__init__(id="email_validator")
# Use a dict to track multiple pending emails by request_id
self._pending_emails: dict[str, str] = {}
This workflow consists of three steps (each is represented by an executor):
1. Sanitize the email address, such as removing leading/trailing spaces.
2. Validate the email address format, such as checking for "@" and domain.
3. Extract the domain from the email address and request domain validation,
after which it completes with the final result.
"""
@handler
async def validate_request(
self,
request: EmailValidationRequest,
ctx: WorkflowContext[DomainCheckRequest | ValidationResult, ValidationResult],
) -> None:
"""Validate an email address."""
print(f"🔍 Sub-workflow validating email: {request.email}")
class EmailSanitizer(Executor):
"""Sanitize email address by trimming spaces."""
# Extract domain
domain = request.email.split("@")[1] if "@" in request.email else ""
@handler
async def handle(self, email_address: str, ctx: WorkflowContext[SanitizedEmailResult]) -> None:
"""Trim leading and trailing spaces from the email address.
if not domain:
print(f"❌ Invalid email format: {request.email}")
result = ValidationResult(email=request.email, is_valid=False, reason="Invalid email format")
await ctx.yield_output(result)
return
This executor doesn't produce any workflow output, but sends the sanitized
email address to the next executor in the workflow.
"""
sanitized = email_address.strip()
print(f"✂️ Sanitized email address: '{sanitized}'")
await ctx.send_message(SanitizedEmailResult(original=email_address, sanitized=sanitized, is_valid=False))
print(f"🌐 Sub-workflow requesting domain check for: {domain}")
# Request domain check
domain_check = DomainCheckRequest(domain=domain)
# Store the pending email with the request_id for correlation
self._pending_emails[domain_check.request_id] = request.email
await ctx.send_message(domain_check, target_id="email_request_info")
class EmailFormatValidator(Executor):
"""Validate email address format."""
@handler
async def handle_domain_response(
self,
response: RequestResponse[DomainCheckRequest, bool],
ctx: WorkflowContext[ValidationResult, ValidationResult],
) -> None:
"""Handle domain check response from RequestInfo with correlation."""
approved = bool(response.data)
domain = (
response.original_request.domain
if (hasattr(response, "original_request") and response.original_request)
else "unknown"
)
print(f"📬 Sub-workflow received domain response for '{domain}': {approved}")
@handler
async def handle(
self,
partial_result: SanitizedEmailResult,
ctx: WorkflowContext[SanitizedEmailResult, SanitizedEmailResult],
) -> None:
"""Validate the email address format.
# Find the corresponding email using the request_id
request_id = (
response.original_request.request_id
if (hasattr(response, "original_request") and response.original_request)
else None
)
if request_id and request_id in self._pending_emails:
email = self._pending_emails.pop(request_id) # Remove from pending
result = ValidationResult(
email=email,
is_valid=approved,
reason="Domain approved" if approved else "Domain not approved",
This executor can potentially produce a workflow output (False if the format is invalid).
When the format is valid, it sends the validated email address to the next executor in the workflow.
"""
if "@" not in partial_result.sanitized or "." not in partial_result.sanitized.split("@")[-1]:
print(f"❌ Invalid email format: '{partial_result.sanitized}'")
await ctx.yield_output(
SanitizedEmailResult(
original=partial_result.original, sanitized=partial_result.sanitized, is_valid=False
)
)
return
print(f"✅ Validated email format: '{partial_result.sanitized}'")
await ctx.send_message(
SanitizedEmailResult(
original=partial_result.original, sanitized=partial_result.sanitized, is_valid=False
)
)
print(f"✅ Sub-workflow completing validation for: {email}")
await ctx.yield_output(result)
class DomainValidator(Executor):
"""Validate email domain."""
def __init__(self, id: str):
super().__init__(id=id)
self._pending_domains: dict[str, SanitizedEmailResult] = {}
@handler
async def handle(self, partial_result: SanitizedEmailResult, ctx: WorkflowContext) -> None:
"""Extract the domain from the email address and request domain validation.
This executor doesn't produce any workflow output, but sends a domain validation request
to an external system to user for validation.
"""
domain = partial_result.sanitized.split("@")[-1]
print(f"🔍 Validating domain: '{domain}'")
self._pending_domains[domain] = partial_result
# Send a request to the external system via the request_info mechanism
await ctx.request_info(domain, str, bool)
@response_handler
async def handle_domain_validation_response(
self, original_request: str, is_valid: bool, ctx: WorkflowContext[Never, SanitizedEmailResult]
) -> None:
"""Handle the domain validation response.
This method receives the response from the external system and yields the final
validation result (True if both format and domain are valid, False otherwise).
"""
if original_request not in self._pending_domains:
raise ValueError(f"Received response for unknown domain: '{original_request}'")
partial_result = self._pending_domains.pop(original_request)
if is_valid:
print(f"✅ Domain '{original_request}' is valid.")
await ctx.yield_output(
SanitizedEmailResult(
original=partial_result.original, sanitized=partial_result.sanitized, is_valid=True
)
)
else:
print(f"❌ Domain '{original_request}' is invalid.")
await ctx.yield_output(
SanitizedEmailResult(
original=partial_result.original, sanitized=partial_result.sanitized, is_valid=False
)
)
# Build the workflow
sanitizer = EmailSanitizer(id="email_sanitizer")
format_validator = EmailFormatValidator(id="email_format_validator")
domain_validator = DomainValidator(id="domain_validator")
return (
WorkflowBuilder()
.set_start_executor(sanitizer)
.add_edge(sanitizer, format_validator)
.add_edge(format_validator, domain_validator)
.build()
)
@dataclass
class Email:
recipient: str
subject: str
body: str
# 3. Implement the parent workflow with request interception
class SmartEmailOrchestrator(Executor):
"""Parent orchestrator that can intercept domain checks."""
"""Orchestrates email address validation using a sub-workflow."""
approved_domains: set[str] = set()
def __init__(self, approved_domains: set[str] | None = None):
"""Initialize the SmartEmailOrchestrator with approved domains.
def __init__(self, id: str, approved_domains: set[str]):
"""Initialize the orchestrator with a set of approved domains.
Args:
approved_domains: Set of pre-approved domains, defaults to example.com, test.org, company.com
id: The executor ID.
approved_domains: A set of domains that are considered valid.
"""
super().__init__(id="email_orchestrator", approved_domains=approved_domains)
self._results: list[ValidationResult] = []
super().__init__(id=id)
self._approved_domains = approved_domains
# Keep track of previously approved and disapproved recipients
self._approved_recipients: set[str] = set()
self._disapproved_recipients: set[str] = set()
# Record pending emails waiting for validation results
self._pending_emails: dict[str, Email] = {}
@handler
async def start_validation(self, emails: list[str], ctx: WorkflowContext[EmailValidationRequest]) -> None:
"""Start validating a batch of emails."""
print(f"📧 Starting validation of {len(emails)} email addresses")
print("=" * 60)
for email in emails:
print(f"📤 Sending '{email}' to sub-workflow for validation")
request = EmailValidationRequest(email=email)
await ctx.send_message(request, target_id="email_validator_workflow")
async def run(self, email: Email, ctx: WorkflowContext[Email | str, bool]) -> None:
"""Start the email delivery process.
This handler receives an Email object. If the recipient has been previously approved,
it sends the email object to the next executor to handle delivery. If the recipient
has been previously disapproved, it yields False as the final result. Otherwise,
it sends the recipient email address to the sub-workflow for validation.
"""
recipient = email.recipient
if recipient in self._approved_recipients:
print(f"📧 Recipient '{recipient}' has been previously approved.")
await ctx.send_message(email)
return
if recipient in self._disapproved_recipients:
print(f"🚫 Blocking email to previously disapproved recipient: '{recipient}'")
await ctx.yield_output(False)
return
print(f"🔍 Validating new recipient email address: '{recipient}'")
self._pending_emails[recipient] = email
await ctx.send_message(recipient)
@handler
async def handle_domain_request(
self,
request: DomainCheckRequest,
ctx: WorkflowContext[RequestResponse[DomainCheckRequest, bool] | DomainCheckRequest],
async def handler_domain_validation_request(
self, request: SubWorkflowRequestMessage, ctx: WorkflowContext[SubWorkflowResponseMessage]
) -> None:
"""Handle requests from sub-workflows."""
print(f"🔍 Parent intercepting domain check for: {request.domain}")
"""Handle requests from the sub-workflow for domain validation.
if request.domain in self.approved_domains:
print(f"✅ Domain '{request.domain}' is pre-approved locally!")
# Send response back to sub-workflow
response = RequestResponse(data=True, original_request=request, request_id=request.request_id)
await ctx.send_message(response, target_id=request.source_executor_id)
else:
print(f"❓ Domain '{request.domain}' unknown, forwarding to external service...")
# Forward to external handler
await ctx.send_message(request)
Note that the message type must be SubWorkflowRequestMessage to intercept the request. And
the response must be sent back using SubWorkflowResponseMessage to route the response
back to the sub-workflow.
"""
if not isinstance(request.source_event.data, str):
raise TypeError(f"Expected domain string, got {type(request.source_event.data)}")
domain = request.source_event.data
is_valid = domain in self._approved_domains
print(f"🌐 External domain validation for '{domain}': {'valid' if is_valid else 'invalid'}")
await ctx.send_message(request.create_response(is_valid), target_id=request.executor_id)
@handler
async def collect_result(self, result: ValidationResult, ctx: WorkflowContext) -> None:
"""Collect validation results. It comes from the sub-workflow yielded output."""
status_icon = "✅" if result.is_valid else "❌"
print(f"📥 {status_icon} Validation result: {result.email} -> {result.reason}")
self._results.append(result)
async def handle_validation_result(self, result: SanitizedEmailResult, ctx: WorkflowContext[Email, bool]) -> None:
"""Handle the email address validation result.
@property
def results(self) -> list[ValidationResult]:
"""Get the collected validation results."""
return self._results
This handler receives the validation result from the sub-workflow.
If the email address is valid, it adds the recipient to the approved list
and sends the email object to the next executor to handle delivery.
If the email address is invalid, it adds the recipient to the disapproved list
and yields False as the final result.
"""
email = self._pending_emails.pop(result.original)
email.recipient = result.sanitized # Use the sanitized email address
if result.is_valid:
print(f"✅ Email address '{result.original}' is valid.")
self._approved_recipients.add(result.original)
await ctx.send_message(email)
else:
print(f"🚫 Email address '{result.original}' is invalid. Blocking email.")
self._disapproved_recipients.add(result.original)
await ctx.yield_output(False)
async def run_example() -> None:
"""Run the sub-workflow example."""
print("🚀 Setting up sub-workflow with request interception...")
print()
class EmailDelivery(Executor):
"""Simulates email delivery."""
# 4. Build the sub-workflow
email_validator = EmailValidator()
# Match the target_id used in EmailValidator ("email_request_info")
request_info = RequestInfoExecutor(id="email_request_info")
@handler
async def handle(self, email: Email, ctx: WorkflowContext[Never, bool]) -> None:
"""Simulate sending the email and yield True as the final result."""
print(f"📤 Sending email to '{email.recipient}' with subject '{email.subject}'")
await asyncio.sleep(1) # Simulate network delay
print(f"✅ Email sent to '{email.recipient}' successfully.")
await ctx.yield_output(True)
validation_workflow = (
WorkflowBuilder()
.set_start_executor(email_validator)
.add_edge(email_validator, request_info)
.add_edge(request_info, email_validator)
.build()
)
# 5. Build the parent workflow with interception
orchestrator = SmartEmailOrchestrator(approved_domains={"example.com", "company.com"})
workflow_executor = WorkflowExecutor(validation_workflow, id="email_validator_workflow")
# Add a RequestInfoExecutor to handle forwarded external requests
main_request_info = RequestInfoExecutor(id="main_request_info")
async def main() -> None:
# A list of approved domains
approved_domains = {"example.com", "company.com"}
main_workflow = (
# Create executors in the main workflow
orchestrator = SmartEmailOrchestrator(id="smart_email_orchestrator", approved_domains=approved_domains)
email_delivery = EmailDelivery(id="email_delivery")
# Create the sub-workflow for email address validation
validation_workflow = build_email_address_validation_workflow()
validation_workflow_executor = WorkflowExecutor(validation_workflow, id="email_validation_workflow")
# Build the main workflow
workflow = (
WorkflowBuilder()
.set_start_executor(orchestrator)
.add_edge(orchestrator, workflow_executor)
.add_edge(workflow_executor, orchestrator) # For ValidationResult collection and request interception
# Add edges for external request handling
.add_edge(orchestrator, main_request_info)
.add_edge(main_request_info, workflow_executor) # Route external responses to sub-workflow
.add_edge(orchestrator, validation_workflow_executor)
.add_edge(validation_workflow_executor, orchestrator)
.add_edge(orchestrator, email_delivery)
.build()
)
# 6. Prepare test inputs: known domain, unknown domain
test_emails = [
"user@example.com", # Should be intercepted and approved
"admin@company.com", # Should be intercepted and approved
"guest@unknown.org", # Should be forwarded externally
Email(recipient="user1@example.com", subject="Hello User1", body="This is a test email."),
Email(recipient=" user2@invalid", subject="Hello User2", body="This is a test email."),
Email(recipient=" user3@company.com ", subject="Hello User3", body="This is a test email."),
Email(recipient="user4@unknown.com", subject="Hello User4", body="This is a test email."),
# Re-send to an approved recipient
Email(recipient="user1@example.com", subject="Hello User1", body="This is a test email."),
# Re-send to a disapproved recipient
Email(recipient=" user2@invalid", subject="Hello User2", body="This is a test email."),
]
# 7. Run the workflow
result = await main_workflow.run(test_emails)
# 8. Handle any external requests
request_events = result.get_request_info_events()
if request_events:
print(f"\n🌐 Handling {len(request_events)} external request(s)...")
for event in request_events:
if event.data and hasattr(event.data, "domain"):
print(f"🔍 External domain check needed for: {event.data.domain}")
# Simulate external responses
external_responses: dict[str, bool] = {}
for event in request_events:
# Simulate external domain checking
if event.data and hasattr(event.data, "domain"):
domain = event.data.domain
# Let's say unknown.org is actually approved externally
approved = domain == "unknown.org"
print(f"🌐 External service response for '{domain}': {'APPROVED' if approved else 'REJECTED'}")
external_responses[event.request_id] = approved
# 9. Send external responses
await main_workflow.send_responses(external_responses)
else:
print("\n🎯 All requests were intercepted and handled locally!")
# 10. Display final summary
print("\n📊 Final Results Summary:")
print("=" * 60)
for result in orchestrator.results:
status = "✅ VALID" if result.is_valid else "❌ INVALID"
print(f"{status} {result.email}: {result.reason}")
print(f"\n🏁 Processed {len(orchestrator.results)} emails total")
# Execute the workflow
for email in test_emails:
print(f"\n🚀 Processing email to '{email.recipient}'")
async for event in workflow.run_stream(email):
if isinstance(event, WorkflowOutputEvent):
print(f"🎉 Final result for '{email.recipient}': {'Delivered' if event.data else 'Blocked'}")
if __name__ == "__main__":
asyncio.run(run_example())
asyncio.run(main())
python/samples/getting_started/workflows/human-in-the-loop/guessing_game_with_human_input.py
+43 -66
View File
@@ -10,16 +10,14 @@ from agent_framework import (
ChatMessage, # Chat message structure
Executor, # Base class for workflow executors
RequestInfoEvent, # Event emitted when human input is requested
RequestInfoExecutor, # Special executor that collects human input out of band
RequestInfoMessage, # Base class for request payloads sent to RequestInfoExecutor
RequestResponse, # Correlates a human response with the original request
Role, # Enum of chat roles (user, assistant, system)
WorkflowBuilder, # Fluent builder for assembling the graph
WorkflowContext, # Per run context and event bus
WorkflowOutputEvent, # Event emitted when workflow yields output
WorkflowRunState, # Enum of workflow run states
WorkflowStatusEvent, # Event emitted on run state changes
handler, # Decorator to expose an Executor method as a step
handler,
response_handler, # Decorator to expose an Executor method as a step
)
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential
@@ -29,12 +27,12 @@ from pydantic import BaseModel
Sample: Human in the loop guessing game
An agent guesses a number, then a human guides it with higher, lower, or
correct via RequestInfoExecutor. The loop continues until the human confirms
correct, at which point the workflow completes when idle with no pending work.
correct. The loop continues until the human confirms correct, at which point
the workflow completes when idle with no pending work.
Purpose:
Show how to integrate a human step in the middle of an LLM workflow using RequestInfoExecutor and correlated
RequestResponse objects.
Show how to integrate a human step in the middle of an LLM workflow by using
`request_info` and `send_responses_streaming`.
Demonstrate:
- Alternating turns between an AgentExecutor and a human, driven by events.
@@ -47,27 +45,20 @@ Prerequisites:
- Basic familiarity with WorkflowBuilder, executors, edges, events, and streaming runs.
"""
# What RequestInfoExecutor does:
# RequestInfoExecutor is a workflow-native bridge that pauses the graph at a request for information,
# emits a RequestInfoEvent with a typed payload, and then resumes the graph only after your application
# supplies a matching RequestResponse keyed by the emitted request_id. It does not gather input by itself.
# Your application is responsible for collecting the human reply from any UI or CLI and then calling
# send_responses_streaming with a dict mapping request_id to the human's answer. The executor exists to
# standardize pause-and-resume human gating, to carry typed request payloads, and to preserve correlation.
# How human-in-the-loop is achieved via `request_info` and `send_responses_streaming`:
# - An executor (TurnManager) calls `ctx.request_info` with a payload (HumanFeedbackRequest).
# - The workflow run pauses and emits a RequestInfoEvent 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 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.
# Request type sent to the RequestInfoExecutor for human feedback.
# Including the agent's last guess allows the UI or CLI to display context and helps
# the turn manager avoid extra state reads.
# Why subclass RequestInfoMessage:
# Subclassing RequestInfoMessage defines the exact schema of the request that the human will see.
# This gives you strong typing, forward-compatible validation, and clear correlation semantics.
# It also lets you attach contextual fields (such as the previous guess) so the UI can render a rich prompt
# without fetching extra state from elsewhere.
@dataclass
class HumanFeedbackRequest(RequestInfoMessage):
prompt: str = ""
guess: int | None = None
class HumanFeedbackRequest:
"""Request sent to the human for feedback on the agent's guess."""
prompt: str
class GuessOutput(BaseModel):
@@ -103,47 +94,45 @@ class TurnManager(Executor):
async def on_agent_response(
self,
result: AgentExecutorResponse,
ctx: WorkflowContext[HumanFeedbackRequest],
ctx: WorkflowContext,
) -> None:
"""Handle the agent's guess and request human guidance.
Steps:
1) Parse the agent's JSON into GuessOutput for robustness.
2) Send a HumanFeedbackRequest to the RequestInfoExecutor with a clear instruction:
- higher means the human's secret number is higher than the agent's guess.
- lower means the human's secret number is lower than the agent's guess.
- correct confirms the guess is exactly right.
- exit quits the demo.
2) Request info with a HumanFeedbackRequest as the payload.
"""
# Parse structured model output (defensive default if the agent did not reply).
text = result.agent_run_response.text or ""
last_guess = GuessOutput.model_validate_json(text).guess if text else None
# Parse structured model output
text = result.agent_run_response.text
last_guess = GuessOutput.model_validate_json(text).guess
# Craft a precise human prompt that defines higher and lower relative to the agent's guess.
prompt = (
f"The agent guessed: {last_guess if last_guess is not None else text}. "
f"The agent guessed: {last_guess}. "
"Type one of: higher (your number is higher than this guess), "
"lower (your number is lower than this guess), correct, or exit."
)
await ctx.send_message(HumanFeedbackRequest(prompt=prompt, guess=last_guess))
# Send a request with a prompt as the payload and expect a string reply.
await ctx.request_info(
request_data=HumanFeedbackRequest(prompt=prompt),
request_type=HumanFeedbackRequest,
response_type=str,
)
@handler
@response_handler
async def on_human_feedback(
self,
feedback: RequestResponse[HumanFeedbackRequest, str],
original_request: HumanFeedbackRequest,
feedback: str,
ctx: WorkflowContext[AgentExecutorRequest, str],
) -> None:
"""Continue the game or finish based on human feedback.
"""Continue the game or finish based on human feedback."""
print(f"Feedback for prompt '{original_request.prompt}' received: {feedback}")
The RequestResponse contains both the human's string reply and the correlated HumanFeedbackRequest,
which carries the prior guess for convenience.
"""
reply = (feedback.data or "").strip().lower()
# Prefer the correlated request's guess to avoid extra shared state reads.
last_guess = getattr(feedback.original_request, "guess", None)
reply = feedback.strip().lower()
if reply == "correct":
await ctx.yield_output(f"Guessed correctly: {last_guess}")
await ctx.yield_output("Guessed correctly!")
return
# Provide feedback to the agent to try again.
@@ -166,35 +155,24 @@ async def main() -> None:
'You MUST return ONLY a JSON object exactly matching this schema: {"guess": <integer 1..10>}. '
"No explanations or additional text."
),
# Structured output enforced via Pydantic model.
response_format=GuessOutput,
)
# Build a simple loop: TurnManager <-> AgentExecutor <-> RequestInfoExecutor.
# TurnManager coordinates, AgentExecutor runs the model, RequestInfoExecutor gathers human replies.
# Build a simple loop: TurnManager <-> AgentExecutor.
# TurnManager coordinates and gathers human replies while AgentExecutor runs the model.
turn_manager = TurnManager(id="turn_manager")
agent_exec = AgentExecutor(agent=agent, id="agent")
# Naming note:
# This variable is currently named hitl for historical reasons. The name can feel ambiguous or magical.
# Consider renaming to request_info_executor in your own code for clarity, since it directly represents
# the RequestInfoExecutor node that gathers human replies out of band.
hitl = RequestInfoExecutor(id="request_info")
top_builder = (
workflow = (
WorkflowBuilder()
.set_start_executor(turn_manager)
.add_edge(turn_manager, agent_exec) # Ask agent to make/adjust a guess
.add_edge(agent_exec, turn_manager) # Agent's response comes back to coordinator
.add_edge(turn_manager, hitl) # Ask human for guidance
.add_edge(hitl, turn_manager) # Feed human guidance back to coordinator
)
# Build the workflow (no checkpointing in this minimal sample).
workflow = top_builder.build()
).build()
# Human in the loop run: alternate between invoking the workflow and supplying collected responses.
pending_responses: dict[str, str] | None = None
completed = False
workflow_output: str | None = None
# User guidance printing:
@@ -206,7 +184,7 @@ async def main() -> None:
# flush=True,
# )
while not completed:
while workflow_output is None:
# First iteration uses run_stream("start").
# Subsequent iterations use send_responses_streaming with pending_responses from the console.
stream = (
@@ -228,7 +206,6 @@ async def main() -> None:
elif isinstance(event, WorkflowOutputEvent):
# Capture workflow output as they're yielded
workflow_output = str(event.data)
completed = True # In this sample, we finish after one output.
# Detect run state transitions for a better developer experience.
pending_status = any(
@@ -245,7 +222,7 @@ async def main() -> None:
print("State: IDLE_WITH_PENDING_REQUESTS (awaiting human input)")
# If we have any human requests, prompt the user and prepare responses.
if requests and not completed:
if requests:
responses: dict[str, str] = {}
for req_id, prompt in requests:
# Simple console prompt for the sample.
python/samples/getting_started/workflows/orchestration/magentic_checkpoint.py
+23 -18
View File
@@ -17,7 +17,8 @@ from agent_framework import (
WorkflowRunState,
WorkflowStatusEvent,
)
from agent_framework.openai import OpenAIChatClient
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity._credentials import AzureCliCredential
"""
Sample: Magentic Orchestration + Checkpointing
@@ -29,8 +30,8 @@ checkpoint, and later resume the workflow by feeding in the saved response.
Concepts highlighted here:
1. **Deterministic executor IDs** - the orchestrator and plan-review request executor
must keep stable IDs so the checkpoint state aligns when we rebuild the graph.
2. **Executor snapshotting** - checkpoints capture the `RequestInfoExecutor` state,
specifically the pending plan-review request map, at superstep boundaries.
2. **Executor snapshotting** - checkpoints capture the pending plan-review request
map, at superstep boundaries.
3. **Resume with responses** - `Workflow.run_stream_from_checkpoint` accepts a
`responses` mapping so we can inject the stored human reply during restoration.
@@ -58,14 +59,14 @@ def build_workflow(checkpoint_storage: FileCheckpointStorage):
name="ResearcherAgent",
description="Collects background facts and references for the project.",
instructions=("You are the research lead. Gather crisp bullet points the team should know."),
chat_client=OpenAIChatClient(),
chat_client=AzureOpenAIChatClient(credential=AzureCliCredential()),
)
writer = ChatAgent(
name="WriterAgent",
description="Synthesizes the final brief for stakeholders.",
instructions=("You convert the research notes into a structured brief with milestones and risks."),
chat_client=OpenAIChatClient(),
chat_client=AzureOpenAIChatClient(credential=AzureCliCredential()),
)
# The builder wires in the Magentic orchestrator, sets the plan review path, and
@@ -75,7 +76,7 @@ def build_workflow(checkpoint_storage: FileCheckpointStorage):
.participants(researcher=researcher, writer=writer)
.with_plan_review()
.with_standard_manager(
chat_client=OpenAIChatClient(),
chat_client=AzureOpenAIChatClient(credential=AzureCliCredential()),
max_round_count=10,
max_stall_count=3,
)
@@ -135,16 +136,23 @@ async def main() -> None:
print("\n=== Stage 2: resume from checkpoint and approve plan ===")
resumed_workflow = build_workflow(checkpoint_storage)
# Construct an approval reply to supply when the plan review request is re-emitted.
approval = MagenticPlanReviewReply(decision=MagenticPlanReviewDecision.APPROVE)
# Resume execution and supply the recorded approval in a single call.
# `run_stream_from_checkpoint` rebuilds executor state, applies the provided responses,
# and then continues the workflow. Because we only captured the initial plan review
# checkpoint, the resumed run should complete almost immediately.
# Resume execution and capture the re-emitted plan review request.
request_info_event: RequestInfoEvent | None = None
async for event in resumed_workflow.run_stream_from_checkpoint(resume_checkpoint.checkpoint_id):
if isinstance(event, RequestInfoEvent) and isinstance(event.data, MagenticPlanReviewRequest):
request_info_event = event
if request_info_event is None:
print("No plan review request re-emitted on resume; cannot approve.")
return
print(f"Resumed plan review request: {request_info_event.request_id}")
# Supply the approval and continue to run to completion.
final_event: WorkflowOutputEvent | None = None
async for event in resumed_workflow.run_stream_from_checkpoint(
resume_checkpoint.checkpoint_id,
responses={plan_review_request_id: approval},
):
async for event in resumed_workflow.send_responses_streaming({request_info_event.request_id: approval}):
if isinstance(event, WorkflowOutputEvent):
final_event = event
@@ -204,10 +212,7 @@ async def main() -> None:
final_event_post: WorkflowOutputEvent | None = None
post_emitted_events = False
post_plan_workflow = build_workflow(checkpoint_storage)
async for event in post_plan_workflow.run_stream_from_checkpoint(
post_plan_checkpoint.checkpoint_id,
responses={},
):
async for event in post_plan_workflow.run_stream_from_checkpoint(post_plan_checkpoint.checkpoint_id):
post_emitted_events = True
if isinstance(event, WorkflowOutputEvent):
final_event_post = event