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: Fix migration samples (#5015)

Browse Source 
* Fix migration samples

* Fix migration samples 2

* Fix formatting

* Comments
This commit is contained in:
Tao Chen
2026-03-31 23:32:30 -07:00
committed by GitHub
Unverified
parent d992febe9b
commit e43fc8ccec
14 changed files with 100 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/samples/semantic-kernel-migration/orchestrations/concurrent_basic.py
+5 -3
View File
@@ -55,7 +55,7 @@ def build_semantic_kernel_agents() -> list[ChatCompletionAgent]:
async def run_semantic_kernel_example(prompt: str) -> Sequence[ChatMessageContent]:
concurrent_orchestration = ConcurrentOrchestration(members=build_semantic_kernel_agents())
concurrent_orchestration = ConcurrentOrchestration(members=build_semantic_kernel_agents()) # type: ignore
runtime = InProcessRuntime()
runtime.start()
@@ -91,12 +91,14 @@ def _print_semantic_kernel_outputs(outputs: Sequence[ChatMessageContent]) -> Non
async def run_agent_framework_example(prompt: str) -> Sequence[list[Message]]:
client = OpenAIChatCompletionClient(credential=AzureCliCredential())
physics = Agent(client=client,
physics = Agent(
client=client,
instructions=("You are an expert in physics. Answer questions from a physics perspective."),
name="physics",
)
chemistry = Agent(client=client,
chemistry = Agent(
client=client,
instructions=("You are an expert in chemistry. Answer questions from a chemistry perspective."),
name="chemistry",
)
python/samples/semantic-kernel-migration/orchestrations/group_chat.py
+38 -28
View File
@@ -16,8 +16,8 @@ import sys
from collections.abc import Sequence
from typing import Any, cast
from agent_framework import Agent, Message
from agent_framework.foundry import FoundryChatClient
from agent_framework import Agent, AgentResponseUpdate, Message
from agent_framework.openai import OpenAIChatCompletionClient
from agent_framework.orchestrations import GroupChatBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -82,9 +82,6 @@ def build_semantic_kernel_agents() -> list[ChatCompletionAgent]:
class ChatCompletionGroupChatManager(GroupChatManager):
"""Group chat manager that delegates orchestration decisions to an Azure OpenAI deployment."""
service: ChatCompletionClientBase
topic: str
termination_prompt: str = (
"You are coordinating a conversation about '{{$topic}}'. "
"Decide if the discussion has produced a solid answer. "
@@ -104,8 +101,11 @@ class ChatCompletionGroupChatManager(GroupChatManager):
)
def __init__(self, *, topic: str, service: ChatCompletionClientBase, max_rounds: int | None = None) -> None:
super().__init__(topic=topic, service=service, max_rounds=max_rounds)
super().__init__(max_rounds=max_rounds)
self._round_robin_index = 0
self._topic = topic
self._service = service
async def _render_prompt(self, template: str, **kwargs: Any) -> str:
prompt_template = KernelPromptTemplate(prompt_template_config=PromptTemplateConfig(template=template))
@@ -117,7 +117,7 @@ class ChatCompletionGroupChatManager(GroupChatManager):
@override
async def should_terminate(self, chat_history: ChatHistory) -> BooleanResult:
rendered_prompt = await self._render_prompt(self.termination_prompt, topic=self.topic)
rendered_prompt = await self._render_prompt(self.termination_prompt, topic=self._topic)
chat_history.messages.insert(
0,
ChatMessageContent(role=AuthorRole.SYSTEM, content=rendered_prompt),
@@ -126,11 +126,11 @@ class ChatCompletionGroupChatManager(GroupChatManager):
ChatMessageContent(role=AuthorRole.USER, content="Decide if the discussion is complete."),
)
response = await self.service.get_chat_message_content(
response = await self._service.get_chat_message_content(
chat_history,
settings=PromptExecutionSettings(response_format=BooleanResult),
)
return BooleanResult.model_validate_json(response.content)
return BooleanResult.model_validate_json(response.content) # type: ignore
@override
async def select_next_agent(
@@ -140,7 +140,7 @@ class ChatCompletionGroupChatManager(GroupChatManager):
) -> StringResult:
rendered_prompt = await self._render_prompt(
self.selection_prompt,
topic=self.topic,
topic=self._topic,
participants=", ".join(participant_descriptions.keys()),
)
chat_history.messages.insert(
@@ -151,18 +151,18 @@ class ChatCompletionGroupChatManager(GroupChatManager):
ChatMessageContent(role=AuthorRole.USER, content="Pick the next participant to speak."),
)
response = await self.service.get_chat_message_content(
response = await self._service.get_chat_message_content(
chat_history,
settings=PromptExecutionSettings(response_format=StringResult),
)
result = StringResult.model_validate_json(response.content)
result = StringResult.model_validate_json(response.content) # type: ignore
if result.result not in participant_descriptions:
raise RuntimeError(f"Unknown participant selected: {result.result}")
return result
@override
async def filter_results(self, chat_history: ChatHistory) -> MessageResult:
rendered_prompt = await self._render_prompt(self.summary_prompt, topic=self.topic)
rendered_prompt = await self._render_prompt(self.summary_prompt, topic=self._topic)
chat_history.messages.insert(
0,
ChatMessageContent(role=AuthorRole.SYSTEM, content=rendered_prompt),
@@ -171,11 +171,11 @@ class ChatCompletionGroupChatManager(GroupChatManager):
ChatMessageContent(role=AuthorRole.USER, content="Summarize the plan."),
)
response = await self.service.get_chat_message_content(
response = await self._service.get_chat_message_content(
chat_history,
settings=PromptExecutionSettings(response_format=StringResult),
)
string_result = StringResult.model_validate_json(response.content)
string_result = StringResult.model_validate_json(response.content) # type: ignore
return MessageResult(
result=ChatMessageContent(role=AuthorRole.ASSISTANT, content=string_result.result),
reason=string_result.reason,
@@ -197,7 +197,7 @@ async def sk_agent_response_callback(message: ChatMessageContent | Sequence[Chat
async def run_semantic_kernel_example(task: str) -> str:
credential = AzureCliCredential()
orchestration = GroupChatOrchestration(
members=build_semantic_kernel_agents(),
members=build_semantic_kernel_agents(), # type: ignore
manager=ChatCompletionGroupChatManager(
topic=DISCUSSION_TOPIC,
service=AzureChatCompletion(credential=credential),
@@ -225,7 +225,7 @@ async def run_semantic_kernel_example(task: str) -> str:
async def run_agent_framework_example(task: str) -> str:
credential = AzureCliCredential()
client = OpenAIChatCompletionClient(credential=AzureCliCredential())
researcher = Agent(
name="Researcher",
@@ -234,32 +234,42 @@ async def run_agent_framework_example(task: str) -> str:
"Gather concise facts or considerations that help plan a community hackathon. "
"Keep your responses factual and scannable."
),
client=FoundryChatClient(credential=credential),
client=client,
)
planner = Agent(
name="Planner",
description="Turns the collected notes into a concrete action plan.",
instructions=("Propose a structured action plan that accounts for logistics, roles, and timeline."),
client=FoundryChatClient(credential=credential),
client=client,
)
workflow = GroupChatBuilder(
participants=[researcher, planner],
orchestrator_agent=Agent(client=FoundryChatClient(credential=credential)),
orchestrator_agent=Agent(client=client),
max_rounds=8,
intermediate_outputs=True,
).build()
final_response = ""
output_messages: list[Message] = []
last_message_id: str | None = None
async for event in workflow.run(task, stream=True):
if event.type == "output":
data = event.data
if isinstance(data, list) and len(data) > 0:
# Get the final message from the conversation
final_message = data[-1]
final_response = final_message.text or "" if isinstance(final_message, Message) else str(data)
if isinstance(event.data, AgentResponseUpdate):
if event.data.message_id != last_message_id:
last_message_id = event.data.message_id
print(f"{event.data.author_name}: {event.data.text}", end="")
else:
print(event.data.text, end="")
else:
final_response = str(data)
return final_response
output_messages.extend(cast(list[Message], event.data))
for message in output_messages:
print(f"[{message.author_name}] {message.text}")
if output_messages:
return output_messages[-1].text
return ""
async def main() -> None:
python/samples/semantic-kernel-migration/orchestrations/handoff.py
+6 -12
View File
@@ -11,15 +11,14 @@
"""Side-by-side handoff orchestrations for Semantic Kernel and Agent Framework."""
import asyncio
import sys
from collections.abc import AsyncIterable, Iterator, Sequence
from collections.abc import AsyncIterable, Callable, Iterator, Sequence
from agent_framework import (
Agent,
Message,
WorkflowEvent,
)
from agent_framework.foundry import FoundryChatClient
from agent_framework.openai import OpenAIChatCompletionClient
from agent_framework.orchestrations import HandoffAgentUserRequest, HandoffBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -36,11 +35,6 @@ from semantic_kernel.contents import (
)
from semantic_kernel.functions import kernel_function
if sys.version_info >= (3, 12):
pass # pragma: no cover
else:
pass # pragma: no cover
# Load environment variables from .env file
load_dotenv()
@@ -149,7 +143,7 @@ def _sk_streaming_callback(message: StreamingChatMessageContent, is_final: bool)
_sk_new_message = True
def _make_sk_human_responder(script: Iterator[str]) -> callable:
def _make_sk_human_responder(script: Iterator[str]) -> Callable[[], ChatMessageContent]:
def _responder() -> ChatMessageContent:
try:
user_text = next(script)
@@ -190,7 +184,7 @@ async def run_semantic_kernel_example(initial_task: str, scripted_responses: Seq
######################################################################
def _create_af_agents(client: FoundryChatClient):
def _create_af_agents(client: OpenAIChatCompletionClient):
triage = Agent(
client=client,
name="triage_agent",
@@ -245,7 +239,7 @@ def _extract_final_conversation(events: list[WorkflowEvent]) -> list[Message]:
async def run_agent_framework_example(initial_task: str, scripted_responses: Sequence[str]) -> str:
client = FoundryChatClient(credential=AzureCliCredential())
client = OpenAIChatCompletionClient(credential=AzureCliCredential())
triage, refund, status, returns = _create_af_agents(client)
workflow = (
@@ -281,7 +275,7 @@ async def run_agent_framework_example(initial_task: str, scripted_responses: Seq
return ""
# Render final transcript succinctly.
lines = []
lines: list[str] = []
for message in conversation:
text = message.text or ""
if not text.strip():
python/samples/semantic-kernel-migration/orchestrations/magentic.py
+28 -18
View File
@@ -13,8 +13,9 @@
import asyncio
from collections.abc import Sequence
from typing import cast
from agent_framework import Agent
from agent_framework import Agent, AgentResponseUpdate, Message
from agent_framework.openai import OpenAIChatClient
from agent_framework.orchestrations import MagenticBuilder
from dotenv import load_dotenv
@@ -46,21 +47,21 @@ PROMPT = (
######################################################################
async def build_semantic_kernel_agents() -> list:
async def build_semantic_kernel_agents() -> list[ChatCompletionAgent | OpenAIAssistantAgent]:
research_agent = ChatCompletionAgent(
name="ResearchAgent",
description="A helpful assistant with access to web search. Ask it to perform web searches.",
instructions=(
"You are a Researcher. You find information without additional computation or quantitative analysis."
),
service=OpenAIChatCompletion(ai_model_id="gpt-4o-search-preview"),
service=OpenAIChatCompletion(ai_model_id="gpt-4o-mini-search-preview"),
)
client = OpenAIAssistantAgent.create_client()
code_interpreter_tool, code_interpreter_tool_resources = OpenAIAssistantAgent.configure_code_interpreter_tool()
openai_settings = OpenAISettings()
model_id = openai_settings.chat_model_id if openai_settings.chat_model_id else "gpt-5"
definition = await client.beta.assistants.create(
definition = await client.beta.assistants.create( # pyright: ignore[reportDeprecated]
model=model_id,
name="CoderAgent",
description="A helpful assistant that writes and executes code to process and analyze data.",
@@ -94,7 +95,7 @@ def sk_agent_response_callback(
async def run_semantic_kernel_example(prompt: str) -> Sequence[ChatMessageContent]:
agents = await build_semantic_kernel_agents()
magentic_orchestration = MagenticOrchestration(
members=agents,
members=agents, # type: ignore
manager=StandardMagenticManager(chat_completion_service=OpenAIChatCompletion()),
agent_response_callback=sk_agent_response_callback,
)
@@ -137,7 +138,7 @@ async def run_agent_framework_example(prompt: str) -> str | None:
instructions=(
"You are a Researcher. You find information without additional computation or quantitative analysis."
),
client=OpenAIChatClient(model="gpt-4o-search-preview"),
client=OpenAIChatClient(model="gpt-4o-mini-search-preview"),
)
# Create code interpreter tool using static method
@@ -160,22 +161,31 @@ async def run_agent_framework_example(prompt: str) -> str | None:
client=OpenAIChatClient(),
)
workflow = MagenticBuilder(participants=[researcher, coder], manager_agent=manager_agent).build()
workflow = MagenticBuilder(
participants=[researcher, coder],
manager_agent=manager_agent, # type: ignore
intermediate_outputs=True,
).build()
final_text: str | None = None
output_messages: list[Message] = []
last_message_id: str | None = None
async for event in workflow.run(prompt, stream=True):
if event.type == "output":
data = event.data
if isinstance(data, str):
final_text = data
elif isinstance(data, list):
# Extract text from the last assistant message
for msg in reversed(data):
if hasattr(msg, "text") and msg.text:
final_text = msg.text
break
if isinstance(event.data, AgentResponseUpdate):
if event.data.message_id != last_message_id:
last_message_id = event.data.message_id
print(f"{event.data.author_name}: {event.data.text}", end="")
else:
print(event.data.text, end="")
else:
output_messages.extend(cast(list[Message], event.data))
for message in output_messages:
print(f"[{message.author_name}] {message.text}")
return final_text
if output_messages:
return output_messages[-1].text
return None
def _print_agent_framework_output(result: str | None) -> None: