mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Python: AutoGen migration samples (#1738)
* add autogen migration samples * fix typo * remove comment * fix typo * fix lab pyright * fix for HuggingFace change
This commit is contained in:
Eric Zhu
@@ -0,0 +1,185 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
"""AutoGen RoundRobinGroupChat vs Agent Framework GroupChatBuilder/SequentialBuilder.
|
||||
|
||||
Demonstrates sequential agent orchestration where agents take turns processing
|
||||
the task in a round-robin fashion.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
|
||||
|
||||
async def run_autogen() -> None:
|
||||
"""AutoGen's RoundRobinGroupChat for sequential agent orchestration."""
|
||||
from autogen_agentchat.agents import AssistantAgent
|
||||
from autogen_agentchat.conditions import TextMentionTermination
|
||||
from autogen_agentchat.teams import RoundRobinGroupChat
|
||||
from autogen_agentchat.ui import Console
|
||||
from autogen_ext.models.openai import OpenAIChatCompletionClient
|
||||
|
||||
client = OpenAIChatCompletionClient(model="gpt-4.1-mini")
|
||||
|
||||
# Create specialized agents
|
||||
researcher = AssistantAgent(
|
||||
name="researcher",
|
||||
model_client=client,
|
||||
system_message="You are a researcher. Provide facts and data about the topic.",
|
||||
model_client_stream=True,
|
||||
)
|
||||
|
||||
writer = AssistantAgent(
|
||||
name="writer",
|
||||
model_client=client,
|
||||
system_message="You are a writer. Turn research into engaging content.",
|
||||
model_client_stream=True,
|
||||
)
|
||||
|
||||
editor = AssistantAgent(
|
||||
name="editor",
|
||||
model_client=client,
|
||||
system_message="You are an editor. Review and finalize the content. End with APPROVED if satisfied.",
|
||||
model_client_stream=True,
|
||||
)
|
||||
|
||||
# Create round-robin team
|
||||
team = RoundRobinGroupChat(
|
||||
participants=[researcher, writer, editor],
|
||||
termination_condition=TextMentionTermination("APPROVED"),
|
||||
)
|
||||
|
||||
# Run the team and display the conversation.
|
||||
print("[AutoGen] Round-robin conversation:")
|
||||
await Console(team.run_stream(task="Create a brief summary about electric vehicles"))
|
||||
|
||||
|
||||
async def run_agent_framework() -> None:
|
||||
"""Agent Framework's SequentialBuilder for sequential agent orchestration."""
|
||||
from agent_framework import AgentRunUpdateEvent, SequentialBuilder
|
||||
from agent_framework.openai import OpenAIChatClient
|
||||
|
||||
client = OpenAIChatClient(model_id="gpt-4.1-mini")
|
||||
|
||||
# Create specialized agents
|
||||
researcher = client.create_agent(
|
||||
name="researcher",
|
||||
instructions="You are a researcher. Provide facts and data about the topic.",
|
||||
)
|
||||
|
||||
writer = client.create_agent(
|
||||
name="writer",
|
||||
instructions="You are a writer. Turn research into engaging content.",
|
||||
)
|
||||
|
||||
editor = client.create_agent(
|
||||
name="editor",
|
||||
instructions="You are an editor. Review and finalize the content.",
|
||||
)
|
||||
|
||||
# Create sequential workflow
|
||||
workflow = SequentialBuilder().participants([researcher, writer, editor]).build()
|
||||
|
||||
# Run the workflow
|
||||
print("[Agent Framework] Sequential conversation:")
|
||||
current_executor = None
|
||||
async for event in workflow.run_stream("Create a brief summary about electric vehicles"):
|
||||
if isinstance(event, AgentRunUpdateEvent):
|
||||
# Print executor name header when switching to a new agent
|
||||
if current_executor != event.executor_id:
|
||||
if current_executor is not None:
|
||||
print() # Newline after previous agent's message
|
||||
print(f"---------- {event.executor_id} ----------")
|
||||
current_executor = event.executor_id
|
||||
if event.data:
|
||||
print(event.data.text, end="", flush=True)
|
||||
print() # Final newline after conversation
|
||||
|
||||
|
||||
async def run_agent_framework_with_cycle() -> None:
|
||||
"""Agent Framework's WorkflowBuilder with cyclic edges and conditional exit."""
|
||||
from agent_framework import (
|
||||
AgentExecutorRequest,
|
||||
AgentExecutorResponse,
|
||||
AgentRunUpdateEvent,
|
||||
WorkflowBuilder,
|
||||
WorkflowContext,
|
||||
WorkflowOutputEvent,
|
||||
executor,
|
||||
)
|
||||
from agent_framework.openai import OpenAIChatClient
|
||||
|
||||
client = OpenAIChatClient(model_id="gpt-4.1-mini")
|
||||
|
||||
# Create specialized agents
|
||||
researcher = client.create_agent(
|
||||
name="researcher",
|
||||
instructions="You are a researcher. Provide facts and data about the topic.",
|
||||
)
|
||||
|
||||
writer = client.create_agent(
|
||||
name="writer",
|
||||
instructions="You are a writer. Turn research into engaging content.",
|
||||
)
|
||||
|
||||
editor = client.create_agent(
|
||||
name="editor",
|
||||
instructions="You are an editor. Review and finalize the content. End with APPROVED if satisfied.",
|
||||
)
|
||||
|
||||
# Create custom executor for checking approval
|
||||
@executor
|
||||
async def check_approval(
|
||||
response: AgentExecutorResponse, context: WorkflowContext[AgentExecutorRequest, str]
|
||||
) -> None:
|
||||
assert response.full_conversation is not None
|
||||
last_message = response.full_conversation[-1]
|
||||
if last_message and "APPROVED" in last_message.text:
|
||||
await context.yield_output("Content approved.")
|
||||
else:
|
||||
await context.send_message(AgentExecutorRequest(messages=response.full_conversation, should_respond=True))
|
||||
|
||||
workflow = (
|
||||
WorkflowBuilder()
|
||||
.add_edge(researcher, writer)
|
||||
.add_edge(writer, editor)
|
||||
.add_edge(
|
||||
editor,
|
||||
check_approval,
|
||||
)
|
||||
.add_edge(check_approval, researcher)
|
||||
.set_start_executor(researcher)
|
||||
.build()
|
||||
)
|
||||
|
||||
# Run the workflow
|
||||
print("[Agent Framework with Cycle] Cyclic conversation:")
|
||||
current_executor = None
|
||||
async for event in workflow.run_stream("Create a brief summary about electric vehicles"):
|
||||
if isinstance(event, WorkflowOutputEvent):
|
||||
print("\n---------- Workflow Output ----------")
|
||||
print(event.data)
|
||||
elif isinstance(event, AgentRunUpdateEvent):
|
||||
# Print executor name header when switching to a new agent
|
||||
if current_executor != event.executor_id:
|
||||
if current_executor is not None:
|
||||
print() # Newline after previous agent's message
|
||||
print(f"---------- {event.executor_id} ----------")
|
||||
current_executor = event.executor_id
|
||||
if event.data:
|
||||
print(event.data.text, end="", flush=True)
|
||||
print() # Final newline after conversation
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
print("=" * 60)
|
||||
print("Round-Robin / Sequential Orchestration Comparison")
|
||||
print("=" * 60)
|
||||
print("AutoGen: RoundRobinGroupChat")
|
||||
print("Agent Framework: SequentialBuilder + WorkflowBuilder with cycles\n")
|
||||
await run_autogen()
|
||||
print()
|
||||
await run_agent_framework()
|
||||
print()
|
||||
await run_agent_framework_with_cycle()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,125 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
"""AutoGen SelectorGroupChat vs Agent Framework GroupChatBuilder.
|
||||
|
||||
Demonstrates LLM-based speaker selection where an orchestrator decides
|
||||
which agent should speak next based on the conversation context.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
|
||||
|
||||
async def run_autogen() -> None:
|
||||
"""AutoGen's SelectorGroupChat with LLM-based speaker selection."""
|
||||
from autogen_agentchat.agents import AssistantAgent
|
||||
from autogen_agentchat.conditions import MaxMessageTermination
|
||||
from autogen_agentchat.teams import SelectorGroupChat
|
||||
from autogen_agentchat.ui import Console
|
||||
from autogen_ext.models.openai import OpenAIChatCompletionClient
|
||||
|
||||
client = OpenAIChatCompletionClient(model="gpt-4.1-mini")
|
||||
|
||||
# Create specialized agents
|
||||
python_expert = AssistantAgent(
|
||||
name="python_expert",
|
||||
model_client=client,
|
||||
system_message="You are a Python programming expert. Answer Python-related questions.",
|
||||
description="Expert in Python programming",
|
||||
model_client_stream=True,
|
||||
)
|
||||
|
||||
javascript_expert = AssistantAgent(
|
||||
name="javascript_expert",
|
||||
model_client=client,
|
||||
system_message="You are a JavaScript programming expert. Answer JavaScript-related questions.",
|
||||
description="Expert in JavaScript programming",
|
||||
model_client_stream=True,
|
||||
)
|
||||
|
||||
database_expert = AssistantAgent(
|
||||
name="database_expert",
|
||||
model_client=client,
|
||||
system_message="You are a database expert. Answer SQL and database-related questions.",
|
||||
description="Expert in databases and SQL",
|
||||
model_client_stream=True,
|
||||
)
|
||||
|
||||
# Create selector group chat - LLM selects appropriate expert
|
||||
team = SelectorGroupChat(
|
||||
participants=[python_expert, javascript_expert, database_expert],
|
||||
model_client=client,
|
||||
termination_condition=MaxMessageTermination(2),
|
||||
selector_prompt="Based on the conversation so far:\n{history}\n, "
|
||||
"select the most appropriate expert from {roles} to respond next.",
|
||||
)
|
||||
|
||||
# Run with a question that requires expert selection
|
||||
print("[AutoGen] Selector group chat conversation:")
|
||||
await Console(team.run_stream(task="How do I connect to a PostgreSQL database using Python?"))
|
||||
|
||||
|
||||
async def run_agent_framework() -> None:
|
||||
"""Agent Framework's GroupChatBuilder with LLM-based speaker selection."""
|
||||
from agent_framework import AgentRunUpdateEvent, GroupChatBuilder
|
||||
from agent_framework.openai import OpenAIChatClient
|
||||
|
||||
client = OpenAIChatClient(model_id="gpt-4.1-mini")
|
||||
|
||||
# Create specialized agents
|
||||
python_expert = client.create_agent(
|
||||
name="python_expert",
|
||||
instructions="You are a Python programming expert. Answer Python-related questions.",
|
||||
description="Expert in Python programming",
|
||||
)
|
||||
|
||||
javascript_expert = client.create_agent(
|
||||
name="javascript_expert",
|
||||
instructions="You are a JavaScript programming expert. Answer JavaScript-related questions.",
|
||||
description="Expert in JavaScript programming",
|
||||
)
|
||||
|
||||
database_expert = client.create_agent(
|
||||
name="database_expert",
|
||||
instructions="You are a database expert. Answer SQL and database-related questions.",
|
||||
description="Expert in databases and SQL",
|
||||
)
|
||||
|
||||
workflow = (
|
||||
GroupChatBuilder()
|
||||
.participants([python_expert, javascript_expert, database_expert])
|
||||
.set_prompt_based_manager(
|
||||
chat_client=client,
|
||||
instructions="Based on the conversation, select the most appropriate expert to respond next.",
|
||||
)
|
||||
.with_max_rounds(1)
|
||||
.build()
|
||||
)
|
||||
|
||||
# Run with a question that requires expert selection
|
||||
print("[Agent Framework] Group chat conversation:")
|
||||
current_executor = None
|
||||
async for event in workflow.run_stream("How do I connect to a PostgreSQL database using Python?"):
|
||||
if isinstance(event, AgentRunUpdateEvent):
|
||||
# Print executor name header when switching to a new agent
|
||||
if current_executor != event.executor_id:
|
||||
if current_executor is not None:
|
||||
print() # Newline after previous agent's message
|
||||
print(f"---------- {event.executor_id} ----------")
|
||||
current_executor = event.executor_id
|
||||
if event.data:
|
||||
print(event.data.text, end="", flush=True)
|
||||
print() # Final newline after conversation
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
print("=" * 60)
|
||||
print("Selector Group Chat Comparison")
|
||||
print("=" * 60)
|
||||
print("AutoGen: SelectorGroupChat")
|
||||
print("Agent Framework: GroupChatBuilder with standard_manager\n")
|
||||
await run_autogen()
|
||||
print()
|
||||
await run_agent_framework()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,238 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
"""AutoGen Swarm pattern vs Agent Framework HandoffBuilder.
|
||||
|
||||
Demonstrates agent handoff coordination where agents can transfer control
|
||||
to other specialized agents based on the task requirements.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
|
||||
|
||||
async def run_autogen() -> None:
|
||||
"""AutoGen's Swarm pattern with human-in-the-loop handoffs."""
|
||||
from autogen_agentchat.agents import AssistantAgent
|
||||
from autogen_agentchat.conditions import HandoffTermination, TextMentionTermination
|
||||
from autogen_agentchat.messages import HandoffMessage
|
||||
from autogen_agentchat.teams import Swarm
|
||||
from autogen_agentchat.ui import Console
|
||||
from autogen_ext.models.openai import OpenAIChatCompletionClient
|
||||
|
||||
client = OpenAIChatCompletionClient(model="gpt-4.1-mini")
|
||||
|
||||
# Create triage agent that routes to specialists
|
||||
triage_agent = AssistantAgent(
|
||||
name="triage",
|
||||
model_client=client,
|
||||
system_message=(
|
||||
"You are a triage agent. Analyze the user's request and hand off to the appropriate specialist.\n"
|
||||
"If you need information from the user, first send your message, then handoff to user.\n"
|
||||
"Use TERMINATE when the issue is fully resolved."
|
||||
),
|
||||
handoffs=["billing_agent", "technical_support", "user"],
|
||||
model_client_stream=True,
|
||||
)
|
||||
|
||||
# Create billing specialist
|
||||
billing_agent = AssistantAgent(
|
||||
name="billing_agent",
|
||||
model_client=client,
|
||||
system_message=(
|
||||
"You are a billing specialist. Help with payment and billing questions.\n"
|
||||
"If you need information from the user, first send your message, then handoff to user.\n"
|
||||
"When the issue is resolved, handoff to triage to finalize."
|
||||
),
|
||||
handoffs=["triage", "user"],
|
||||
model_client_stream=True,
|
||||
)
|
||||
|
||||
# Create technical support specialist
|
||||
tech_support = AssistantAgent(
|
||||
name="technical_support",
|
||||
model_client=client,
|
||||
system_message=(
|
||||
"You are technical support. Help with technical issues.\n"
|
||||
"If you need information from the user, first send your message, then handoff to user.\n"
|
||||
"When the issue is resolved, handoff to triage to finalize."
|
||||
),
|
||||
handoffs=["triage", "user"],
|
||||
model_client_stream=True,
|
||||
)
|
||||
|
||||
# Create swarm team with human-in-the-loop termination
|
||||
termination = HandoffTermination(target="user") | TextMentionTermination("TERMINATE")
|
||||
team = Swarm(
|
||||
participants=[triage_agent, billing_agent, tech_support],
|
||||
termination_condition=termination,
|
||||
)
|
||||
|
||||
# Scripted user responses for demonstration
|
||||
scripted_responses = [
|
||||
"I was charged twice for my subscription",
|
||||
"Yes, the charge of $49.99 appears twice on my credit card statement.",
|
||||
"Thank you for your help!",
|
||||
]
|
||||
response_index = 0
|
||||
|
||||
# Run with human-in-the-loop pattern
|
||||
print("[AutoGen] Swarm handoff conversation:")
|
||||
task_result = await Console(team.run_stream(task=scripted_responses[response_index]))
|
||||
last_message = task_result.messages[-1]
|
||||
response_index += 1
|
||||
|
||||
# Continue conversation when agents handoff to user
|
||||
while (
|
||||
isinstance(last_message, HandoffMessage)
|
||||
and last_message.target == "user"
|
||||
and response_index < len(scripted_responses)
|
||||
):
|
||||
user_message = scripted_responses[response_index]
|
||||
task_result = await Console(
|
||||
team.run_stream(task=HandoffMessage(source="user", target=last_message.source, content=user_message))
|
||||
)
|
||||
last_message = task_result.messages[-1]
|
||||
response_index += 1
|
||||
|
||||
|
||||
async def run_agent_framework() -> None:
|
||||
"""Agent Framework's HandoffBuilder for agent coordination."""
|
||||
from agent_framework import (
|
||||
AgentRunUpdateEvent,
|
||||
HandoffBuilder,
|
||||
HandoffUserInputRequest,
|
||||
RequestInfoEvent,
|
||||
WorkflowRunState,
|
||||
WorkflowStatusEvent,
|
||||
)
|
||||
from agent_framework.openai import OpenAIChatClient
|
||||
|
||||
client = OpenAIChatClient(model_id="gpt-4.1-mini")
|
||||
|
||||
# Create triage agent
|
||||
triage_agent = client.create_agent(
|
||||
name="triage",
|
||||
instructions=(
|
||||
"You are a triage agent. Analyze the user's request and route to the appropriate specialist:\n"
|
||||
"- For billing issues: call handoff_to_billing_agent\n"
|
||||
"- For technical issues: call handoff_to_technical_support"
|
||||
),
|
||||
description="Routes requests to appropriate specialists",
|
||||
)
|
||||
|
||||
# Create billing specialist
|
||||
billing_agent = client.create_agent(
|
||||
name="billing_agent",
|
||||
instructions="You are a billing specialist. Help with payment and billing questions. Provide clear assistance.",
|
||||
description="Handles billing and payment questions",
|
||||
)
|
||||
|
||||
# Create technical support specialist
|
||||
tech_support = client.create_agent(
|
||||
name="technical_support",
|
||||
instructions="You are technical support. Help with technical issues. Provide clear assistance.",
|
||||
description="Handles technical support questions",
|
||||
)
|
||||
|
||||
# Create handoff workflow - simpler configuration
|
||||
# After specialists respond, control returns to user (via triage as coordinator)
|
||||
workflow = (
|
||||
HandoffBuilder(
|
||||
name="support_handoff",
|
||||
participants=[triage_agent, billing_agent, tech_support],
|
||||
)
|
||||
.set_coordinator(triage_agent)
|
||||
.add_handoff(triage_agent, [billing_agent, tech_support])
|
||||
.with_termination_condition(lambda conv: sum(1 for msg in conv if msg.role.value == "user") > 3)
|
||||
.build()
|
||||
)
|
||||
|
||||
# Scripted user responses
|
||||
scripted_responses = [
|
||||
"I was charged twice for my subscription",
|
||||
"Yes, the charge of $49.99 appears twice on my credit card statement.",
|
||||
"Thank you for your help!",
|
||||
]
|
||||
|
||||
# Run with initial message
|
||||
print("[Agent Framework] Handoff conversation:")
|
||||
print("---------- user ----------")
|
||||
print(scripted_responses[0])
|
||||
|
||||
current_executor = None
|
||||
stream_line_open = False
|
||||
pending_requests: list[RequestInfoEvent] = []
|
||||
|
||||
async for event in workflow.run_stream(scripted_responses[0]):
|
||||
if isinstance(event, AgentRunUpdateEvent):
|
||||
# Print executor name header when switching to a new agent
|
||||
if current_executor != event.executor_id:
|
||||
if stream_line_open:
|
||||
print()
|
||||
stream_line_open = False
|
||||
print(f"---------- {event.executor_id} ----------")
|
||||
current_executor = event.executor_id
|
||||
stream_line_open = True
|
||||
if event.data:
|
||||
print(event.data.text, end="", flush=True)
|
||||
elif isinstance(event, RequestInfoEvent):
|
||||
if isinstance(event.data, HandoffUserInputRequest):
|
||||
pending_requests.append(event)
|
||||
elif isinstance(event, WorkflowStatusEvent):
|
||||
if event.state in {WorkflowRunState.IDLE_WITH_PENDING_REQUESTS} and stream_line_open:
|
||||
print()
|
||||
stream_line_open = False
|
||||
|
||||
# Process scripted responses
|
||||
response_index = 1
|
||||
while pending_requests and response_index < len(scripted_responses):
|
||||
user_response = scripted_responses[response_index]
|
||||
print("---------- user ----------")
|
||||
print(user_response)
|
||||
|
||||
responses = {req.request_id: user_response for req in pending_requests}
|
||||
pending_requests = []
|
||||
current_executor = None
|
||||
stream_line_open = False
|
||||
|
||||
async for event in workflow.send_responses_streaming(responses):
|
||||
if isinstance(event, AgentRunUpdateEvent):
|
||||
# Print executor name header when switching to a new agent
|
||||
if current_executor != event.executor_id:
|
||||
if stream_line_open:
|
||||
print()
|
||||
stream_line_open = False
|
||||
print(f"---------- {event.executor_id} ----------")
|
||||
current_executor = event.executor_id
|
||||
stream_line_open = True
|
||||
if event.data:
|
||||
print(event.data.text, end="", flush=True)
|
||||
elif isinstance(event, RequestInfoEvent):
|
||||
if isinstance(event.data, HandoffUserInputRequest):
|
||||
pending_requests.append(event)
|
||||
elif isinstance(event, WorkflowStatusEvent):
|
||||
if (
|
||||
event.state in {WorkflowRunState.IDLE_WITH_PENDING_REQUESTS, WorkflowRunState.IDLE}
|
||||
and stream_line_open
|
||||
):
|
||||
print()
|
||||
stream_line_open = False
|
||||
|
||||
response_index += 1
|
||||
|
||||
if stream_line_open:
|
||||
print()
|
||||
print() # Final newline after conversation
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
print("=" * 60)
|
||||
print("Swarm / Handoff Pattern Comparison")
|
||||
print("=" * 60)
|
||||
print("AutoGen: Swarm with handoffs")
|
||||
print("Agent Framework: HandoffBuilder\n")
|
||||
await run_autogen()
|
||||
print()
|
||||
await run_agent_framework()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,153 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
"""AutoGen MagenticOneGroupChat vs Agent Framework MagenticBuilder.
|
||||
|
||||
Demonstrates orchestrated multi-agent workflows with a central coordinator
|
||||
managing specialized agents for complex tasks.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
|
||||
|
||||
async def run_autogen() -> None:
|
||||
"""AutoGen's MagenticOneGroupChat for orchestrated collaboration."""
|
||||
from autogen_agentchat.agents import AssistantAgent
|
||||
from autogen_agentchat.teams import MagenticOneGroupChat
|
||||
from autogen_agentchat.ui import Console
|
||||
from autogen_ext.models.openai import OpenAIChatCompletionClient
|
||||
|
||||
client = OpenAIChatCompletionClient(model="gpt-4.1-mini")
|
||||
|
||||
# Create specialized agents
|
||||
researcher = AssistantAgent(
|
||||
name="researcher",
|
||||
model_client=client,
|
||||
system_message="You are a research analyst. Gather and analyze information.",
|
||||
description="Research analyst for data gathering",
|
||||
model_client_stream=True,
|
||||
)
|
||||
|
||||
coder = AssistantAgent(
|
||||
name="coder",
|
||||
model_client=client,
|
||||
system_message="You are a programmer. Write code based on requirements.",
|
||||
description="Software developer for implementation",
|
||||
model_client_stream=True,
|
||||
)
|
||||
|
||||
reviewer = AssistantAgent(
|
||||
name="reviewer",
|
||||
model_client=client,
|
||||
system_message="You are a code reviewer. Review code for quality and correctness.",
|
||||
description="Code reviewer for quality assurance",
|
||||
model_client_stream=True,
|
||||
)
|
||||
|
||||
# Create MagenticOne team with coordinator
|
||||
team = MagenticOneGroupChat(
|
||||
participants=[researcher, coder, reviewer],
|
||||
model_client=client, # Coordinator uses this client
|
||||
max_turns=20,
|
||||
max_stalls=3,
|
||||
)
|
||||
|
||||
# Run complex task and display the conversation
|
||||
print("[AutoGen] Magentic One conversation:")
|
||||
await Console(team.run_stream(task="Research Python async patterns and write a simple example"))
|
||||
|
||||
|
||||
async def run_agent_framework() -> None:
|
||||
"""Agent Framework's MagenticBuilder for orchestrated collaboration."""
|
||||
from agent_framework import (
|
||||
MagenticAgentDeltaEvent,
|
||||
MagenticAgentMessageEvent,
|
||||
MagenticBuilder,
|
||||
MagenticFinalResultEvent,
|
||||
MagenticOrchestratorMessageEvent,
|
||||
)
|
||||
from agent_framework.openai import OpenAIChatClient
|
||||
|
||||
client = OpenAIChatClient(model_id="gpt-4.1-mini")
|
||||
|
||||
# Create specialized agents
|
||||
researcher = client.create_agent(
|
||||
name="researcher",
|
||||
instructions="You are a research analyst. Gather and analyze information.",
|
||||
description="Research analyst for data gathering",
|
||||
)
|
||||
|
||||
coder = client.create_agent(
|
||||
name="coder",
|
||||
instructions="You are a programmer. Write code based on requirements.",
|
||||
description="Software developer for implementation",
|
||||
)
|
||||
|
||||
reviewer = client.create_agent(
|
||||
name="reviewer",
|
||||
instructions="You are a code reviewer. Review code for quality and correctness.",
|
||||
description="Code reviewer for quality assurance",
|
||||
)
|
||||
|
||||
# Create Magentic workflow
|
||||
workflow = (
|
||||
MagenticBuilder()
|
||||
.participants(researcher=researcher, coder=coder, reviewer=reviewer)
|
||||
.with_standard_manager(
|
||||
chat_client=client,
|
||||
max_round_count=20,
|
||||
max_stall_count=3,
|
||||
max_reset_count=1,
|
||||
)
|
||||
.build()
|
||||
)
|
||||
|
||||
# Run complex task
|
||||
print("[Agent Framework] Magentic conversation:")
|
||||
last_stream_agent_id: str | None = None
|
||||
stream_line_open: bool = False
|
||||
|
||||
async for event in workflow.run_stream("Research Python async patterns and write a simple example"):
|
||||
if isinstance(event, MagenticOrchestratorMessageEvent):
|
||||
if stream_line_open:
|
||||
print()
|
||||
stream_line_open = False
|
||||
print(f"---------- Orchestrator:{event.kind} ----------")
|
||||
print(getattr(event.message, "text", ""))
|
||||
elif isinstance(event, MagenticAgentDeltaEvent):
|
||||
if last_stream_agent_id != event.agent_id or not stream_line_open:
|
||||
if stream_line_open:
|
||||
print()
|
||||
print(f"---------- {event.agent_id} ----------")
|
||||
last_stream_agent_id = event.agent_id
|
||||
stream_line_open = True
|
||||
if event.text:
|
||||
print(event.text, end="", flush=True)
|
||||
elif isinstance(event, MagenticAgentMessageEvent):
|
||||
if stream_line_open:
|
||||
print()
|
||||
stream_line_open = False
|
||||
elif isinstance(event, MagenticFinalResultEvent):
|
||||
if stream_line_open:
|
||||
print()
|
||||
stream_line_open = False
|
||||
print("---------- Final Result ----------")
|
||||
if event.message is not None:
|
||||
print(event.message.text)
|
||||
|
||||
if stream_line_open:
|
||||
print()
|
||||
print() # Final newline after conversation
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
print("=" * 60)
|
||||
print("Magentic One Orchestration Comparison")
|
||||
print("=" * 60)
|
||||
print("AutoGen: MagenticOneGroupChat")
|
||||
print("Agent Framework: MagenticBuilder\n")
|
||||
await run_autogen()
|
||||
print()
|
||||
await run_agent_framework()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user