Multi-Agent AI Swarm with Telegram

Build 3 AI agents that collaborate on tasks and post live to Telegram.

Introduction

30 Second Summary

Imagine having a team of AI specialists you can assign any task to, then watch them debate and collaborate in real time from your phone. That is exactly what you will build in this project.

In this project, you will create a multi-agent chat system where three AI agents autonomously discuss tasks you assign. Their conversations are visible through Telegram, and they remember everything from past sessions.

What You'll Build

A Python-based multi-agent orchestrator that coordinates three specialized AI agents through Telegram, powered by Google Gemini's free tier.

By the end of this project, you'll have:

  • A working multi-agent orchestrator where three specialized AI agents (Architect, Developer, Reviewer) discuss and solve tasks autonomously.
  • A Telegram group dashboard showing live agent conversations with a command bot for triggering and monitoring tasks.
  • A structured markdown log library with keyword-based memory retrieval that agents use to recall and reference past discussions.
  • Secret Mission: Add Groq as a fallback LLM provider with automatic failover when Gemini rate limits are hit.

Are there any prerequisites?

You need Python 3.10+ and pip installed on your machine, a Google account (for a free Gemini API key), and a Telegram account. No billing or credit card is required for any service in this project.

Set Up Your Environment

Before you can orchestrate a swarm of AI agents, you need the right tools in place. This step gets your Python packages installed, your Telegram bots created, and your Gemini API key working.

By the end of this step, you'll have four Telegram bots talking to a shared group and a confirmed connection to Google's free Gemini 2.5 Flash model.

In this step, get ready to:
  • Install the required Python packages for AI calls, Telegram integration, and configuration.
  • Create four Telegram bots and a shared group via BotFather.
  • Obtain and test a free Gemini API key from Google AI Studio.
Install the required Python packages

This project relies on three Python packages. The openai package talks to Gemini's OpenAI-compatible endpoint. The python-telegram-bot package handles Telegram Bot API interactions. The pyyaml package reads your configuration file.

  • Confirm your Python version is 3.10+ by running this command:
python3 --version

✔️ I see version 3.10 or higher

Your Python version is ready for this project.

ⓧ I see a version lower than 3.10

The python-telegram-bot library requires Python 3.10+. You need to upgrade your Python installation.

  • Visit python.org/downloads and download Python 3.10 or newer.
  • Run the installer and follow the prompts.
  • After installation, close and reopen your terminal.
  • Run python3 --version again to confirm it now shows 3.10 or higher.

ⓧ Command not found

Python is not installed on your system.

  • Visit python.org/downloads and download Python 3.10 or newer.
  • Run the installer and follow the prompts.
  • Close and reopen your terminal after installation.
  • Run python3 --version again to confirm it now shows 3.10 or higher.
  • Install the three project packages by running this command:
pip install openai python-telegram-bot pyyaml
  • Verify the packages installed correctly by running:
python3 -c "import openai; import telegram; import yaml; print('All packages imported successfully')"

You should see All packages imported successfully printed in your terminal.

Seeing an import error?

If you see ModuleNotFoundError, your pip might be installing to a different Python version than the one you're running. Try using pip3 instead of pip, or run python3 -m pip install openai python-telegram-bot pyyaml to make sure they match.

help me debug a ModuleNotFoundError when importing openai, telegram, or yaml in Python

Create your Telegram bots and group

You need four separate Telegram bots for this project. One command bot (SwarmCommander) takes your instructions. Three agent bots (AgentArchitect, AgentDeveloper, AgentReviewer) will post as the AI agents in a shared group.

  • Open Telegram on your phone or desktop.
  • Search for @BotFather and start a chat with it.
  • Send /newbot to create your first bot.
  • When prompted for a name, enter SwarmCommander.
  • When prompted for a username, enter a unique username ending in bot (for example, swarm_commander_yourname_bot).
  • Copy the bot token that BotFather gives you and save it somewhere safe.

What is this bot token?

The bot token is a long string that looks like 123456789:ABCdefGhIjKlmNoPqRsTuVwXyZ. It authenticates your Python code as the bot when sending messages. Keep it private.

  • Repeat the /newbot process three more times to create:
  • A bot named AgentArchitect (username example: agent_architect_yourname_bot).
  • A bot named AgentDeveloper (username example: agent_developer_yourname_bot).
  • A bot named AgentReviewer (username example: agent_reviewer_yourname_bot).
  • Save each bot token separately. You now have four tokens total.

Next, you need to disable privacy mode on each bot so they can read messages in the group.

  • In the BotFather chat, send /setprivacy.
  • Select your SwarmCommander bot from the list.
  • Choose Disable to turn off privacy mode.
  • Repeat /setprivacy and Disable for AgentArchitect, AgentDeveloper, and AgentReviewer.

Why disable privacy mode?

By default, Telegram bots in groups only see messages starting with a / command. Disabling privacy mode lets bots receive all messages in the group. Note that bots still cannot see messages from other bots. This is a Telegram limitation. Your Python orchestrator handles all coordination server-side.

Now create the shared Telegram group and add your bots.

  • In Telegram, tap the compose/new message button and select New Group.
  • Name the group something like Agent Swarm.
  • Add all four bots to the group by searching for their usernames.
  • Open the group settings and promote each bot to Administrator.
  • Verify you see all four bots listed as admins in the group info.

Finding your group chat ID

You'll need your group's chat ID later for the config file. Add @RawDataBot to the group, and it will post a message showing the chat ID (a negative number like -1001234567890). Save this number, then remove RawDataBot from the group.

Can't find your bots when adding to the group?

Make sure you're searching by the bot's username (the one ending in bot), not the display name. If bots still don't appear, send a /start message to each bot first in a private chat, then try adding them to the group again.

help me add my Telegram bots to a group

Get and test your Gemini API key

Google's Gemini 2.5 Flash model is completely free to use through Google AI Studio. No billing setup is required. You'll connect to it using the OpenAI-compatible endpoint, which lets you swap providers easily later.

  • Open aistudio.google.com in your browser.
  • Sign in with your Google account.
  • Click Get API key in the left sidebar.
  • Click Create API key.
  • Copy the generated key and save it somewhere safe.

What does the free tier give you?

Gemini 2.5 Flash on the free tier gives you 1,500 requests per day and 15 requests per minute. This is more than enough for development and testing your agent swarm.

Now verify your API key works by sending a test request to the OpenAI-compatible endpoint.

  • Test your Gemini API key by running this curl command (replace your-gemini-api-key-here with your actual key):
curl "https://generativelanguage.googleapis.com/v1beta/openai/chat/completions" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer your-gemini-api-key-here" \
  -d '{
    "model": "gemini-2.5-flash",
    "messages": [
      {
        "role": "user",
        "content": "Say hello in one sentence."
      }
    ]
  }'

You should see a JSON response containing a choices array with the model's reply. This confirms your key is valid and gemini-2.5-flash is accessible.

Why the OpenAI-compatible endpoint?

Instead of using Google's native SDK, you're using Gemini's OpenAI-compatible endpoint at https://generativelanguage.googleapis.com/v1beta/openai/. This means you can swap between Gemini, Groq, or any other OpenAI-compatible provider just by changing the base URL and API key.

Getting an authentication error?

Double-check that you pasted the full API key with no extra spaces. Make sure you're using Bearer (with a capital B) before the key in the Authorization header. If you just created the key, wait 30 seconds and try again.

help me debug a Gemini API authentication error

Your environment is locked and loaded. Next up, you'll build the agent orchestrator that manages conversations between your three AI specialists.

Build the Agent Orchestrator

Your Telegram bots and Gemini API key are ready. Now you need the brain that ties everything together.

In this step, you will build the core orchestrator that manages three AI agent roles, routes conversations between them in round-robin turns, and calls Gemini via the OpenAI-compatible API. You will also create the logging and memory modules that record every conversation and retrieve relevant context for future discussions.

In this step, get ready to:
  • Create the project configuration and connect to Gemini's AI.
  • Build a conversation logger and memory retriever for structured markdown logs.
  • Wire up an orchestrator that manages multi-agent discussions in round-robin turns.
Create the project structure and connect to Gemini

Before writing any orchestration logic, you need a project folder, a configuration file that defines your agents, and a provider module that wraps all LLM calls.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir agent-swarm
cd agent-swarm
  • Confirm the folder exists by running:
ls

You should see the agent-swarm folder listed in the output.

  • Create the configuration file by running:
touch config.yaml
  • Open config.yaml in your preferred text editor and paste the following configuration:
gemini:
  api_key: "your-gemini-api-key-here"
  base_url: "https://generativelanguage.googleapis.com/v1beta/openai/"
  model: "gemini-2.5-flash"

groq:
  api_key: "your-groq-api-key-here"
  base_url: "https://api.groq.com/openai/v1"
  model: "llama-3.3-70b-versatile"

telegram:
  group_chat_id: -1001234567890
  command_bot:
    token: "your-commander-bot-token-here"
    name: "SwarmCommander"

agents:
  - name: "Architect"
    token: "your-architect-bot-token-here"
    system_prompt: |
      You are a software architect agent. Your role is to:
      - Analyze requirements and propose high-level system designs
      - Define component boundaries, data flow, and API contracts
      - Consider scalability, maintainability, and trade-offs
      - Respond concisely (2-4 paragraphs max)
      When other agents provide feedback, incorporate it constructively.

  - name: "Developer"
    token: "your-developer-bot-token-here"
    system_prompt: |
      You are a developer agent. Your role is to:
      - Take architectural designs and propose implementation details
      - Suggest specific technologies, libraries, and patterns
      - Identify potential implementation challenges and solutions
      - Respond concisely (2-4 paragraphs max)
      Build on what the Architect proposes and flag any practical concerns.

  - name: "Reviewer"
    token: "your-reviewer-bot-token-here"
    system_prompt: |
      You are a code reviewer agent. Your role is to:
      - Critically evaluate proposals from Architect and Developer
      - Identify security concerns, edge cases, and missing requirements
      - Suggest improvements and best practices
      - Respond concisely (2-4 paragraphs max)
      Be constructive but thorough. If something looks good, say so briefly.

orchestrator:
  max_rounds: 3
  log_dir: "logs"

What does this config do?

This file defines the entire swarm in one place. The gemini section points the OpenAI SDK at Gemini's compatible endpoint. The groq section is a placeholder for the Secret Mission later.

Each agent has a unique Telegram bot token and a system prompt that shapes its personality. The orchestrator section controls how many rounds the agents discuss before stopping.

  • Replace your-gemini-api-key-here with your actual Gemini API key from Google AI Studio.
  • Replace each bot token placeholder (your-commander-bot-token-here, your-architect-bot-token-here, etc.) with the tokens you received from BotFather.
  • Leave the groq section as-is for now. You will configure it in the Secret Mission.

How do I find my group chat ID?

Open this URL in your browser, replacing the token with your SwarmCommander bot token:

https://api.telegram.org/bot<YOUR-COMMANDER-TOKEN>/getUpdates

Send any message in your Telegram group first. Then refresh that URL and look for the chat object in the JSON response. The id field (a negative number starting with -100) is your group chat ID. Paste it into group_chat_id in config.yaml.

  • Save config.yaml.

Now you need a provider module that wraps all LLM calls behind a single interface. This lets you swap providers later without touching the orchestrator.

  • Create provider.py in your agent-swarm folder and paste this code:
import time
from openai import OpenAI


class LLMProvider:
    def __init__(self, config):
        self.gemini = OpenAI(
            api_key=config["gemini"]["api_key"],
            base_url=config["gemini"]["base_url"],
        )
        self.gemini_model = config["gemini"]["model"]
        self.groq = None
        self.groq_model = None
        if config.get("groq", {}).get("api_key"):
            self.groq = OpenAI(
                api_key=config["groq"]["api_key"],
                base_url=config["groq"]["base_url"],
            )
            self.groq_model = config["groq"]["model"]
        self.using_fallback = False
        self.fallback_until = 0

    def chat(self, messages, system_prompt=None):
        full_messages = []
        if system_prompt:
            full_messages.append({"role": "system", "content": system_prompt})
        full_messages.extend(messages)

        if self.using_fallback and time.time() > self.fallback_until:
            self.using_fallback = False

        if not self.using_fallback:
            try:
                response = self.gemini.chat.completions.create(
                    model=self.gemini_model,
                    messages=full_messages,
                )
                return response.choices[0].message.content, "gemini"
            except Exception as e:
                if "429" in str(e) and self.groq:
                    self.using_fallback = True
                    self.fallback_until = time.time() + 60
                else:
                    raise

        if self.groq:
            response = self.groq.chat.completions.create(
                model=self.groq_model,
                messages=full_messages,
            )
            return response.choices[0].message.content, "groq"

        raise RuntimeError("All providers exhausted")

What does this code do?

The LLMProvider class creates an OpenAI SDK client pointed at Gemini's base URL. Every call to chat() sends messages to gemini-2.5-flash and returns both the response text and which provider served it.

The fallback logic is pre-wired for the Secret Mission. Right now, with no Groq key configured, it simply uses Gemini for every call.

  • Save provider.py.
  • Test the connection to Gemini by running this command:
python3 -c "
import yaml
from provider import LLMProvider

with open('config.yaml') as f:
    config = yaml.safe_load(f)

provider = LLMProvider(config)
response, source = provider.chat([{'role': 'user', 'content': 'Say hello in one sentence.'}])
print(f'[{source}] {response}')
"

You should see a one-sentence greeting prefixed with [gemini] in your terminal output.

Seeing an authentication error?

Double-check that the api_key in config.yaml matches your Gemini key exactly. Make sure there are no trailing spaces or line breaks inside the quotes.

If you see a connection timeout, confirm your machine has internet access and that https://generativelanguage.googleapis.com is not blocked by a firewall.

help me debug my Gemini API connection error

Build the conversation router

The orchestrator needs two support modules before it can manage conversations. A logger records each agent's message as structured markdown. A memory retriever searches past logs so agents can reference earlier discussions.

  • Create logger.py in your agent-swarm folder and paste this code:
import os
import re
from datetime import datetime


class ConversationLogger:
    def __init__(self, log_dir="logs"):
        self.log_dir = log_dir
        self.sessions_dir = os.path.join(log_dir, "sessions")
        os.makedirs(self.sessions_dir, exist_ok=True)

    def _slugify(self, text):
        slug = re.sub(r"[^\w\s-]", "", text.lower())
        slug = re.sub(r"[\s_]+", "-", slug)
        return slug[:50]

    def create_session(self, task, agents, categories=None):
        date_str = datetime.now().strftime("%Y-%m-%d")
        slug = self._slugify(task)
        session_dir = os.path.join(self.sessions_dir, f"{date_str}_{slug}")
        os.makedirs(session_dir, exist_ok=True)

        filepath = os.path.join(session_dir, "conversation.md")
        frontmatter = (
            f"---\n"
            f"task: \"{task}\"\n"
            f"agents: [{', '.join(agents)}]\n"
            f"categories: [{', '.join(categories or ['general'])}]\n"
            f"started: \"{datetime.now().isoformat()}\"\n"
            f"status: in-progress\n"
            f"---\n\n"
            f"# {task}\n\n"
        )

        with open(filepath, "w") as f:
            f.write(frontmatter)

        return filepath

    def log_message(self, filepath, agent_name, content, provider="gemini"):
        timestamp = datetime.now().strftime("%H:%M:%S")
        entry = (
            f"## [{timestamp}] {agent_name}\n\n"
            f"{content}\n\n"
            f"_provider: {provider}_\n\n"
            f"---\n\n"
        )

        with open(filepath, "a") as f:
            f.write(entry)

    def complete_session(self, filepath):
        with open(filepath, "r") as f:
            content = f.read()

        content = content.replace("status: in-progress", "status: completed")

        with open(filepath, "w") as f:
            f.write(content)

What does this code do?

The ConversationLogger class manages a logs/sessions/ directory. Each task gets its own dated folder with a conversation.md file containing YAML frontmatter (task name, agents, timestamp, status).

Every agent message is appended with a timestamp and provider tag. When the task finishes, complete_session() flips the status from in-progress to completed.

  • Save logger.py.
  • Create memory.py in the same folder and paste this code:
import os
import re
from datetime import datetime


class MemoryRetriever:
    def __init__(self, log_dir="logs"):
        self.sessions_dir = os.path.join(log_dir, "sessions")

    def search(self, query, max_results=3, category=None):
        results = []
        query_terms = query.lower().split()

        if not os.path.exists(self.sessions_dir):
            return results

        for session_name in sorted(os.listdir(self.sessions_dir), reverse=True):
            session_path = os.path.join(
                self.sessions_dir, session_name, "conversation.md"
            )
            if not os.path.exists(session_path):
                continue

            with open(session_path, "r") as f:
                content = f.read()

            if category:
                cat_match = re.search(r"categories:\s*\[(.+?)\]", content)
                if cat_match and category.lower() not in cat_match.group(1).lower():
                    continue

            score = 0
            content_lower = content.lower()

            task_match = re.search(r'task:\s*"(.+?)"', content)
            task_text = task_match.group(1).lower() if task_match else ""

            for term in query_terms:
                if term in task_text:
                    score += 10
                score += content_lower.count(term)

            if score > 0:
                snippet = self._extract_snippet(content, query_terms)
                results.append(
                    {"session": session_name, "score": score, "snippet": snippet}
                )

        results.sort(key=lambda x: x["score"], reverse=True)
        return results[:max_results]

    def _extract_snippet(self, content, terms, max_length=500):
        lines = content.split("\n")
        relevant_lines = []

        for i, line in enumerate(lines):
            if any(term in line.lower() for term in terms):
                start = max(0, i - 1)
                end = min(len(lines), i + 3)
                relevant_lines.extend(lines[start:end])
                if len("\n".join(relevant_lines)) > max_length:
                    break

        snippet = "\n".join(relevant_lines)[:max_length]
        return snippet if snippet else content[200:700]

    def get_context_for_agent(self, task_description, max_snippets=2):
        results = self.search(task_description, max_results=max_snippets)
        if not results:
            return ""

        context = "## Relevant past discussions:\n\n"
        for r in results:
            context += f"### From session: {r['session']}\n"
            context += f"{r['snippet']}\n\n"

        return context

What does this code do?

The MemoryRetriever searches all past conversation logs using keyword matching. It scores results by checking whether query terms appear in the task title (high weight) or in the content body (lower weight).

The get_context_for_agent() method returns a formatted string of relevant past discussions. The orchestrator injects this into the conversation so agents have memory of previous sessions.

Why keyword search instead of vector embeddings?

Embeddings require a vector database and extra API calls. For the volume of conversations you will generate in this project, keyword search over structured markdown is fast, free, and sufficient.

Gemini's 1M token context window means you can inject full snippets directly. The markdown structure is designed so you can add embeddings later without restructuring any files.

  • Save memory.py.

Now you have the support modules. Time to build the orchestrator itself.

  • Create orchestrator.py in the same folder and paste the imports, Agent class, and Orchestrator initialization:
import asyncio
import yaml

from provider import LLMProvider
from logger import ConversationLogger
from memory import MemoryRetriever


class Agent:
    def __init__(self, name, system_prompt, token):
        self.name = name
        self.system_prompt = system_prompt
        self.token = token


class Orchestrator:
    def __init__(self, config_path="config.yaml"):
        with open(config_path, "r") as f:
            self.config = yaml.safe_load(f)

        self.provider = LLMProvider(self.config)
        self.logger = ConversationLogger(self.config["orchestrator"]["log_dir"])
        self.memory = MemoryRetriever(self.config["orchestrator"]["log_dir"])
        self.agents = []
        for agent_cfg in self.config["agents"]:
            self.agents.append(
                Agent(
                    name=agent_cfg["name"],
                    system_prompt=agent_cfg["system_prompt"],
                    token=agent_cfg["token"],
                )
            )
        self.max_rounds = self.config["orchestrator"]["max_rounds"]
        self.running_task = None

What does this code do?

The Agent class is a simple data holder for each agent's name, system prompt, and Telegram token.

The Orchestrator.__init__ reads config.yaml, creates a provider, logger, and memory retriever, then instantiates all three agents from the config. The max_rounds setting controls how many full cycles the agents complete per task.

  • Save orchestrator.py. Do not close the file yet. You will add more code in the next substep.
  • Verify the Orchestrator loads correctly by running:
python3 -c "
from orchestrator import Orchestrator
o = Orchestrator()
print(f'Loaded {len(o.agents)} agents: {[a.name for a in o.agents]}')
"

You should see Loaded 3 agents: ['Architect', 'Developer', 'Reviewer'] in your terminal.

Seeing a ModuleNotFoundError?

Make sure you are running the command from inside the agent-swarm folder where all your .py files live. Run ls and confirm you see config.yaml, provider.py, logger.py, memory.py, and orchestrator.py.

help me fix a ModuleNotFoundError when importing my orchestrator modules

Add the task manager and run a full discussion

The orchestrator knows its agents and can call Gemini. Now it needs the run_task() method that drives a multi-round discussion and a stop_task() method to halt one mid-conversation.

  • In orchestrator.py, add the following methods inside the Orchestrator class, directly below the __init__ method:
    async def run_task(self, task, on_message=None):
        self.running_task = task
        agent_names = [a.name for a in self.agents]
        log_path = self.logger.create_session(task, agent_names)

        memory_context = self.memory.get_context_for_agent(task)
        conversation = []

        if memory_context:
            conversation.append(
                {"role": "user", "content": f"[CONTEXT FROM PAST SESSIONS]\n{memory_context}"}
            )
            conversation.append(
                {"role": "assistant", "content": "Understood, I will reference this context."}
            )

        conversation.append({"role": "user", "content": f"Task: {task}"})

        results = []

        for round_num in range(self.max_rounds):
            for agent in self.agents:
                if self.running_task is None:
                    break

                response, used_provider = self.provider.chat(
                    messages=conversation,
                    system_prompt=agent.system_prompt,
                )

                conversation.append({"role": "assistant", "content": f"[{agent.name}]: {response}"})
                conversation.append(
                    {"role": "user", "content": f"Next agent, please continue the discussion."}
                )

                self.logger.log_message(log_path, agent.name, response, used_provider)
                results.append({"agent": agent.name, "content": response, "provider": used_provider})

                if on_message:
                    await on_message(agent, response, used_provider)

        self.logger.complete_session(log_path)
        self.running_task = None
        return results

    def stop_task(self):
        self.running_task = None

What does this code do?

The run_task() method is the heart of the swarm. It starts a logging session, checks for relevant past discussions via the memory module, then enters a loop that cycles through all agents for the configured number of rounds.

Each agent receives the full conversation history plus its own system prompt. Its response is tagged with the agent name and appended to the shared history. This round-robin pattern means every agent builds on what all previous agents said.

The optional on_message callback lets you hook in Telegram posting later without changing the orchestrator code. The stop_task() method sets running_task to None, which the loop checks before each agent turn.

  • Save orchestrator.py.
  • Test the full multi-agent discussion by running:
python3 -c "
import asyncio
from orchestrator import Orchestrator

async def main():
    o = Orchestrator()
    results = await o.run_task('Design a REST API for a todo app')
    for r in results:
        print(f\"[{r['agent']}] ({r['provider']}): {r['content'][:80]}...\")
    print(f'\nTotal messages: {len(results)}')

asyncio.run(main())
"

You should see 9 messages (3 agents across 3 rounds), each prefixed with the agent name and provider. This confirms your orchestrator is routing conversations correctly.

What should I see?

Each agent responds in sequence: Architect first with a high-level design, Developer second with implementation details, Reviewer third with feedback. This repeats for 3 rounds. The total output is 9 messages.

A logs/sessions/ folder has also been created inside your project directory. It contains a dated subfolder with a conversation.md file recording the full discussion.

  • Verify the log was created by running:
ls logs/sessions/

You should see a folder named with today's date and the task slug, like 2026-06-09_design-a-rest-api-for-a-todo-app.

Seeing a rate limit error?

Gemini's free tier allows 15 requests per minute. Nine agent calls in quick succession can get close to this limit. If you see a 429 error, wait 60 seconds and try again.

You can also reduce max_rounds from 3 to 1 in config.yaml for testing, then set it back to 3 once you confirm it works.

help me fix a Gemini rate limit error in my multi-agent orchestrator

✔️ Awesome, I've got everything!

Great. Double check you have saved all five files: config.yaml, provider.py, logger.py, memory.py, and orchestrator.py.

ⓧ I'd like to double check the full code

Here is the complete code for each file at this point:

gemini:
  api_key: "your-gemini-api-key-here"
  base_url: "https://generativelanguage.googleapis.com/v1beta/openai/"
  model: "gemini-2.5-flash"

groq:
  api_key: "your-groq-api-key-here"
  base_url: "https://api.groq.com/openai/v1"
  model: "llama-3.3-70b-versatile"

telegram:
  group_chat_id: -1001234567890
  command_bot:
    token: "your-commander-bot-token-here"
    name: "SwarmCommander"

agents:
  - name: "Architect"
    token: "your-architect-bot-token-here"
    system_prompt: |
      You are a software architect agent. Your role is to:
      - Analyze requirements and propose high-level system designs
      - Define component boundaries, data flow, and API contracts
      - Consider scalability, maintainability, and trade-offs
      - Respond concisely (2-4 paragraphs max)
      When other agents provide feedback, incorporate it constructively.

  - name: "Developer"
    token: "your-developer-bot-token-here"
    system_prompt: |
      You are a developer agent. Your role is to:
      - Take architectural designs and propose implementation details
      - Suggest specific technologies, libraries, and patterns
      - Identify potential implementation challenges and solutions
      - Respond concisely (2-4 paragraphs max)
      Build on what the Architect proposes and flag any practical concerns.

  - name: "Reviewer"
    token: "your-reviewer-bot-token-here"
    system_prompt: |
      You are a code reviewer agent. Your role is to:
      - Critically evaluate proposals from Architect and Developer
      - Identify security concerns, edge cases, and missing requirements
      - Suggest improvements and best practices
      - Respond concisely (2-4 paragraphs max)
      Be constructive but thorough. If something looks good, say so briefly.

orchestrator:
  max_rounds: 3
  log_dir: "logs"
import time
from openai import OpenAI


class LLMProvider:
    def __init__(self, config):
        self.gemini = OpenAI(
            api_key=config["gemini"]["api_key"],
            base_url=config["gemini"]["base_url"],
        )
        self.gemini_model = config["gemini"]["model"]
        self.groq = None
        self.groq_model = None
        if config.get("groq", {}).get("api_key"):
            self.groq = OpenAI(
                api_key=config["groq"]["api_key"],
                base_url=config["groq"]["base_url"],
            )
            self.groq_model = config["groq"]["model"]
        self.using_fallback = False
        self.fallback_until = 0

    def chat(self, messages, system_prompt=None):
        full_messages = []
        if system_prompt:
            full_messages.append({"role": "system", "content": system_prompt})
        full_messages.extend(messages)

        if self.using_fallback and time.time() > self.fallback_until:
            self.using_fallback = False

        if not self.using_fallback:
            try:
                response = self.gemini.chat.completions.create(
                    model=self.gemini_model,
                    messages=full_messages,
                )
                return response.choices[0].message.content, "gemini"
            except Exception as e:
                if "429" in str(e) and self.groq:
                    self.using_fallback = True
                    self.fallback_until = time.time() + 60
                else:
                    raise

        if self.groq:
            response = self.groq.chat.completions.create(
                model=self.groq_model,
                messages=full_messages,
            )
            return response.choices[0].message.content, "groq"

        raise RuntimeError("All providers exhausted")
import os
import re
from datetime import datetime


class ConversationLogger:
    def __init__(self, log_dir="logs"):
        self.log_dir = log_dir
        self.sessions_dir = os.path.join(log_dir, "sessions")
        os.makedirs(self.sessions_dir, exist_ok=True)

    def _slugify(self, text):
        slug = re.sub(r"[^\w\s-]", "", text.lower())
        slug = re.sub(r"[\s_]+", "-", slug)
        return slug[:50]

    def create_session(self, task, agents, categories=None):
        date_str = datetime.now().strftime("%Y-%m-%d")
        slug = self._slugify(task)
        session_dir = os.path.join(self.sessions_dir, f"{date_str}_{slug}")
        os.makedirs(session_dir, exist_ok=True)

        filepath = os.path.join(session_dir, "conversation.md")
        frontmatter = (
            f"---\n"
            f"task: \"{task}\"\n"
            f"agents: [{', '.join(agents)}]\n"
            f"categories: [{', '.join(categories or ['general'])}]\n"
            f"started: \"{datetime.now().isoformat()}\"\n"
            f"status: in-progress\n"
            f"---\n\n"
            f"# {task}\n\n"
        )

        with open(filepath, "w") as f:
            f.write(frontmatter)

        return filepath

    def log_message(self, filepath, agent_name, content, provider="gemini"):
        timestamp = datetime.now().strftime("%H:%M:%S")
        entry = (
            f"## [{timestamp}] {agent_name}\n\n"
            f"{content}\n\n"
            f"_provider: {provider}_\n\n"
            f"---\n\n"
        )

        with open(filepath, "a") as f:
            f.write(entry)

    def complete_session(self, filepath):
        with open(filepath, "r") as f:
            content = f.read()

        content = content.replace("status: in-progress", "status: completed")

        with open(filepath, "w") as f:
            f.write(content)
import os
import re
from datetime import datetime


class MemoryRetriever:
    def __init__(self, log_dir="logs"):
        self.sessions_dir = os.path.join(log_dir, "sessions")

    def search(self, query, max_results=3, category=None):
        results = []
        query_terms = query.lower().split()

        if not os.path.exists(self.sessions_dir):
            return results

        for session_name in sorted(os.listdir(self.sessions_dir), reverse=True):
            session_path = os.path.join(
                self.sessions_dir, session_name, "conversation.md"
            )
            if not os.path.exists(session_path):
                continue

            with open(session_path, "r") as f:
                content = f.read()

            if category:
                cat_match = re.search(r"categories:\s*\[(.+?)\]", content)
                if cat_match and category.lower() not in cat_match.group(1).lower():
                    continue

            score = 0
            content_lower = content.lower()

            task_match = re.search(r'task:\s*"(.+?)"', content)
            task_text = task_match.group(1).lower() if task_match else ""

            for term in query_terms:
                if term in task_text:
                    score += 10
                score += content_lower.count(term)

            if score > 0:
                snippet = self._extract_snippet(content, query_terms)
                results.append(
                    {"session": session_name, "score": score, "snippet": snippet}
                )

        results.sort(key=lambda x: x["score"], reverse=True)
        return results[:max_results]

    def _extract_snippet(self, content, terms, max_length=500):
        lines = content.split("\n")
        relevant_lines = []

        for i, line in enumerate(lines):
            if any(term in line.lower() for term in terms):
                start = max(0, i - 1)
                end = min(len(lines), i + 3)
                relevant_lines.extend(lines[start:end])
                if len("\n".join(relevant_lines)) > max_length:
                    break

        snippet = "\n".join(relevant_lines)[:max_length]
        return snippet if snippet else content[200:700]

    def get_context_for_agent(self, task_description, max_snippets=2):
        results = self.search(task_description, max_results=max_snippets)
        if not results:
            return ""

        context = "## Relevant past discussions:\n\n"
        for r in results:
            context += f"### From session: {r['session']}\n"
            context += f"{r['snippet']}\n\n"

        return context
import asyncio
import yaml

from provider import LLMProvider
from logger import ConversationLogger
from memory import MemoryRetriever


class Agent:
    def __init__(self, name, system_prompt, token):
        self.name = name
        self.system_prompt = system_prompt
        self.token = token


class Orchestrator:
    def __init__(self, config_path="config.yaml"):
        with open(config_path, "r") as f:
            self.config = yaml.safe_load(f)

        self.provider = LLMProvider(self.config)
        self.logger = ConversationLogger(self.config["orchestrator"]["log_dir"])
        self.memory = MemoryRetriever(self.config["orchestrator"]["log_dir"])
        self.agents = []
        for agent_cfg in self.config["agents"]:
            self.agents.append(
                Agent(
                    name=agent_cfg["name"],
                    system_prompt=agent_cfg["system_prompt"],
                    token=agent_cfg["token"],
                )
            )
        self.max_rounds = self.config["orchestrator"]["max_rounds"]
        self.running_task = None

    async def run_task(self, task, on_message=None):
        self.running_task = task
        agent_names = [a.name for a in self.agents]
        log_path = self.logger.create_session(task, agent_names)

        memory_context = self.memory.get_context_for_agent(task)
        conversation = []

        if memory_context:
            conversation.append(
                {"role": "user", "content": f"[CONTEXT FROM PAST SESSIONS]\n{memory_context}"}
            )
            conversation.append(
                {"role": "assistant", "content": "Understood, I will reference this context."}
            )

        conversation.append({"role": "user", "content": f"Task: {task}"})

        results = []

        for round_num in range(self.max_rounds):
            for agent in self.agents:
                if self.running_task is None:
                    break

                response, used_provider = self.provider.chat(
                    messages=conversation,
                    system_prompt=agent.system_prompt,
                )

                conversation.append({"role": "assistant", "content": f"[{agent.name}]: {response}"})
                conversation.append(
                    {"role": "user", "content": f"Next agent, please continue the discussion."}
                )

                self.logger.log_message(log_path, agent.name, response, used_provider)
                results.append({"agent": agent.name, "content": response, "provider": used_provider})

                if on_message:
                    await on_message(agent, response, used_provider)

        self.logger.complete_session(log_path)
        self.running_task = None
        return results

    def stop_task(self):
        self.running_task = None

Your orchestrator is alive. Three agents can now discuss any task you throw at them, and every word is logged as structured markdown. Next up, you will wire these agents to Telegram so you can watch the conversation in real time from your phone.

Wire Up Telegram Integration

Your orchestrator can run multi-agent discussions and log every message. But right now, the only way to see what the agents are saying is by reading log files after the fact.

In this step, you will wire everything into Telegram so you can trigger tasks, watch agents talk in real time, and search conversation memory directly from your phone.

In this step, get ready to:
  • Build a command bot that responds to task, status, and stop commands.
  • Create a posting module that sends each agent's response from its own bot identity.
  • Test the full end-to-end flow with live agent discussion in your Telegram group.
Set up the bot module and command handlers

The python-telegram-bot library provides an Application class that handles all the polling and message routing for you. You just define handler functions for each command.

How does the bot architecture work?

You have 4 Telegram bots total. The command bot (SwarmCommander) receives your commands and orchestrates the flow. The 3 agent bots (Architect, Developer, Reviewer) only post messages to the group. They do not receive or process any commands.

Your Python script controls all 4 bots from a single process. The Application class manages the command bot's polling loop, while the agent bots are used as simple message senders via their Bot instances.

Before writing code, you need your group's chat ID. This tells the bots where to post agent messages.

  • Send any message in your Telegram group from your personal account.
  • Open this URL in your browser, replacing the token with your command bot's token:
https://api.telegram.org/botYOUR-COMMANDER-BOT-TOKEN/getUpdates
  • In the JSON response, look for the "chat":{"id":-100XXXXXXXXXX value. This negative number is your group chat ID.
  • Copy that number (including the minus sign) and update the group_chat_id field in your config.yaml.

Group chat IDs are always negative

Supergroup IDs typically start with -100 followed by a numeric ID. If you see a positive number, you are looking at a private chat ID instead.

  • Create a new file called bot.py in your project folder.
  • Add the imports and module-level setup by pasting this code:
import asyncio
import yaml
from telegram import Update, Bot
from telegram.ext import Application, CommandHandler, ContextTypes

from orchestrator import Orchestrator


# Initialize the orchestrator (loads config, connects to Gemini)
orchestrator = Orchestrator()

# Load config for Telegram-specific values
with open("config.yaml", "r") as f:
    config = yaml.safe_load(f)

# The group where agent messages will be posted
GROUP_CHAT_ID = config["telegram"]["group_chat_id"]

# Create a Bot instance for each agent so messages come from their identity
agent_bots = {}
for agent_cfg in config["agents"]:
    agent_bots[agent_cfg["name"]] = Bot(token=agent_cfg["token"])

What does this code do?

The module-level code runs once when the script starts. It creates an Orchestrator instance (which loads your config and connects to Gemini), reads the config for Telegram-specific values, and builds a dictionary mapping agent names to their Bot instances.

The agent_bots dictionary lets you post messages as any agent by looking up its bot object by name.

  • Below the setup code in bot.py, add the status, stop, and memory command handlers:
async def status_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if orchestrator.running_task:
        await update.message.reply_text(f"Running: {orchestrator.running_task}")
    else:
        await update.message.reply_text("No task running.")


async def stop_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    orchestrator.stop_task()
    await update.message.reply_text("Task stopped.")

What does this code do?

Each handler is an async function that receives the Telegram Update (the incoming message) and a context object. The /status handler checks orchestrator.running_task to report what is in progress. The /stop handler calls stop_task() which sets running_task to None, causing the orchestrator loop to break.

  • Next, add the memory search handler below the stop handler:
async def memory_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if not context.args:
        await update.message.reply_text("Usage: /memory <search query>")
        return

    query = " ".join(context.args)
    results = orchestrator.memory.search(query)

    if not results:
        await update.message.reply_text("No relevant memories found.")
        return

    response = "Memory search results:\n\n"
    for r in results:
        response += f"Session: {r['session']}\nScore: {r['score']}\n"
        response += f"Snippet: {r['snippet'][:200]}...\n\n"

    await update.message.reply_text(response[:4000])

What does this code do?

The /memory handler takes whatever words follow the command (via context.args) and searches your log library. It formats the results with session names, relevance scores, and snippet previews. The [:4000] slice prevents hitting Telegram's message length limit.

  • Save bot.py.

Seeing an import error?

If Python cannot find telegram, confirm you installed the library with pip install python-telegram-bot in an earlier step. Make sure you are using the same Python environment.

Help me troubleshoot this import error.

Create the Telegram posting module

The orchestrator already accepts an on_message callback that fires every time an agent produces a response. You need a function that takes that response and posts it to the Telegram group using the correct agent's bot token.

  • In bot.py, add the post_agent_message function directly below the agent_bots setup block (above status_command):
async def post_agent_message(agent, content, provider):
    bot = agent_bots.get(agent.name)
    if bot:
        # Append a failover tag if the response came from the backup provider
        tag = " [FAILOVER]" if provider != "gemini" else ""
        await bot.send_message(
            chat_id=GROUP_CHAT_ID,
            text=f"{content}{tag}",
        )
    # Pause between messages to avoid Telegram rate limits
    await asyncio.sleep(1)

What does this code do?

This function is the bridge between the orchestrator and Telegram. It looks up the correct bot for the speaking agent, appends a [FAILOVER] tag if the response came from Groq instead of Gemini, and sends the message to your group.

The asyncio.sleep(1) pause gives Telegram time between messages. Without it, rapid-fire posts might hit rate limits (20 messages per minute per bot in a group).

Why does each agent have its own bot?

When a message arrives from "AgentArchitect" in the group, it looks like a distinct participant speaking. This creates a natural multi-person conversation feel instead of one bot relaying everything.

The tradeoff is that bots cannot see messages from other bots in Telegram. That is fine here because the Python orchestrator manages all coordination server-side. The bots are purely for visual output.

  • Save bot.py.
Add the task command and run the full flow

The /task command is the main entry point. It acknowledges the request, posts a "task started" banner, runs the orchestrator with your posting callback, and posts a completion summary.

  • In bot.py, add the task_command function between post_agent_message and status_command:
async def task_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if not context.args:
        await update.message.reply_text("Usage: /task <description>")
        return

    task = " ".join(context.args)
    await update.message.reply_text(
        f"Starting task: {task}\nAgents: Architect, Developer, Reviewer"
    )

    # Post a visible banner to the group via the commander bot
    commander_bot = Bot(token=config["telegram"]["command_bot"]["token"])
    await commander_bot.send_message(
        chat_id=GROUP_CHAT_ID,
        text=f"--- NEW TASK ---\n{task}\n--- DISCUSSION START ---",
    )

    # Run the multi-agent discussion, posting each response to Telegram
    results = await orchestrator.run_task(task, on_message=post_agent_message)

    # Post a completion summary to the group
    await commander_bot.send_message(
        chat_id=GROUP_CHAT_ID,
        text=f"--- TASK COMPLETE ---\n{len(results)} messages exchanged.",
    )
    await update.message.reply_text(f"Task complete. {len(results)} messages logged.")

What does this code do?

The function first validates that the user provided a task description. It then acknowledges the command in the private chat and posts a visible "NEW TASK" banner to the group via the commander bot.

The key line is orchestrator.run_task(task, on_message=post_agent_message). This starts the multi-agent discussion and calls your posting function after every agent turn. When the discussion ends, the commander posts a "TASK COMPLETE" summary with the total message count.

Finally, add the main() function that registers all handlers and starts polling.

  • At the bottom of bot.py, add the main function and entry point:
def main():
    # Build the application using the command bot's token
    app = Application.builder().token(config["telegram"]["command_bot"]["token"]).build()

    # Register command handlers
    app.add_handler(CommandHandler("task", task_command))
    app.add_handler(CommandHandler("status", status_command))
    app.add_handler(CommandHandler("stop", stop_command))
    app.add_handler(CommandHandler("memory", memory_command))

    print("Bot started. Send /task in Telegram to begin.")
    app.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == "__main__":
    main()

What does this code do?

The Application.builder() pattern creates a bot application using your command bot's token. Each add_handler call maps a slash command to its handler function.

The run_polling method starts an infinite loop that checks Telegram for new messages. The allowed_updates=Update.ALL_TYPES ensures the bot receives all update types, not just text messages.

  • Save bot.py.
  • Start the bot by running:
python3 bot.py
  • You should see Bot started. Send /task in Telegram to begin. printed in your terminal.
  • In your Telegram group, send this command to the command bot:
/task Design a REST API for a todo app
  • Watch the group. You should see the commander bot post "--- NEW TASK ---", then each agent bot posts its response in sequence, and finally the commander posts "--- TASK COMPLETE ---" with the message count.

Bot not responding to commands?

Make sure you are sending the command in a direct message to the command bot, or that the command bot has privacy mode disabled (via /setprivacy in BotFather). If the bot is in a group but has privacy mode enabled, it only sees commands addressed to it (e.g. /task@YourBotName).

Also verify your group_chat_id in config.yaml is correct. It must be a negative number starting with -100 for supergroups.

Help me debug why my bot is not responding.

Seeing a Forbidden or Unauthorized error?

Double-check that all 4 bot tokens in config.yaml are correct. Also confirm all 4 bots are added to the group as administrators. Bots need admin rights to post in groups without restrictions.

Help me fix this authorization error.

✔️ Awesome, I've got everything!

Great. Double check you have saved bot.py and confirmed the end-to-end flow works.

ⓧ I'd like to double check the full code

import asyncio
import yaml
from telegram import Update, Bot
from telegram.ext import Application, CommandHandler, ContextTypes

from orchestrator import Orchestrator


orchestrator = Orchestrator()

with open("config.yaml", "r") as f:
    config = yaml.safe_load(f)

GROUP_CHAT_ID = config["telegram"]["group_chat_id"]

agent_bots = {}
for agent_cfg in config["agents"]:
    agent_bots[agent_cfg["name"]] = Bot(token=agent_cfg["token"])


async def post_agent_message(agent, content, provider):
    bot = agent_bots.get(agent.name)
    if bot:
        tag = " [FAILOVER]" if provider != "gemini" else ""
        await bot.send_message(
            chat_id=GROUP_CHAT_ID,
            text=f"{content}{tag}",
        )
    await asyncio.sleep(1)


async def task_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if not context.args:
        await update.message.reply_text("Usage: /task <description>")
        return

    task = " ".join(context.args)
    await update.message.reply_text(
        f"Starting task: {task}\nAgents: Architect, Developer, Reviewer"
    )

    commander_bot = Bot(token=config["telegram"]["command_bot"]["token"])
    await commander_bot.send_message(
        chat_id=GROUP_CHAT_ID,
        text=f"--- NEW TASK ---\n{task}\n--- DISCUSSION START ---",
    )

    results = await orchestrator.run_task(task, on_message=post_agent_message)

    await commander_bot.send_message(
        chat_id=GROUP_CHAT_ID,
        text=f"--- TASK COMPLETE ---\n{len(results)} messages exchanged.",
    )
    await update.message.reply_text(f"Task complete. {len(results)} messages logged.")


async def status_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if orchestrator.running_task:
        await update.message.reply_text(f"Running: {orchestrator.running_task}")
    else:
        await update.message.reply_text("No task running.")


async def stop_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    orchestrator.stop_task()
    await update.message.reply_text("Task stopped.")


async def memory_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if not context.args:
        await update.message.reply_text("Usage: /memory <search query>")
        return

    query = " ".join(context.args)
    results = orchestrator.memory.search(query)

    if not results:
        await update.message.reply_text("No relevant memories found.")
        return

    response = "Memory search results:\n\n"
    for r in results:
        response += f"Session: {r['session']}\nScore: {r['score']}\n"
        response += f"Snippet: {r['snippet'][:200]}...\n\n"

    await update.message.reply_text(response[:4000])


def main():
    app = Application.builder().token(config["telegram"]["command_bot"]["token"]).build()

    app.add_handler(CommandHandler("task", task_command))
    app.add_handler(CommandHandler("status", status_command))
    app.add_handler(CommandHandler("stop", stop_command))
    app.add_handler(CommandHandler("memory", memory_command))

    print("Bot started. Send /task in Telegram to begin.")
    app.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == "__main__":
    main()

Your swarm is now live on Telegram. Next up, you will build the structured logging and memory system so your agents can remember past conversations.

Build the Markdown Log Library and Memory

Your agent swarm can discuss tasks and post to Telegram, but every conversation disappears the moment it finishes. Without a record of past discussions, agents repeat themselves and lose valuable context between sessions.

Your project already includes two modules designed to solve this: logger.py saves every conversation as structured markdown, and memory.py searches those logs to inject relevant history into future agent turns. In this step, you'll validate both systems end-to-end.

In this step, get ready to:
  • Confirm that running a task produces a correctly structured markdown log file with YAML frontmatter.
  • Validate keyword-based memory search across multiple completed sessions.
  • Verify that agents automatically recall past discussions and that the /memory command works in Telegram.
Validate the structured logging system

Every time the orchestrator runs a task, logger.py automatically saves the full conversation as a timestamped markdown file. The file lands in a directory tree structured as logs/sessions/YYYY-MM-DD_task-slug/conversation.md.

What does logger.py do?

The ConversationLogger class handles three things:

  • create_session() creates a dated directory and writes YAML frontmatter (task name, agents, categories, timestamp, and status) to a new conversation.md file.
  • log_message() appends each agent response with a timestamp, agent name, and provider tag.
  • complete_session() flips the frontmatter status from in-progress to completed.
  • Switch to your Telegram group and trigger a test task by sending this message to the command bot:
/task Design a REST API for a todo app
  • Wait for all three agents to finish responding and for the completion summary to appear in the group.

Once the task completes, a new log file should exist on disk.

  • Check the log directory structure by running this command:
ls logs/sessions/
  • You should see a folder named something like 2026-06-09_design-a-rest-api-for-a-todo-app (the date will match today and the task is slugified).
  • View the log contents by running:
cat logs/sessions/*/conversation.md

You should see a file that starts with YAML frontmatter between --- markers, followed by timestamped agent messages separated by horizontal rules.

What should the log file look like?

The file starts with frontmatter containing the task, agent names, categories, start time, and a status: completed field. Below that, each agent message appears as a level-2 heading with a timestamp and agent name, the response body, and a provider tag (e.g. _provider: gemini_).

No logs directory or empty file?

  • Make sure the task actually completed in Telegram. If the bot responded with an error, the session may not have been created.
  • Check that log_dir in config.yaml is set to "logs" under the orchestrator section.
  • If the file exists but shows status: in-progress, the task may have been interrupted before complete_session() ran. Run another task and let it finish fully.

Help me debug why my log files aren't being created

Test memory retrieval across multiple sessions

For memory retrieval to be useful, the system needs multiple sessions to search across. The MemoryRetriever class in memory.py uses keyword-based search with a scoring system: matches in the task title score 10 points per term, while matches in the conversation body score 1 point per occurrence.

Why keyword search instead of embeddings?

A vector database with embeddings would offer semantic search, but it adds infrastructure cost and complexity. Keyword search over structured markdown is free, runs locally, and works well for the volume of conversations you'll generate.

The markdown structure is "RAG-ready." You can bolt on embeddings later without restructuring any of the log files.

  • Run two more tasks in Telegram to build up your log history. Send each one separately and wait for it to complete:
/task Design authentication for a REST API
  • Wait for that task to complete, then send a second task:
/task Build a notification microservice with webhooks
  • Wait for that task to complete as well.
  • Confirm you now have multiple session folders by running:
ls logs/sessions/
  • You should see 3 dated folders (one per task you've run so far).

Now test the memory search directly. Open a Python shell to query the MemoryRetriever.

  • Test memory search by running this script:
python3 -c "from memory import MemoryRetriever; m = MemoryRetriever('logs'); results = m.search('REST API'); print(f'Found {len(results)} results'); [print(f\"  {r['session']} (score: {r['score']})\") for r in results]"

What does this test?

This creates a MemoryRetriever pointing at your logs directory and searches for "REST API". The results should show the REST API task scoring highest (title match = 10 points per term), with the authentication task scoring next (it mentions REST API in the conversation body).

  • You should see 2 or more results with the REST API task ranked first by score.

Getting 0 results?

  • Run ls logs/sessions/ to confirm session folders exist. If the folder is empty, your tasks may not have completed successfully.
  • Check that each session folder contains a conversation.md file with actual content (not just frontmatter).
  • Make sure your search term appears somewhere in the logs. Try a broader term like "API" if "REST API" returns nothing.

Help me debug why memory search returns empty results

Confirm memory integration and the /memory command

The real power of the memory system is automatic context injection. Before each agent discussion starts, the orchestrator calls get_context_for_agent() to find relevant past sessions and prepends them to the conversation. Agents can then reference earlier discussions without being explicitly told about them.

How does memory injection work?

In orchestrator.py, the run_task() method calls self.memory.get_context_for_agent(task) before starting the agent loop. If relevant past conversations exist, they're added as a [CONTEXT FROM PAST SESSIONS] message at the start of the conversation history.

This works because Gemini 2.5 Flash has a 1M token context window. Injecting a few hundred tokens of past conversation snippets is trivial within that budget.

  • In Telegram, send a task that directly relates to your earlier REST API task:
/task Add rate limiting to the REST API we designed earlier
  • Watch the agent responses in the Telegram group as they come in.
  • You should see agents referencing details from the earlier REST API discussion (like endpoint names, authentication approach, or architectural decisions they made before).

What if agents don't reference past context?

Memory injection depends on the search returning relevant results. If no prior session matches the new task's keywords, the context will be empty and agents won't have past conversations to reference. Make sure your new task shares keywords with an earlier one (like "REST API" in this case).

Finally, test the manual memory search command from Telegram.

  • Send the following command in Telegram to search your logs:
/memory authentication
  • The command bot should reply with memory search results showing session names, scores, and text snippets from conversations that mention authentication.

Getting "No relevant memories found"?

  • Try a different keyword that definitely appears in your earlier tasks. Use a single word that you know agents discussed (like "API" or "todo").
  • If /memory doesn't respond at all, check that the memory command handler is registered in bot.py. Look for CommandHandler("memory", memory_command) in the main() function.

Help me troubleshoot the /memory Telegram command

Let's take a look at the full code for the two key files in this step.

✔️ Awesome, I've got everything!

Great. Double check you've saved both logger.py and memory.py and that your log files are being generated correctly.

ⓧ I'd like to double check the full code

Here is the full logger.py:

import os
import re
from datetime import datetime


class ConversationLogger:
    def __init__(self, log_dir="logs"):
        self.log_dir = log_dir
        self.sessions_dir = os.path.join(log_dir, "sessions")
        os.makedirs(self.sessions_dir, exist_ok=True)

    def _slugify(self, text):
        slug = re.sub(r"[^\w\s-]", "", text.lower())
        slug = re.sub(r"[\s_]+", "-", slug)
        return slug[:50]

    def create_session(self, task, agents, categories=None):
        date_str = datetime.now().strftime("%Y-%m-%d")
        slug = self._slugify(task)
        session_dir = os.path.join(self.sessions_dir, f"{date_str}_{slug}")
        os.makedirs(session_dir, exist_ok=True)

        filepath = os.path.join(session_dir, "conversation.md")
        frontmatter = (
            f"---\n"
            f"task: \"{task}\"\n"
            f"agents: [{', '.join(agents)}]\n"
            f"categories: [{', '.join(categories or ['general'])}]\n"
            f"started: \"{datetime.now().isoformat()}\"\n"
            f"status: in-progress\n"
            f"---\n\n"
            f"# {task}\n\n"
        )

        with open(filepath, "w") as f:
            f.write(frontmatter)

        return filepath

    def log_message(self, filepath, agent_name, content, provider="gemini"):
        timestamp = datetime.now().strftime("%H:%M:%S")
        entry = (
            f"## [{timestamp}] {agent_name}\n\n"
            f"{content}\n\n"
            f"_provider: {provider}_\n\n"
            f"---\n\n"
        )

        with open(filepath, "a") as f:
            f.write(entry)

    def complete_session(self, filepath):
        with open(filepath, "r") as f:
            content = f.read()

        content = content.replace("status: in-progress", "status: completed")

        with open(filepath, "w") as f:
            f.write(content)

And here is the full memory.py:

import os
import re
from datetime import datetime


class MemoryRetriever:
    def __init__(self, log_dir="logs"):
        self.sessions_dir = os.path.join(log_dir, "sessions")

    def search(self, query, max_results=3, category=None):
        results = []
        query_terms = query.lower().split()

        if not os.path.exists(self.sessions_dir):
            return results

        for session_name in sorted(os.listdir(self.sessions_dir), reverse=True):
            session_path = os.path.join(
                self.sessions_dir, session_name, "conversation.md"
            )
            if not os.path.exists(session_path):
                continue

            with open(session_path, "r") as f:
                content = f.read()

            if category:
                cat_match = re.search(r"categories:\s*\[(.+?)\]", content)
                if cat_match and category.lower() not in cat_match.group(1).lower():
                    continue

            score = 0
            content_lower = content.lower()

            task_match = re.search(r'task:\s*"(.+?)"', content)
            task_text = task_match.group(1).lower() if task_match else ""

            for term in query_terms:
                if term in task_text:
                    score += 10
                score += content_lower.count(term)

            if score > 0:
                snippet = self._extract_snippet(content, query_terms)
                results.append(
                    {"session": session_name, "score": score, "snippet": snippet}
                )

        results.sort(key=lambda x: x["score"], reverse=True)
        return results[:max_results]

    def _extract_snippet(self, content, terms, max_length=500):
        lines = content.split("\n")
        relevant_lines = []

        for i, line in enumerate(lines):
            if any(term in line.lower() for term in terms):
                start = max(0, i - 1)
                end = min(len(lines), i + 3)
                relevant_lines.extend(lines[start:end])
                if len("\n".join(relevant_lines)) > max_length:
                    break

        snippet = "\n".join(relevant_lines)[:max_length]
        return snippet if snippet else content[200:700]

    def get_context_for_agent(self, task_description, max_snippets=2):
        results = self.search(task_description, max_results=max_snippets)
        if not results:
            return ""

        context = "## Relevant past discussions:\n\n"
        for r in results:
            context += f"### From session: {r['session']}\n"
            context += f"{r['snippet']}\n\n"

        return context

Your swarm now has persistent memory. Every conversation is logged as structured markdown, and agents automatically recall relevant past discussions when tackling new tasks. Next up, you'll put the whole system through its paces with a final integration test.

Secret mission

Multi-Provider Failover

Your swarm works great until Gemini's 15 RPM rate limit kicks in mid-conversation. In this secret mission, add Groq as a backup LLM provider that activates automatically when Gemini hits its limit. The system detects 429 responses, switches to Groq's free tier, then switches back once the cooldown expires.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally and uses free-tier APIs, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • 4 Telegram bots (SwarmCommander, Architect, Developer, Reviewer).
  • Google Gemini API key (free tier, 1,500 requests/day).
  • Groq API key (free tier, if you completed the Secret Mission).
  • Local project files and logs/ directory.
  • Python packages (openai, python-telegram-bot, pyyaml).

Keep everything running

No action needed. Choose this if you are still actively building on the swarm or planning to extend it.

  • Your Telegram bots remain active and ready for commands.
  • Your API keys stay valid within their free-tier limits.
  • Your log files persist in the logs/sessions/ directory for memory retrieval.
  • To keep the bot running while you close your terminal, use nohup python3 bot.py & so it stays alive in the background.

Pause - I'll come back to this later

Stop the running bot process to free up system resources. Your files, logs, and API keys all persist for when you return.

  • If the bot is running in your terminal, press Ctrl+C to stop it.
  • If you started it in the background with nohup, find the process ID and stop it by running:
ps aux | grep bot.py
kill <PID>
  • Your log files and configuration remain intact. Resume anytime by running python3 bot.py again.

Delete - I don't want to use this again

Remove all project resources and start fresh. Follow each step below to fully clean up.

  • Delete your Telegram bots. Open Telegram, message @BotFather, and send /deletebot. Select each bot (SwarmCommander, AgentArchitect, AgentDeveloper, AgentReviewer) and confirm deletion. Repeat for all four bots.
  • Revoke your Gemini API key. Go to aistudio.google.com/api-keys in your browser. Find your key in the list and delete it. If the key does not appear there, visit the Google Cloud Console credentials page for the same project to delete it.
  • Revoke your Groq API key (if you completed the Secret Mission). Go to console.groq.com/keys in your browser. Find your key in the list and delete it.
  • Uninstall the Python packages by running:
pip uninstall openai python-telegram-bot pyyaml -y
  • Delete your local project files by running:
cd ~/Desktop
rm -rf agent-swarm

What if my folder has a different name?

Replace agent-swarm with whatever you named your project folder when you created it in Step 1. You can run ls ~/Desktop to confirm the name before deleting.

Nice Work!

Nice Work!

You just built a fully autonomous multi-agent swarm that collaborates on tasks, posts live to Telegram, and remembers everything across sessions.

You've learned how to:

  • Build a multi-agent orchestrator managing 3 specialized AI agents (Architect, Developer, Reviewer) powered by free-tier Gemini 2.5 Flash through the OpenAI-compatible API endpoint.
  • Connect 4 Telegram bots to create a live dashboard for triggering tasks, monitoring agent collaboration, and searching conversation memory with commands like /task and /memory.
  • Implement a structured markdown log library with YAML frontmatter, timestamped session records, and keyword-based memory retrieval so agents automatically recall and reference past discussions.
  • Secret Mission: Implemented Groq failover with automatic provider switching when Gemini rate limits are hit, keeping your swarm running without interruption.

Ready to quiz yourself?