Agent در عمل — ساخت یه دستیار تلگرامی

مقدمه: بیا یه چیز واقعی بسازیم

تا الان خیلی تئوری حرف زدیم. حالا وقتشه دست به کد بشیم و یه Agent واقعی بسازیم که آدم‌ها واقعاً ازش استفاده کنن.

تو این اپیزود، یه دستیار هوشمند تلگرامی می‌سازیم. نه یه چت‌بات ساده — بلکه یه Agent واقعی که ابزار داره، حافظه داره و می‌تونه کارای مفید انجام بده. قدم به قدم، از صفر تا دیپلوی.

چی می‌سازیم؟

یه دستیار تلگرامی که:

• مکالمه طبیعی به فارسی داره
• می‌تونه آب‌وهوا رو چک کنه
• می‌تونه محاسبات ریاضی انجام بده
• حافظه داره و مکالمات قبلی رو یادش می‌مونه
• تو گروه و چت خصوصی کار می‌کنه

قدم ۱: نصب و راه‌اندازی

# نصب کتابخانه‌ها
# pip install python-telegram-bot openai redis

# ساختار پروژه:
# telegram_agent/
# ├── bot.py          (فایل اصلی)
# ├── agent.py        (منطق Agent)
# ├── tools.py        (ابزارها)
# ├── memory.py       (حافظه)
# ├── config.py       (تنظیمات)
# └── requirements.txt

قدم ۲: تنظیمات

# config.py
import os

TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
WEATHER_API_KEY = os.getenv("WEATHER_API_KEY")  # OpenWeatherMap
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")

# مدل
MODEL_NAME = "gpt-4o-mini"
MAX_TOKENS = 1000

# محدودیت‌ها
MAX_MESSAGES_PER_MINUTE = 10
MAX_HISTORY_LENGTH = 20

قدم ۳: ابزارها

# tools.py
import json
import httpx
from config import WEATHER_API_KEY

# تعریف ابزارها برای OpenAI
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "آب‌وهوای فعلی یه شهر رو می‌ده",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "نام شهر (انگلیسی)"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "عبارت ریاضی رو محاسبه می‌کنه",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "عبارت ریاضی"
                    }
                },
                "required": ["expression"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "تاریخ و ساعت فعلی رو برمی‌گردونه",
            "parameters": {
                "type": "object",
                "properties": {
                    "timezone": {
                        "type": "string",
                        "description": "منطقه زمانی",
                        "default": "Asia/Tehran"
                    }
                }
            }
        }
    },
]

# پیاده‌سازی ابزارها
async def get_weather(city: str) -> str:
    """آب‌وهوای شهر رو از API می‌گیره."""
    url = "https://api.openweathermap.org/data/2.5/weather"
    params = {
        "q": city,
        "appid": WEATHER_API_KEY,
        "units": "metric",
        "lang": "fa",
    }
    
    async with httpx.AsyncClient() as client:
        try:
            resp = await client.get(url, params=params)
            data = resp.json()
            
            if resp.status_code != 200:
                return f"خطا: شهر '{city}' پیدا نشد."
            
            temp = data["main"]["temp"]
            feels = data["main"]["feels_like"]
            desc = data["weather"][0]["description"]
            humidity = data["main"]["humidity"]
            
            return (
                f"شهر: {city}\n"
                f"دما: {temp}°C (حس واقعی: {feels}°C)\n"
                f"وضعیت: {desc}\n"
                f"رطوبت: {humidity}%"
            )
        except Exception as e:
            return f"خطا در دریافت آب‌وهوا: {str(e)}"


def calculate(expression: str) -> str:
    """عبارت ریاضی رو محاسبه می‌کنه (امن)."""
    # فقط کاراکترهای ریاضی مجازن
    allowed = set("0123456789+-*/().% ")
    if not all(c in allowed for c in expression):
        return "خطا: فقط عبارات ریاضی ساده مجازه."
    
    try:
        result = eval(expression)
        return f"{expression} = {result}"
    except Exception as e:
        return f"خطا در محاسبه: {str(e)}"


def get_current_time(timezone: str = "Asia/Tehran") -> str:
    """تاریخ و ساعت فعلی."""
    from datetime import datetime
    import pytz
    
    try:
        tz = pytz.timezone(timezone)
        now = datetime.now(tz)
        return now.strftime("%Y-%m-%d %H:%M:%S %Z")
    except:
        return "منطقه زمانی نامعتبره."


# مپ نام ابزار به تابع
TOOL_FUNCTIONS = {
    "get_weather": get_weather,
    "calculate": calculate,
    "get_current_time": get_current_time,
}

async def execute_tool(name: str, args: dict) -> str:
    """ابزار رو اجرا می‌کنه."""
    func = TOOL_FUNCTIONS.get(name)
    if not func:
        return f"ابزار '{name}' وجود نداره."
    
    import asyncio
    if asyncio.iscoroutinefunction(func):
        return await func(**args)
    return func(**args)

قدم ۴: حافظه

# memory.py
import json
import redis.asyncio as redis
from config import REDIS_URL, MAX_HISTORY_LENGTH

class Memory:
    """حافظه مکالمه با Redis."""
    
    def __init__(self):
        self.redis = redis.from_url(REDIS_URL)
    
    def _key(self, chat_id: int) -> str:
        return f"chat:{chat_id}:history"
    
    async def get_history(self, chat_id: int) -> list:
        """تاریخچه مکالمه رو برمی‌گردونه."""
        data = await self.redis.get(self._key(chat_id))
        if data:
            return json.loads(data)
        return []
    
    async def add_message(
        self, chat_id: int, role: str, content: str
    ):
        """یه پیام به تاریخچه اضافه می‌کنه."""
        history = await self.get_history(chat_id)
        history.append({"role": role, "content": content})
        
        # فقط آخرین N پیام رو نگه دار
        if len(history) > MAX_HISTORY_LENGTH:
            history = history[-MAX_HISTORY_LENGTH:]
        
        await self.redis.set(
            self._key(chat_id),
            json.dumps(history, ensure_ascii=False),
            ex=86400 * 7,  # ۷ روز نگه‌داری
        )
    
    async def clear(self, chat_id: int):
        """تاریخچه رو پاک می‌کنه."""
        await self.redis.delete(self._key(chat_id))

قدم ۵: منطق Agent

# agent.py
from openai import AsyncOpenAI
from config import OPENAI_API_KEY, MODEL_NAME, MAX_TOKENS
from tools import TOOLS, execute_tool
from memory import Memory

client = AsyncOpenAI(api_key=OPENAI_API_KEY)
memory = Memory()

SYSTEM_PROMPT = """تو یه دستیار هوشمند فارسی‌زبان هستی.

قوانین:
- همیشه فارسی جواب بده
- مختصر و مفید باش
- اگه نمی‌دونی، بگو نمی‌دونم
- از ابزارها برای کارای واقعی استفاده کن
- هیچ‌وقت اطلاعات جعلی نده

ابزارهای تو:
- get_weather: آب‌وهوای شهر
- calculate: ماشین‌حساب
- get_current_time: تاریخ و ساعت
"""


async def process_message(
    chat_id: int, user_message: str, user_name: str
) -> str:
    """پیام کاربر رو پردازش می‌کنه و جواب می‌ده."""
    
    # تاریخچه مکالمه
    history = await memory.get_history(chat_id)
    
    # ساخت لیست پیام‌ها
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    messages.extend(history)
    messages.append({"role": "user", "content": user_message})
    
    # فراخوانی LLM
    response = await client.chat.completions.create(
        model=MODEL_NAME,
        messages=messages,
        tools=TOOLS,
        max_tokens=MAX_TOKENS,
    )
    
    msg = response.choices[0].message
    
    # اگه ابزار می‌خواد استفاده کنه
    if msg.tool_calls:
        # اجرای همه ابزارها
        tool_results = []
        for tool_call in msg.tool_calls:
            import json
            args = json.loads(tool_call.function.arguments)
            result = await execute_tool(
                tool_call.function.name, args
            )
            tool_results.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            })
        
        # فراخوانی دوباره LLM با نتایج ابزار
        messages.append({
            "role": "assistant",
            "content": msg.content,
            "tool_calls": [
                {
                    "id": tc.id,
                    "type": "function",
                    "function": {
                        "name": tc.function.name,
                        "arguments": tc.function.arguments,
                    }
                }
                for tc in msg.tool_calls
            ]
        })
        messages.extend(tool_results)
        
        response = await client.chat.completions.create(
            model=MODEL_NAME,
            messages=messages,
            max_tokens=MAX_TOKENS,
        )
        final_content = response.choices[0].message.content
    else:
        final_content = msg.content
    
    # ذخیره تو حافظه
    await memory.add_message(chat_id, "user", user_message)
    await memory.add_message(chat_id, "assistant", final_content)
    
    return final_content

قدم ۶: بات تلگرام

# bot.py
import logging
from telegram import Update
from telegram.ext import (
    Application,
    CommandHandler,
    MessageHandler,
    filters,
    ContextTypes,
)
from config import TELEGRAM_TOKEN
from agent import process_message
from memory import Memory

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

memory = Memory()


async def start(
    update: Update, context: ContextTypes.DEFAULT_TYPE
):
    """دستور /start"""
    await update.message.reply_text(
        "سلام! من دستیار هوشمند هستم.\n\n"
        "می‌تونم:\n"
        "- سوالاتت رو جواب بدم\n"
        "- آب‌وهوا رو بگم\n"
        "- محاسبات ریاضی انجام بدم\n\n"
        "هر چیزی بپرس!"
    )


async def clear(
    update: Update, context: ContextTypes.DEFAULT_TYPE
):
    """دستور /clear - پاک کردن حافظه"""
    await memory.clear(update.effective_chat.id)
    await update.message.reply_text(
        "حافظه پاک شد! از اول شروع می‌کنیم."
    )


async def handle_message(
    update: Update, context: ContextTypes.DEFAULT_TYPE
):
    """پردازش پیام‌های متنی."""
    if not update.message or not update.message.text:
        return
    
    chat_id = update.effective_chat.id
    user_message = update.message.text
    user_name = update.effective_user.first_name
    
    # تو گروه فقط وقتی mention بشه جواب بده
    if update.effective_chat.type in ["group", "supergroup"]:
        bot_username = context.bot.username
        if f"@{bot_username}" not in user_message:
            return
        # mention رو حذف کن
        user_message = user_message.replace(
            f"@{bot_username}", ""
        ).strip()
    
    # نشون بده داره تایپ می‌کنه
    await context.bot.send_chat_action(
        chat_id=chat_id, action="typing"
    )
    
    try:
        response = await process_message(
            chat_id, user_message, user_name
        )
        await update.message.reply_text(response)
    except Exception as e:
        logger.error(f"Error: {e}")
        await update.message.reply_text(
            "متأسفم، یه مشکلی پیش اومد. دوباره تلاش کن."
        )


def main():
    """راه‌اندازی بات."""
    app = Application.builder().token(TELEGRAM_TOKEN).build()
    
    # دستورات
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("clear", clear))
    
    # پیام‌های متنی
    app.add_handler(
        MessageHandler(
            filters.TEXT & ~filters.COMMAND,
            handle_message,
        )
    )
    
    logger.info("Bot is running...")
    app.run_polling()


if __name__ == "__main__":
    main()

قدم ۷: مدیریت گروه و خصوصی

یه نکته مهم: رفتار بات تو گروه باید فرق داشته باشه با چت خصوصی.

چت خصوصی: به هر پیامی جواب بده. حافظه مختص همون کاربره.

گروه: فقط وقتی mention بشه (@bot_name) جواب بده. حافظه مشترک بین همه اعضای گروهه. این رو تو کد بالا پیاده‌سازی کردیم — توجه کن به بخش handle_message که چک می‌کنه آیا تو گروه هستیم و آیا بات mention شده.

قدم ۸: دیپلوی

# Dockerfile
# FROM python:3.11-slim
# WORKDIR /app
# COPY requirements.txt .
# RUN pip install -r requirements.txt
# COPY . .
# CMD ["python", "bot.py"]

# requirements.txt
# python-telegram-bot==21.0
# openai==1.30.0
# redis==5.0.0
# httpx==0.27.0
# pytz==2024.1

# docker-compose.yml
# version: "3.8"
# services:
#   bot:
#     build: .
#     env_file: .env
#     depends_on:
#       - redis
#     restart: always
#   redis:
#     image: redis:7-alpine
#     volumes:
#       - redis_data:/data
# volumes:
#   redis_data:

# اجرا:
# docker-compose up -d

بهبودهای پیشنهادی

این نسخه پایه‌ست. چند ایده برای بهترش کردن:

• Rate limiting: جلوی اسپم رو بگیر — حداکثر ۱۰ پیام در دقیقه از هر کاربر
• ابزارهای بیشتر: جستجوی وب، ترجمه، خلاصه‌سازی لینک
• پشتیبانی تصویر: با مدل‌های Vision، تصاویر رو هم تحلیل کن
• دستور /help: راهنمای کامل قابلیت‌ها
• لاگ و مانیتورینگ: ثبت همه مکالمات برای بهبود
• Guardrails: از اپیزود قبلی، لایه‌های امنیتی اضافه کن

جمع‌بندی

• ساخت Agent تلگرامی ترکیب python-telegram-bot + OpenAI API + Redis ه
• ابزارها (Tools) قدرت واقعی Agent رو نشون می‌دن
• حافظه (Memory) با Redis پایدار و سریعه
• مدیریت گروه و خصوصی مهمه
• دیپلوی با Docker ساده‌ترینه

اپیزود بعدی: تست و دیباگ Agent ها — چطور بفهمی Agent ت درست کار می‌کنه.