Learning Hub - Free AI Education | QuantaSkool

Architecture Overview

ClawdBot goes far beyond simple chatbots. It features a modular architecture with three core systems:

┌─────────────────────────────────────────────────────────────┐
│                        User Message                          │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                    Telegram Bot Handler                      │
│  • Receives messages via python-telegram-bot                 │
│  • Routes commands (/start, /help, /tasks)                  │
│  • Sends responses back to users                             │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                       AI Agent Core                          │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐│
│  │   Memory    │ │     RAG     │ │      Tool Executor      ││
│  │   (mem0)    │ │  (Vector)   │ │  (Telegram, MCP, etc)   ││
│  └─────────────┘ └─────────────┘ └─────────────────────────┘│
│                        │                                     │
│                        ▼                                     │
│              ┌─────────────────────┐                        │
│              │   OpenAI GPT-4o    │                        │
│              │   (with tools)      │                        │
│              └─────────────────────┘                        │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                    Data Persistence                          │
│  • SQLite: Sessions, Messages, Tasks                         │
│  • Vector Store: Embeddings for semantic search              │
│  • mem0.ai: Long-term user memories                          │
└─────────────────────────────────────────────────────────────┘

Prerequisites

Python 3.10+ installed

Telegram Bot Token (from @BotFather)

OpenAI API Key

(Optional) mem0.ai API Key

Project Structure

telegram-clawdbot/
├── src/
│   ├── main.py              # Entry point
│   ├── config.py            # Configuration management
│   ├── agents/
│   │   └── agent.py         # Core AI logic with tool calling
│   ├── bot/
│   │   ├── handlers.py      # Telegram command handlers
│   │   └── telegram_bot.py  # Bot initialization
│   ├── rag/
│   │   ├── embeddings.py    # OpenAI embeddings
│   │   ├── vectorstore.py   # Vector database
│   │   └── retriever.py     # Semantic search
│   ├── memory/
│   │   ├── database.py      # SQLite for sessions/messages
│   │   └── mem0_client.py   # Long-term memory
│   └── tools/
│       ├── scheduler.py     # Reminders and tasks
│       └── telegram_actions.py
├── requirements.txt
└── .env

Step-by-Step Implementation

Set Up the Project

mkdir telegram-clawdbot && cd telegram-clawdbot
python -m venv venv
source venv/bin/activate

pip install python-telegram-bot openai mem0ai pydantic \
            pydantic-settings python-dotenv apscheduler

Create Configuration

Create a .env file with your API keys:

# Required
TELEGRAM_BOT_TOKEN=your_bot_token_here
OPENAI_API_KEY=your_openai_key_here

# Optional - Memory
MEM0_API_KEY=your_mem0_key_here
MEMORY_ENABLED=true

# Optional - RAG
RAG_ENABLED=true

Build Configuration System

Use Pydantic for type-safe configuration in src/config.py:

from pydantic_settings import BaseSettings
from pydantic import Field

class TelegramSettings(BaseSettings):
    bot_token: str = Field(alias="TELEGRAM_BOT_TOKEN")

class AISettings(BaseSettings):
    openai_api_key: str = Field(alias="OPENAI_API_KEY")
    model: str = Field(default="gpt-4o", alias="AI_MODEL")

class Settings(BaseSettings):
    telegram: TelegramSettings = TelegramSettings()
    ai: AISettings = AISettings()
    
    class Config:
        env_file = ".env"
        extra = "ignore"

Implement Database Layer

SQLite stores sessions, messages, and scheduled tasks in src/memory/database.py:

import sqlite3
from dataclasses import dataclass
from datetime import datetime

@dataclass
class Session:
    id: str
    user_id: int
    chat_id: int
    chat_type: str
    created_at: datetime

def init_database():
    conn = sqlite3.connect("data/clawdbot.db")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS sessions (
            id TEXT PRIMARY KEY,
            user_id INTEGER NOT NULL,
            chat_id INTEGER NOT NULL,
            chat_type TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)

Build RAG System

Enable semantic search over conversation history in src/rag/embeddings.py:

from openai import OpenAI
import numpy as np

def create_embedding(text: str) -> list[float]:
    client = OpenAI()
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

def cosine_similarity(a: list[float], b: list[float]) -> float:
    a, b = np.array(a), np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

Implement Long-Term Memory

Use mem0.ai for persistent user memories in src/memory/mem0_client.py:

from mem0 import MemoryClient

_client: MemoryClient | None = None

def initialize_memory(api_key: str):
    global _client
    _client = MemoryClient(api_key=api_key)

async def add_memory(messages: list[dict], user_id: int):
    if _client:
        _client.add(messages, user_id=str(user_id))

async def search_memory(query: str, user_id: int, limit: int = 5):
    if _client is None:
        return []
    results = _client.search(query, user_id=str(user_id), limit=limit)
    return results.get("results", [])

Create AI Agent Core

The agent orchestrates Memory, RAG, and Tools in src/agents/agent.py:

from openai import OpenAI

SYSTEM_PROMPT = """You are a helpful AI assistant in Telegram.
## CAPABILITIES:
- Answer questions and have conversations
- Search knowledge base for relevant context
- Remember user preferences across sessions
- Use external tools like GitHub and Notion
"""

async def process_message(user_message: str, context):
    client = OpenAI()
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    
    # 1. Retrieve memories
    memories = await search_memory(user_message, context.user_id)
    
    # 2. Retrieve RAG context
    rag_results = await retrieve(user_message, chat_id=context.chat_id)
    
    # 3. Call OpenAI with tools
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=get_all_tools(),
        tool_choice="auto"
    )
    
    return response.choices[0].message.content

Build Telegram Handlers & Run

Wire it all together in src/main.py:

import asyncio
from telegram.ext import Application
from dotenv import load_dotenv

load_dotenv()

async def main():
    config = get_config()
    init_database()
    
    if config.memory.api_key:
        await initialize_memory(config.memory.api_key)
    
    application = Application.builder() \
        .token(config.telegram.bot_token) \
        .build()
    
    setup_handlers(application)
    
    print("🚀 ClawdBot is running!")
    await application.run_polling()

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

Key Takeaways

🔹Modular Architecture — Each component (RAG, Memory, Tools) is isolated and testable
🔹Async Everything — Non-blocking I/O for responsive user experience
🔹Tool Calling Loop — Let the LLM decide which tools to use, execute, and continue
🔹Memory Layers — Combine short-term (SQLite) with long-term (mem0) memory

Next Steps

Add more tools: Weather, calendar, web search

Implement webhooks for production deployment

Add role-based authentication for team bots

Monitor & log usage, errors, and performance

How to Build ClawdBot