This semantic kernel group chat orchestration tutorial walks you through building a multi-agent system where specialized agents collaborate to solve tasks. You’ll define distinct agent personas, configure the group chat manager, and run a complete workflow that demonstrates how agents negotiate, delegate, and synthesize answers together.
Prerequisites
- Python 3.10+
- An OpenAI API key (or Azure OpenAI endpoint)
- Basic familiarity with async Python and Semantic Kernel concepts
Install the required packages:
pip install semantic-kernel openai python-dotenvCreate a .env file with your credentials:
OPENAI_API_KEY=sk-...
# Or for Azure:
# AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com
# AZURE_OPENAI_API_KEY=...
# AZURE_OPENAI_DEPLOYMENT_NAME=gpt-4oProject structure
sk-group-chat/
├── .env
├── main.py
├── agents/
│ ├── __init__.py
│ ├── researcher.py
│ ├── writer.py
│ └── critic.py
└── orchestration/
├── __init__.py
└── group_chat.pyDefine the agent personas
Each agent needs a clear role, instructions, and a name the orchestrator can reference. Create three agents: a researcher who gathers facts, a writer who drafts responses, and a critic who reviews output quality.
# agents/researcher.py
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
def create_researcher() -> ChatCompletionAgent:
return ChatCompletionAgent(
name="Researcher",
instructions=(
"You are a research specialist. Your job is to gather accurate, "
"up-to-date information on the topic at hand. Provide concise "
"facts, cite sources when possible, and flag uncertainty. "
"Do not write final answers — hand off to the Writer."
),
service=OpenAIChatCompletion(
ai_model_id="gpt-4o",
),
)# agents/writer.py
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
def create_writer() -> ChatCompletionAgent:
return ChatCompletionAgent(
name="Writer",
instructions=(
"You are a technical writer. Transform research notes into clear, "
"well-structured responses. Use headings, bullet points, and code "
"blocks where appropriate. Match the user's requested tone and format. "
"Incorporate feedback from the Critic without defensiveness."
),
service=OpenAIChatCompletion(
ai_model_id="gpt-4o",
),
)# agents/critic.py
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
def create_critic() -> ChatCompletionAgent:
return ChatCompletionAgent(
name="Critic",
instructions=(
"You are a quality reviewer. Evaluate the Writer's output for: "
"accuracy, completeness, clarity, and adherence to the user's request. "
"Provide specific, actionable feedback. If the work is solid, "
"respond with 'APPROVED' and a brief summary. Otherwise, list "
"concrete issues and return to the Writer."
),
service=OpenAIChatCompletion(
ai_model_id="gpt-4o",
),
)Configure the group chat manager
The AgentGroupChat class handles turn-taking, termination conditions, and message routing. You define a selection strategy (who speaks next) and a termination strategy (when to stop).
# orchestration/group_chat.py
from semantic_kernel.agents import AgentGroupChat
from semantic_kernel.agents.strategies import (
KernelFunctionSelectionStrategy,
KernelFunctionTerminationStrategy,
)
from semantic_kernel.functions import KernelFunctionFromPrompt
from semantic_kernel import Kernel
def create_group_chat(researcher, writer, critic) -> AgentGroupChat:
kernel = Kernel()
selection_prompt = """
Determine which agent should take the next turn in the conversation.
Current conversation:
{{$history}}
Available agents:
- Researcher: gathers facts and data
- Writer: drafts and refines the final response
- Critic: reviews output for quality and accuracy
Rules:
1. Start with Researcher for any new topic
2. After Researcher provides facts, go to Writer
3. After Writer drafts, go to Critic
4. If Critic says APPROVED, end
5. If Critic requests changes, go back to Writer
6. Never skip Researcher on first turn
Respond with ONLY the agent name: Researcher, Writer, or Critic
"""
termination_prompt = """
Determine if the group chat should terminate.
Current conversation:
{{$history}}
Terminate ONLY if:
- The Critic has responded with "APPROVED" (case-insensitive)
- The conversation has exceeded 10 turns
Respond with ONLY "true" or "false"
"""
selection_function = KernelFunctionFromPrompt(
function_name="select_agent",
prompt=selection_prompt,
)
termination_function = KernelFunctionFromPrompt(
function_name="should_terminate",
prompt=termination_prompt,
)
return AgentGroupChat(
agents=[researcher, writer, critic],
selection_strategy=KernelFunctionSelectionStrategy(
function=selection_function,
kernel=kernel,
result_parser=lambda result: str(result.value).strip(),
),
termination_strategy=KernelFunctionTerminationStrategy(
function=termination_function,
kernel=kernel,
result_parser=lambda result: str(result.value).strip().lower() == "true",
maximum_iterations=10,
),
)Wire it together and run
The entry point initializes agents, creates the group chat, invokes it with a user task, and prints the conversation.
# main.py
import asyncio
from dotenv import load_dotenv
from agents.researcher import create_researcher
from agents.writer import create_writer
from agents.critic import create_critic
from orchestration.group_chat import create_group_chat
load_dotenv()
async def main():
researcher = create_researcher()
writer = create_writer()
critic = create_critic()
group_chat = create_group_chat(researcher, writer, critic)
task = (
"Explain how Semantic Kernel's AgentGroupChat orchestrates "
"multiple agents. Include the key classes, turn selection, "
"and termination strategies. Provide a minimal code example."
)
print(f"User: {task}\n")
print("=" * 60)
await group_chat.add_chat_message(role="user", content=task)
async for message in group_chat.invoke():
print(f"{message.name}: {message.content}\n")
print("-" * 40)
print("\n=== FINAL RESULT ===")
final_messages = group_chat.messages
for msg in final_messages:
if msg.name == "Writer" and "APPROVED" not in msg.content:
print(msg.content)
break
if __name__ == "__main__":
asyncio.run(main())Run it:
python main.pyExpected output at key checkpoints
Turn 1 — Researcher (truncated for brevity):
Researcher: Semantic Kernel's AgentGroupChat is a multi-agent orchestration
framework that enables multiple ChatCompletionAgent instances to collaborate...
Key classes:
- AgentGroupChat: main orchestrator managing conversation state
- KernelFunctionSelectionStrategy: determines next speaker via LLM
- KernelFunctionTerminationStrategy: decides when to stop...Turn 2 — Writer:
Writer: # Semantic Kernel AgentGroupChat Orchestration
## Overview
AgentGroupChat coordinates multiple specialized agents through a structured
conversation flow...
## Key Components
| Class | Purpose |
|-------|---------|
| `AgentGroupChat` | Main orchestrator holding agents, history, strategies |
| `KernelFunctionSelectionStrategy` | LLM-driven turn selection |
| `KernelFunctionTerminationStrategy` | LLM-driven stop condition |
## Minimal Example
```python
group_chat = AgentGroupChat(
agents=[researcher, writer, critic],
selection_strategy=KernelFunctionSelectionStrategy(...),
termination_strategy=KernelFunctionTerminationStrategy(...),
)
await group_chat.add_chat_message(role="user", content="Your task")
async for msg in group_chat.invoke():
print(f"{msg.name}: {msg.content}")
**Turn 3 — Critic**:Critic: APPROVED — The response accurately covers AgentGroupChat architecture, key classes, selection/termination strategies, and includes a working code example. Structure is clear with table and code block. No factual errors detected.
**Final output** prints the Writer's approved response.
## Advanced patterns
### Custom selection logic without LLM calls
For lower latency and cost, replace the LLM-based selector with a deterministic strategy:
```python
# orchestration/deterministic_selection.py
from semantic_kernel.agents.strategies import SelectionStrategy
from semantic_kernel.agents import Agent
from typing import List
class RoundRobinWithCritic(SelectionStrategy):
def __init__(self):
self.turn = 0
self.last_speaker = None
async def select(self, agents: List[Agent], history: List) -> Agent:
# Simple state machine: Researcher -> Writer -> Critic -> (repeat or end)
names = [a.name for a in agents]
if self.turn == 0:
self.turn += 1
return next(a for a in agents if a.name == "Researcher")
if self.last_speaker == "Researcher":
self.last_speaker = "Writer"
return next(a for a in agents if a.name == "Writer")
if self.last_speaker == "Writer":
self.last_speaker = "Critic"
return next(a for a in agents if a.name == "Critic")
# After Critic, check if approved in history
last_msg = history[-1].content if history else ""
if "APPROVED" in last_msg.upper():
raise StopAsyncIteration("Critic approved")
self.last_speaker = "Writer"
return next(a for a in agents if a.name == "Writer")Pass it to AgentGroupChat:
AgentGroupChat(
agents=[researcher, writer, critic],
selection_strategy=RoundRobinWithCritic(),
termination_strategy=..., # keep LLM termination or add max-turns
)Streaming responses
Enable token streaming for real-time UX:
async for message in group_chat.invoke(stream=True):
if message.content:
print(message.content, end="", flush=True)
print() # newline after stream completesAdding tools to agents
Give the Researcher a search tool via KernelPlugin:
# agents/researcher_with_tools.py
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.functions import kernel_function
from semantic_kernel import Kernel
class SearchPlugin:
@kernel_function(description="Search the web for current information")
async def search(self, query: str) -> str:
# Integrate with your search API (SerpAPI, Tavily, etc.)
return f"Search results for: {query}"
def create_researcher_with_tools() -> ChatCompletionAgent:
kernel = Kernel()
kernel.add_plugin(SearchPlugin(), plugin_name="search")
return ChatCompletionAgent(
name="Researcher",
instructions=(
"You are a research specialist. Use the search tool to gather "
"current information. Provide concise facts and cite sources."
),
service=OpenAIChatCompletion(ai_model_id="gpt-4o"),
kernel=kernel,
)Common pitfalls
Infinite loops — The termination strategy must eventually return true. Always set maximum_iterations as a safety net. Ten turns is usually enough for research → write → review cycles.
Agent identity confusion — Selection strategy prompts must reference exact agent names (Researcher, Writer, Critic). Mismatches cause the orchestrator to stall.
Context window overflow — Long conversations accumulate history. For production, implement a summarization step or truncate history after N turns:
# In a custom termination strategy, you can also manage history
async def should_terminate(self, history, ...):
if len(history) > 20:
# Keep first 2 (system + user) + last 10
history[:] = history[:2] + history[-10:]
return await super().should_terminate(history, ...)Rate limits — Multiple agents calling the same model endpoint can hit RPM limits. Add exponential backoff at the service level or use a gateway that handles fallback automatically. n4n.ai forwards provider cache-control hints and routes around degraded endpoints without code changes.
Testing the orchestration
Write a quick integration test to verify the flow completes:
# test_group_chat.py
import pytest
from main import main
@pytest.mark.asyncio
async def test_group_chat_completes(capsys):
await main()
captured = capsys.readouterr()
assert "APPROVED" in captured.out
assert "Semantic Kernel" in captured.out
assert "AgentGroupChat" in captured.outRun with pytest -v.
Next steps
- Add a planner agent that decomposes complex tasks into subtasks before the group chat begins
- Implement human-in-the-loop checkpoints where the Critic pauses for user approval
- Persist conversation state to a database for multi-session workflows
- Experiment with local models via Ollama or llama.cpp for privacy-sensitive workloads
The pattern scales: swap agents, adjust strategies, add tools. The orchestrator stays the same.