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

Python: WorkflowBuilder registry (#2486)

Browse Source 
* Add workflow builder factory pattern

* Add internal edge groups to registered executors; next samples

* Update samples: Part 1

* register -> register_executor

* update hil samples

* Update other samples

* Update agent  samples

* Update doc string

* Add new sample

* Fix mypy

* Address comments

* Fix mypy
This commit is contained in:
Tao Chen
2025-12-04 21:26:10 -08:00
committed by GitHub
Unverified
parent 6809510413
commit f2ed5b55f6
33 changed files with 1609 additions and 696 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/samples/getting_started/workflows/state-management/shared_states_with_agents.py
+33 -22
View File
@@ -1,14 +1,15 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from uuid import uuid4
from agent_framework import (
AgentExecutorRequest,
AgentExecutorResponse,
ChatAgent,
ChatMessage,
Role,
WorkflowBuilder,
@@ -154,28 +155,35 @@ async def handle_spam(detection: DetectionResult, ctx: WorkflowContext[Never, st
raise RuntimeError("This executor should only handle spam messages.")
async def main() -> None:
# Create chat client and agents. response_format enforces structured JSON from each agent.
chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())
spam_detection_agent = chat_client.create_agent(
def create_spam_detection_agent() -> ChatAgent:
"""Creates a spam detection agent."""
return AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
instructions=(
"You are a spam detection assistant that identifies spam emails. "
"Always return JSON with fields is_spam (bool) and reason (string)."
),
response_format=DetectionResultAgent,
# response_format enforces structured JSON from each agent.
name="spam_detection_agent",
)
email_assistant_agent = chat_client.create_agent(
def create_email_assistant_agent() -> ChatAgent:
"""Creates an email assistant agent."""
return AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
instructions=(
"You are an email assistant that helps users draft responses to emails with professionalism. "
"Return JSON with a single field 'response' containing the drafted reply."
),
# response_format enforces structured JSON from each agent.
response_format=EmailResponse,
name="email_assistant_agent",
)
async def main() -> None:
"""Build and run the shared state with agents and conditional routing workflow."""
# Build the workflow graph with conditional edges.
# Flow:
# store_email -> spam_detection_agent -> to_detection_result -> branch:
@@ -183,25 +191,28 @@ async def main() -> None:
# True -> handle_spam
workflow = (
WorkflowBuilder()
.set_start_executor(store_email)
.add_edge(store_email, spam_detection_agent)
.add_edge(spam_detection_agent, to_detection_result)
.add_edge(to_detection_result, submit_to_email_assistant, condition=get_condition(False))
.add_edge(to_detection_result, handle_spam, condition=get_condition(True))
.add_edge(submit_to_email_assistant, email_assistant_agent)
.add_edge(email_assistant_agent, finalize_and_send)
.register_agent(create_spam_detection_agent, name="spam_detection_agent")
.register_agent(create_email_assistant_agent, name="email_assistant_agent")
.register_executor(lambda: store_email, name="store_email")
.register_executor(lambda: to_detection_result, name="to_detection_result")
.register_executor(lambda: submit_to_email_assistant, name="submit_to_email_assistant")
.register_executor(lambda: finalize_and_send, name="finalize_and_send")
.register_executor(lambda: handle_spam, name="handle_spam")
.set_start_executor("store_email")
.add_edge("store_email", "spam_detection_agent")
.add_edge("spam_detection_agent", "to_detection_result")
.add_edge("to_detection_result", "submit_to_email_assistant", condition=get_condition(False))
.add_edge("to_detection_result", "handle_spam", condition=get_condition(True))
.add_edge("submit_to_email_assistant", "email_assistant_agent")
.add_edge("email_assistant_agent", "finalize_and_send")
.build()
)
# Read an email from resources/spam.txt if available; otherwise use a default sample.
resources_path = os.path.join(
os.path.dirname(os.path.dirname(os.path.realpath(__file__))),
"resources",
"spam.txt",
)
if os.path.exists(resources_path):
with open(resources_path, encoding="utf-8") as f: # noqa: ASYNC230
email = f.read()
current_file = Path(__file__)
resources_path = current_file.parent.parent / "resources" / "spam.txt"
if resources_path.exists():
email = resources_path.read_text(encoding="utf-8")
else:
print("Unable to find resource file, using default text.")
email = "You are a WINNER! Click here for a free lottery offer!!!"