Skip to content

Tutorial

This tutorial walks you through a complete example that demonstrates how Group Genie enables AI agents to participate in group chat conversations. You'll learn how to set up a group session, configure a group reasoner to detect conversation patterns, and connect an agent that responds to queries generated by the group reasoner.

About Group Genie

While many AI agents excel at responding to direct queries from individual users, they typically cannot handle multi-party conversations where relevant information emerges from complex exchanges between multiple participants. Group Genie combines Group Sense's intelligent pattern detection with a flexible agent integration layer, allowing existing single-user agents to participate naturally in group chats without requiring any modification to the agents themselves.

Example Scenario

In this example, we'll create a fact-checking assistant that monitors a group chat for factual inconsistencies. When the group reasoner detects contradictory statements, it sends a self-contained query, reformulated from the conversation context, to the system agent to verify facts through web search and respond to the group.

Consider this group chat exchange:

  • user1: "I'm going to Vienna tomorrow"
  • user2: "Enjoy your time there!"
  • user3: "Cool, plan a visit to the Hofbräuhaus!"

The third message contains a factual inconsistency (Hofbräuhaus is in Munich, not Vienna). The group reasoner will detect this and delegate to the system agent, which will search the web, identify the mistake, and respond with a clarification.

Core Components

The tutorial demonstrates four essential components:

Group Session

GroupSession is the main entry point that orchestrates message flow through group reasoners and agents. It provides concurrent processing for different users while ensuring messages from the same sender are processed sequentially. It also manages the lifecycle of both reasoners and agents.

Group Reasoner

A GroupReasoner analyzes group chat messages and decides whether to ignore them or generate queries for agents. In this example, we use DefaultGroupReasoner configured with a fact-checking prompt that detects contradictory statements between messages.

System Agent

The system agent is the primary target for processing delegated queries generated by the group reasoner. It can be any Agent; the name system agent reflects that it can serve as a facade to a larger system of agents (e.g. acting as a coordinator that delegates to subagents), but it can also be a standalone agent that directly handles queries.

In this example, we use a standalone DefaultAgent with access to web search through an MCP server, allowing it to verify facts by searching online.

Factories

Each group chat participant owns both a group reasoner instance and a system agent instance, allowing independent reasoning state and credentials. These are created by:

  • GroupReasonerFactory creates group reasoners with user-specific system prompts and reasoning states
  • AgentFactory creates system agent instances with user-specific credentials, enabling agents to act on behalf of individual group members

Implementation Walkthrough

Secrets Provider

We implement SecretsProvider, an interface designed to retrieve user-specific credentials:

examples/factory/secrets.py
import os

from group_genie.secrets import SecretsProvider


class EnvironmentSecretsProvider(SecretsProvider):
    def get_secrets(self, username: str) -> dict[str, str] | None:
        # For development: use environment variables for all users
        var_names = ["OPENAI_API_KEY", "GOOGLE_API_KEY", "BRAVE_API_KEY"]
        return {var_name: os.getenv(var_name, "") for var_name in var_names}

In this development example, we just return the same set of environment variables for all users. In production, you would implement per-user credential storage and retrieval.

Group Reasoner Factory

We use the get_group_reasoner_factory helper to obtain a GroupReasonerFactory for creating user-specific reasoner instances:

examples/factory/pydantic_ai/reasoner_factory.py
from functools import partial

from pydantic_ai.models.google import GoogleModel
from pydantic_ai.providers.google import GoogleProvider

from examples.utils import load_reasoner_template
from group_genie.agent.provider.pydantic_ai import DefaultGroupReasoner
from group_genie.reasoner import GroupReasoner, GroupReasonerFactory
from group_genie.secrets import SecretsProvider


def create_group_reasoner(
    system_template: str,
    secrets: dict[str, str],
    owner: str,
) -> GroupReasoner:
    model = GoogleModel(
        "gemini-2.5-flash",
        provider=GoogleProvider(api_key=secrets.get("GOOGLE_API_KEY", "")),
    )
    return DefaultGroupReasoner(
        system_prompt=system_template.format(owner=owner),
        model=model,
    )


def get_group_reasoner_factory(
    secrets_provider: SecretsProvider | None = None,
    template_name: str = "general_assist",
):
    system_template = load_reasoner_template(template_name)
    return GroupReasonerFactory(
        group_reasoner_factory_fn=partial(create_group_reasoner, system_template),
        secrets_provider=secrets_provider,
    )

The create_group_reasoner function receives a system prompt template, secrets, and the owner's username and returns a configured GroupReasoner. It:

  1. Creates a Gemini model instance with the owner's Google API key from their secrets
  2. Formats the system prompt template with the owner's username
  3. Returns a DefaultGroupReasoner configured with the formatted system prompt and model

Agent Factory

Agent frameworks

Group Genie supports multiple agent frameworks through the Agent interface. The following example factory is defined in pydantic_ai/agent_factory_1.py. It uses Pydantic AI through a default implementation of the Agent interface.

An equivalent example using the OpenAI Agents SDK with another default implementation of the Agent interface is defined in openai/agent_factory_1.py. You can also integrate any other agent framework or API by implementing the Agent interface directly.

We use the get_agent_factory helper to obtain an AgentFactory for creating user-specific system agent instances:

examples/factory/pydantic_ai/agent_factory_1.py
from pydantic_ai.mcp import MCPServerStdio
from pydantic_ai.models.google import GoogleModel, GoogleModelSettings
from pydantic_ai.providers.google import GoogleProvider

from group_genie.agent import Agent, AgentFactory
from group_genie.agent.provider.pydantic_ai import DefaultAgent
from group_genie.secrets import SecretsProvider


def create_system_agent(secrets: dict[str, str]) -> Agent:
    brave_mcp_server = MCPServerStdio(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-brave-search"],
        env={
            "BRAVE_API_KEY": secrets.get("BRAVE_API_KEY", ""),
        },
    )

    model = GoogleModel(
        "gemini-2.5-flash",
        provider=GoogleProvider(api_key=secrets.get("GOOGLE_API_KEY", "")),
    )

    return DefaultAgent(
        system_prompt=(
            "You are a helpful assistant. "
            "Always search the web for checking facts. "
            "Provide short, concise answers."
        ),
        model=model,
        model_settings=GoogleModelSettings(
            google_thinking_config={
                "thinking_budget": 0,
            }
        ),
        toolsets=[brave_mcp_server],
    )


def get_agent_factory(secrets_provider: SecretsProvider | None = None):
    return AgentFactory(
        system_agent_factory=create_system_agent,
        secrets_provider=secrets_provider,
    )

The create_system_agent function receives the owner's secrets and returns a configured Agent. It:

  1. Configures a Brave Search MCP server with the owner's Brave API key
  2. Configures a Gemini model instance with the owner's Google API key
  3. Returns a DefaultAgent configured with:
    • A system prompt instructing it to search the web for fact-checking
    • The configured Gemini model
    • The configured Brave Search MCP server

Group Session

Now we bring everything together by creating a GroupSession:

examples/guide/tutorial.py
import asyncio
import logging
from pathlib import Path
from uuid import uuid4

from examples.factory.pydantic_ai.agent_factory_1 import get_agent_factory
from examples.factory.pydantic_ai.reasoner_factory import get_group_reasoner_factory
from examples.factory.secrets import EnvironmentSecretsProvider
from group_genie.agent import Approval, Decision
from group_genie.datastore import DataStore
from group_genie.message import Message
from group_genie.session import Execution, GroupSession


async def complete_execution(execution: Execution) -> None:
    async for elem in execution.stream():
        match elem:
            case Decision():
                # log group reasoner decision
                logger.debug(elem)
            case Approval():
                # log tool call approval request
                logger.debug(elem)
                # approve tool call
                elem.approve()
            case Message():
                # log agent response
                logger.debug(elem)


secrets_provider = EnvironmentSecretsProvider()
session_id = uuid4().hex[:8]
session = GroupSession(
    id=session_id,
    group_reasoner_factory=get_group_reasoner_factory(
        secrets_provider=secrets_provider,
        template_name="fact_check",
    ),
    agent_factory=get_agent_factory(secrets_provider=secrets_provider),
    data_store=DataStore(root_path=Path(".data", "tutorial")),
)

chat = [  # example group chat messages
    # no factual inconsistency, group reasoner will ignore the message.
    Message(content="I'm going to Vienna tomorrow", sender="user1"),
    # no factual inconsistency, group reasoner will ignore the message.
    Message(content="Enjoy your time there!", sender="user2"),
    # factual inconsistency in response to user1's message.
    # Group reasoner will delegate to system agent for fact checking.
    Message(content="Cool, plan a visit to the Hofbräuhaus!", sender="user3"),
]

# Add chat messages to session and create execution objects
executions = [session.handle(msg) for msg in chat]

# Concurrently process group chat messages. The complete_execution()
# helper logs reasoner decisions and agent responses to the console.
coros = [complete_execution(exec) for exec in executions]
await asyncio.gather(*coros)

This code:

  1. Creates a secrets provider to supply API keys
  2. Generates a unique session ID for this group chat
  3. Initializes a GroupSession with:
    • A group reasoner factory configured for fact-checking
    • An agent factory that creates system agents with web search capabilities
    • A DataStore for persisting messages, reasoner and agent states
  4. Defines a sample group chat with three messages, where the third contains a factual inconsistency
  5. Handles each message by calling handle(), which returns an Execution object
  6. Concurrently consumes execution streams with the complete_execution helper and asyncio.gather()

An execution stream yields three types of elements:

  • Decision: The group reasoner's decision about whether to ignore or delegate the message
  • Approval: Requests for approval of tool calls (e.g., web searches)
  • Message: The agent's response to be sent to the group

In this example, we log all events and automatically approve all tool calls. In a production application, you might implement selective approval logic or user confirmation for sensitive operations.

Running the Example

Development Setup

To set up the environment for running the example, see Development Setup.

To run this example:

  1. Set up your API keys: 

    export GOOGLE_API_KEY="your-gemini-api-key"
export BRAVE_API_KEY="your-brave-search-api-key"

  2. Run the tutorial script: 

    python examples/guide/tutorial.py

The output will show the group reasoner detecting the factual contradiction and the agent searching the web to verify that Hofbräuhaus is actually in Munich, not Vienna. The agent will then generate a response clarifying this mistake for the group. The output should look like this:

2025-11-05 11:06:15,947 DEBUG __main__: Decision.IGNORE
2025-11-05 11:06:17,085 DEBUG __main__: Decision.IGNORE
2025-11-05 11:06:19,526 DEBUG __main__: Decision.DELEGATE
2025-11-05 11:06:20,336 DEBUG __main__: [sender="system"] brave_web_search(query='Hofbräuhaus location')
2025-11-05 11:06:22,399 DEBUG __main__: Message(content='The Hofbräuhaus is a famous beer hall located in Munich, Germany, not Vienna, Austria. Therefore, a visit to the Hofbräuhaus would not be possible if you are going to Vienna.', sender='system', receiver=None, threads=[], attachments=[], request_id=None)