Semantic Kernel agent tutorial: multi-agent conversations

This semantic kernel multi-agent conversation tutorial walks you through building a working multi-agent system using Semantic Kernel’s Agent Framework. You’ll create specialized agents, wire them into a group chat, and implement termination strategies that prevent infinite loops. The code runs against any OpenAI-compatible endpoint.

Prerequisites

• Python 3.10+
• An OpenAI-compatible API endpoint and key (works with OpenAI, Azure OpenAI, or any compatible gateway)
• Basic familiarity with async Python and Semantic Kernel concepts

Install the required packages:

pip install semantic-kernel[azure] openai python-dotenv

The [azure] extra pulls in the Azure OpenAI client; if you’re using plain OpenAI or another compatible endpoint, pip install semantic-kernel openai python-dotenv works fine.

Create a .env file in your project root:

# .env
AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com
AZURE_OPENAI_API_KEY=your-key
AZURE_OPENAI_DEPLOYMENT=gpt-4o
AZURE_OPENAI_API_VERSION=2024-08-01-preview

# Or for standard OpenAI / compatible gateway:
# OPENAI_API_KEY=sk-...
# OPENAI_MODEL=gpt-4o

Project structure

multi_agent_tutorial/
├── .env
├── main.py
├── agents/
│   ├── __init__.py
│   ├── researcher.py
│   ├── writer.py
│   └── critic.py
└── orchestration/
    ├── __init__.py
    └── group_chat.py

Define the agent roles

Each agent gets a distinct system prompt and, optionally, a function plugin. We’ll build three agents: a researcher that fetches facts, a writer that drafts content, and a critic that reviews output.

Researcher agent

# agents/researcher.py
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
from semantic_kernel.contents import ChatHistory
from semantic_kernel.functions import kernel_function, KernelArguments
from typing import Annotated

class ResearchPlugin:
    """Simple in-memory 'search' for demo purposes. Replace with real search in production."""
    
    @kernel_function(
        name="lookup_fact",
        description="Retrieve a known fact about a topic. Returns a short statement."
    )
    def lookup_fact(self, topic: Annotated[str, "The topic to look up"]) -> str:
        facts = {
            "semantic kernel": "Semantic Kernel is an open-source SDK from Microsoft that lets you combine LLMs with conventional code via planners, plugins, and agents.",
            "multi-agent": "Multi-agent systems coordinate multiple specialized agents to solve complex tasks through conversation.",
            "group chat": "Group chat in Semantic Kernel orchestrates turn-taking among agents using a selection strategy and termination condition.",
        }
        return facts.get(topic.lower(), f"No cached fact for '{topic}'. In production, call a search API here.")

def create_researcher_agent(service: AzureChatCompletion) -> ChatCompletionAgent:
    return ChatCompletionAgent(
        service=service,
        name="Researcher",
        instructions=(
            "You are a research specialist. When asked about a topic, use the lookup_fact tool "
            "to retrieve accurate information. Always cite the tool result. Do not hallucinate."
        ),
        plugins=[ResearchPlugin()],
    )

Writer agent

# agents/writer.py
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion

def create_writer_agent(service: AzureChatCompletion) -> ChatCompletionAgent:
    return ChatCompletionAgent(
        service=service,
        name="Writer",
        instructions=(
            "You are a technical writer. Take research findings and produce clear, concise "
            "explanations suitable for software engineers. Use code blocks for examples. "
            "Keep responses under 300 words unless asked for more detail."
        ),
    )

Critic agent

# agents/critic.py
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion

def create_critic_agent(service: AzureChatCompletion) -> ChatCompletionAgent:
    return ChatCompletionAgent(
        service=service,
        name="Critic",
        instructions=(
            "You are a critical reviewer. Evaluate the writer's output for accuracy, clarity, "
            "and completeness. Identify missing details, unclear phrasing, or hallucinations. "
            "Respond with either 'APPROVED' if the content is ready, or specific feedback "
            "starting with 'REVISION NEEDED:' followed by actionable items."
        ),
    )

Wire the group chat

Semantic Kernel’s AgentGroupChat handles turn-taking. You provide 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.connectors.ai.open_ai import AzureChatCompletion
from semantic_kernel.contents import ChatMessageContent, AuthorRole
from semantic_kernel.functions import kernel_function, KernelArguments
from agents.researcher import create_researcher_agent
from agents.writer import create_writer_agent
from agents.critic import create_critic_agent
from typing import List

SELECTION_PROMPT = """
Determine which agent should speak next in the conversation.
Current conversation:
{{$history}}

Agents:
- Researcher: gathers facts using tools
- Writer: drafts content based on research
- Critic: reviews and approves or requests revisions

Rules:
1. Start with Researcher for any new topic.
2. After Researcher provides facts, Writer drafts.
3. After Writer drafts, Critic reviews.
4. If Critic says REVISION NEEDED, Writer revises.
5. If Critic says APPROVED, stop.

Return only the agent name: Researcher, Writer, or Critic.
"""

TERMINATION_PROMPT = """
Determine if the conversation should end.
Last message: {{$last_message}}

End the conversation ONLY if the Critic's last message contains "APPROVED".
Return "true" to terminate, "false" to continue.
"""

class MultiAgentOrchestrator:
    def __init__(self, service: AzureChatCompletion):
        self.service = service
        self.researcher = create_researcher_agent(service)
        self.writer = create_writer_agent(service)
        self.critic = create_critic_agent(service)
        self.chat = self._build_chat()

    def _build_chat(self) -> AgentGroupChat:
        selection_strategy = KernelFunctionSelectionStrategy(
            function=self._create_selection_function(),
            kernel=self.service.kernel,
            result_parser=lambda result: str(result.value).strip(),
            agent_variable_name="agents",
            history_variable_name="history",
        )
        termination_strategy = KernelFunctionTerminationStrategy(
            function=self._create_termination_function(),
            kernel=self.service.kernel,
            result_parser=lambda result: str(result.value).strip().lower() == "true",
            agent_variable_name="agents",
            history_variable_name="history",
            maximum_iterations=10,  # Safety net
        )
        return AgentGroupChat(
            agents=[self.researcher, self.writer, self.critic],
            selection_strategy=selection_strategy,
            termination_strategy=termination_strategy,
        )

    def _create_selection_function(self):
        from semantic_kernel.functions import KernelFunctionFromPrompt
        return KernelFunctionFromPrompt(
            function_name="select_next_agent",
            prompt=SELECTION_PROMPT,
        )

    def _create_termination_function(self):
        from semantic_kernel.functions import KernelFunctionFromPrompt
        return KernelFunctionFromPrompt(
            function_name="should_terminate",
            prompt=TERMINATION_PROMPT,
        )

    async def run(self, topic: str) -> List[ChatMessageContent]:
        """Run the multi-agent conversation on a topic. Returns full history."""
        await self.chat.add_chat_message(
            ChatMessageContent(role=AuthorRole.USER, content=f"Topic: {topic}")
        )
        async for message in self.chat.invoke():
            print(f"[{message.name}] {message.content}")
        return self.chat.history

Main entry point

# main.py
import asyncio
import os
from dotenv import load_dotenv
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
from orchestration.group_chat import MultiAgentOrchestrator

load_dotenv()

async def main():
    # Configure the LLM service
    service = AzureChatCompletion(
        deployment_name=os.getenv("AZURE_OPENAI_DEPLOYMENT", "gpt-4o"),
        endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
        api_key=os.getenv("AZURE_OPENAI_API_KEY"),
        api_version=os.getenv("AZURE_OPENAI_API_VERSION", "2024-08-01-preview"),
    )
    # For standard OpenAI, use:
    # service = OpenAIChatCompletion(ai_model_id=os.getenv("OPENAI_MODEL", "gpt-4o"), api_key=os.getenv("OPENAI_API_KEY"))

    orchestrator = MultiAgentOrchestrator(service)
    
    topic = "semantic kernel multi-agent conversation tutorial"
    print(f"=== Starting conversation on: {topic} ===\n")
    
    history = await orchestrator.run(topic)
    
    print("\n=== Conversation complete ===")
    print(f"Total messages: {len(history)}")

if __name__ == "__main__":
    asyncio.run(main())

Run it

python main.py

Expected output (abridged)

=== Starting conversation on: semantic kernel multi-agent conversation tutorial ===

[Researcher] Semantic Kernel is an open-source SDK from Microsoft that lets you combine LLMs with conventional code via planners, plugins, and agents. Multi-agent systems coordinate multiple specialized agents to solve complex tasks through conversation. Group chat in Semantic Kernel orchestrates turn-taking among agents using a selection strategy and termination condition.

[Writer] # Semantic Kernel Multi-Agent Conversations

Semantic Kernel's Agent Framework enables you to build systems where multiple specialized agents collaborate...

## Key Components

1. **ChatCompletionAgent** — wraps an LLM with instructions and optional plugins
2. **AgentGroupChat** — orchestrates turn-taking via selection/termination strategies
3. **KernelFunctionSelectionStrategy** — uses an LLM to pick the next speaker
4. **KernelFunctionTerminationStrategy** — uses an LLM to decide when to stop

## Turn Flow

User → Researcher (tools) → Writer → Critic → (revise?) → Writer → Critic → APPROVED

[Critic] REVISION NEEDED: The explanation lacks a concrete code example showing how to instantiate the group chat. Add a minimal snippet for `AgentGroupChat` construction.

[Writer] # Semantic Kernel Multi-Agent Conversations (Revised)

... [updated draft with code snippet] ...

[Critic] APPROVED

=== Conversation complete ===
Total messages: 5

Key implementation details

Selection strategy nuances

The selection prompt receives {{$history}} — the full conversation so far. The LLM decides the next speaker. This is flexible but adds latency (one extra LLM call per turn). For lower latency, implement a deterministic strategy:

# orchestration/deterministic_selection.py
from semantic_kernel.agents.strategies import SelectionStrategy
from semantic_kernel.contents import ChatHistory
from typing import List
from semantic_kernel.agents import Agent

class RoundRobinSelection(SelectionStrategy):
    def __init__(self, agents: List[Agent]):
        self.agents = agents
        self.index = 0
    
    async def select_next_agent(self, history: ChatHistory, agents: List[Agent]) -> Agent:
        agent = self.agents[self.index % len(self.agents)]
        self.index += 1
        return agent

Use this when the conversation flow is fixed (e.g., always Researcher → Writer → Critic).

Termination strategy safety

Always set maximum_iterations on the termination strategy. The LLM-based termination can fail to trigger if the critic never emits “APPROVED” exactly. The iteration cap prevents infinite loops.

termination_strategy = KernelFunctionTerminationStrategy(
    function=termination_function,
    kernel=service.kernel,
    result_parser=lambda r: str(r.value).strip().lower() == "true",
    maximum_iterations=15,  # Hard stop
)

Passing data between agents

Agents share the ChatHistory. The researcher’s tool results become part of the history automatically. If you need structured handoffs (e.g., researcher outputs JSON that writer parses), add a serialization step in the writer’s instructions:

instructions=(
    "The Researcher's last message contains a JSON object with keys: "
    "'topic', 'facts' (array), 'sources' (array). Parse it and write the article."
)

Then have the researcher emit JSON via a function plugin that returns structured data.

Common pitfalls

Extending the pattern

Add a planner agent

Insert a planner before the researcher to break complex topics into sub-questions:

planner = ChatCompletionAgent(
    service=service,
    name="Planner",
    instructions="Break the user's topic into 3-5 specific research questions. Output as a numbered list.",
)

Parallel research

For independent sub-topics, spawn multiple researcher agents concurrently and merge results. Semantic Kernel’s Process framework (preview) handles this with explicit step definitions.

Human-in-the-loop

Add a HumanAgent that pauses execution and waits for console input:

class HumanAgent(ChatCompletionAgent):
    async def invoke(self, history: ChatHistory) -> AsyncIterable[ChatMessageContent]:
        user_input = input("[Human] Your feedback: ")
        yield ChatMessageContent(role=AuthorRole.USER, content=user_input, name="Human")

Wire it into the group chat like any other agent.

Production considerations

• Observability: Log every turn with agent name, token counts, and latency. Semantic Kernel emits FunctionInvocationContext events you can subscribe to.
• Cost control: Set max_tokens on each agent’s service config. Use cheaper models for selection/termination strategies (they only need reasoning, not knowledge).
• Reliability: Wrap the group chat in a retry policy with exponential backoff for transient LLM errors.
• Routing: If you use a gateway that supports per-model routing (e.g., n4n.ai), you can assign different models to different agents — GPT-4o for the writer, a smaller model for the critic — without changing agent code.

Next steps

1. Replace the in-memory ResearchPlugin with a real search API (Bing, SerpAPI, or your internal knowledge base).
2. Add structured output validation using Pydantic models and KernelFunctionFromPrompt with response_format.
3. Explore the Process framework for declarative multi-step workflows with explicit state machines.
4. Add evaluation harnesses that score critic approval rates and revision counts across test topics.

The complete runnable example is in the multi_agent_tutorial/ structure above. Start there, swap the plugins for your domain, and iterate on the selection/termination prompts until the conversation flow matches your requirements.