คุณกำลังพัฒนา Telegram Bot ที่ใช้ AI ตอบกลับลูกค้าอัตโนมัติใช่ไหม? วันนี้ผมจะพาคุณสร้าง Bot ที่ใช้งานได้จริง พร้อมวิธีแก้ไขข้อผิดพลาดที่ผมเจอมาด้วยตัวเอง

เหตุการณ์จริง: ConnectionError ที่ทำให้ Bot หยุดทำงาน

ผมเคยเจอปัญหา ConnectionError: timeout after 30 seconds ตอนที่ Bot กำลังรับข้อความจากผู้ใช้จำนวนมากพร้อมกัน เหตุผลคือ Server ของ OpenAI ตอบสนองช้าในช่วง Peak Hour ทำให้ Telegram ส่ง Error 401 Unauthorized กลับมาเพราะ Webhook Timeout

หลังจากลองใช้ HolySheep AI ราคาถูกกว่า 85% ความหน่วงต่ำกว่า 50ms ใช้งานได้ราบรื่นมาก บทความนี้จะสอนคุณตั้งแต่ต้นจนได้ Bot ที่ใช้งานได้จริง

เตรียม Environment และติดตั้ง Dependencies

# สร้าง virtual environment
python -m venv telegram-ai-bot
source telegram-ai-bot/bin/activate  # Windows: telegram-ai-bot\Scripts\activate


ติดตั้ง packages ที่จำเป็น

pip install python-telegram-bot requests python-dotenv aiohttp

โค้ดหลัก: Telegram Bot พร้อม AI Reply

import os
import logging
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
import requests
from dotenv import load_dotenv

load_dotenv()


ตั้งค่า Configuration

TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")
BASE_URL = "https://api.holysheep.ai/v1"  # HolySheep API Endpoint


ตั้งค่า Logging

logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.INFO
)
logger = logging.getLogger(__name__)

async def get_ai_response(user_message: str) -> str:
    """
    ส่งข้อความไปยัง HolySheep API และรับการตอบกลับ
    """
    headers = {
        "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "gpt-4.1",
        "messages": [
            {"role": "system", "content": "คุณคือผู้ช่วยที่เป็นมิตร ตอบกลับเป็นภาษาไทย"},
            {"role": "user", "content": user_message}
        ],
        "max_tokens": 500,
        "temperature": 0.7
    }
    
    try:
        response = requests.post(
            f"{BASE_URL}/chat/completions",
            headers=headers,
            json=payload,
            timeout=10  # Timeout 10 วินาที
        )
        response.raise_for_status()
        data = response.json()
        return data["choices"][0]["message"]["content"]
    except requests.exceptions.Timeout:
        logger.error("API Timeout - ใช้เวลานานเกินไป")
        return "ขออภัย ระบบกำลังยุ่ง กรุณาลองใหม่อีกครั้ง"
    except requests.exceptions.RequestException as e:
        logger.error(f"Request Error: {e}")
        return "เกิดข้อผิดพลาด กรุณาลองใหม่ภายหลัง"

async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """จัดการคำสั่ง /start"""
    await update.message.reply_text(
        "สวัสดีครับ! 👋\n"
        "ผมคือ AI Bot ที่พร้อมตอบคำถามคุณ\n"
        "พิมพ์ข้อความมาได้เลยครับ!"
    )

async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """จัดการคำสั่ง /help"""
    await update.message.reply_text(
        "📌 วิธีใช้งาน:\n"
        "• พิมพ์ข้อความถามได้เลย\n"
        "• /start - เริ่มต้นใช้งาน\n"
        "• /help - แสดงวิธีใช้\n"
        "• /status - ตรวจสอบสถานะ API"
    )

async def status_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """ตรวจสอบสถานะ API"""
    try:
        headers = {"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"}
        response = requests.get(f"{BASE_URL}/models", headers=headers, timeout=5)
        if response.status_code == 200:
            await update.message.reply_text("✅ API ทำงานปกติ")
        else:
            await update.message.reply_text(f"⚠️ API สถานะ: {response.status_code}")
    except Exception as e:
        await update.message.reply_text(f"❌ ไม่สามารถเชื่อมต่อ API: {str(e)}")

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """จัดการข้อความที่ส่งมาจากผู้ใช้"""
    message_text = update.message.text
    logger.info(f"ข้อความจาก {update.effective_user.first_name}: {message_text}")
    
    # แสดงว่ากำลังพิมพ์
    await update.message.chat.send_action("typing")
    
    # รับการตอบกลับจาก AI
    ai_response = await get_ai_response(message_text)
    
    # ส่งการตอบกลับ
    await update.message.reply_text(ai_response)

async def error_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """จัดการข้อผิดพลาด"""
    logger.error(f"Update {update} ทำให้เกิดข้อผิดพลาด: {context.error}")

def main():
    """เริ่มต้น Bot"""
    application = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
    
    # ลงทะเบียน Command Handlers
    application.add_handler(CommandHandler("start", start_command))
    application.add_handler(CommandHandler("help", help_command))
    application.add_handler(CommandHandler("status", status_command))
    
    # ลงทะเบียน Message Handler
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    
    # ลงทะเบียน Error Handler
    application.add_error_handler(error_handler)
    
    logger.info("Bot เริ่มทำงานแล้ว...")
    application.run_polling(allowed_updates=Update.ALL_TYPES)

if __name__ == "__main__":
    main()

สร้างไฟล์ .env สำหรับเก็บ API Keys

# สร้างไฟล์ .env
touch .env


เนื้อหาในไฟล์ .env


TELEGRAM_BOT_TOKEN=your_telegram_bot_token_here


HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY



วิธีรับ Telegram Bot Token:


1. ไปที่ @BotFather บน Telegram


2. ส่งคำสั่ง /newbot


3. ทำตามขั้นตอนและรับ Token



วิธีรับ HolySheep API Key:


1. สมัครที่ https://www.holysheep.ai/register


2. ไปที่ Dashboard > API Keys


3. สร้าง Key ใหม่

ราคาและเปรียบเทียบค่าใช้จ่าย

HolySheep AI มีราคาที่ประหยัดมากเมื่อเทียบกับ Provider อื่น:

รองรับ WeChat และ Alipay ชำระเงินได้สะดวก

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

กรณีที่ 1: 401 Unauthorized

สาเหตุ: API Key ไม่ถูกต้องหรือหมดอายุ

# วิธีแก้ไข: ตรวจสอบ API Key
import os
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")

if not HOLYSHEEP_API_KEY or HOLYSHEEP_API_KEY == "YOUR_HOLYSHEEP_API_KEY":
    raise ValueError("กรุณาตั้งค่า HOLYSHEEP_API_KEY ในไฟล์ .env")


หรือตรวจสอบ Format ของ Key

headers = {
    "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
    "Content-Type": "application/json"
}


ตรวจสอบว่า Key ขึ้นต้นด้วย "sk-" หรือไม่

if not HOLYSHEEP_API_KEY.startswith("sk-"):
    raise ValueError("API Key ไม่ถูกต้อง กรุณาตรวจสอบที่ https://www.holysheep.ai/register")

กรณีที่ 2: ConnectionError: timeout

สาเหตุ: Server ใช้เวลานานเกินกว่า Timeout ที่กำหนด

# วิธีแก้ไข: เพิ่ม Retry Logic และ Timeout ที่เหมาะสม
import time
from functools import wraps

def retry_on_failure(max_retries=3, delay=1):
    """Decorator สำหรับ Retry เมื่อเกิดข้อผิดพลาด"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return await func(*args, **kwargs)
                except requests.exceptions.Timeout:
                    if attempt < max_retries - 1:
                        time.sleep(delay * (attempt + 1))
                        continue
                    return "ขออภัย ระบบไม่ตอบสนอง กรุณาลองใหม่ภายหลัง"
            return "เกิดข้อผิดพลาด กรุณาลองใหม่"
        return wrapper
    return decorator


ใช้งาน

@retry_on_failure(max_retries=3, delay=2)
async def get_ai_response(user_message: str) -> str:
    # ... โค้ดเดิม

กรณีที่ 3: Telegram API Error: Bad Request

สาเหตุ: ข้อความยาวเกินกว่าขีดจำกัดของ Telegram (4096 ตัวอักษร)

# วิธีแก้ไข: ตัดข้อความให้เหมาะสม
MAX_MESSAGE_LENGTH = 4096

async def safe_send_message(update: Update, text: str):
    """ส่งข้อความอย่างปลอดภัย ไม่ให้เกินขีดจำกัด"""
    if len(text) <= MAX_MESSAGE_LENGTH:
        await update.message.reply_text(text)
    else:
        # ตัดข้อความเป็นส่วนๆ
        chunks = [text[i:i+MAX_MESSAGE_LENGTH] for i in range(0, len(text), MAX_MESSAGE_LENGTH)]
        for chunk in chunks:
            await update.message.reply_text(chunk)
            time.sleep(0.5)  # รอครู่ระหว่างส่งแต่ละส่วน


ใน handle_message

ai_response = await get_ai_response(message_text)
await safe_send_message(update, ai_response)

กรณีที่ 4: Rate Limit Exceeded

สาเหตุ: ส่งคำขอไปยัง API บ่อยเกินไป

# วิธีแก้ไข: ใช้ Rate Limiter
import asyncio
from collections import defaultdict
from datetime import datetime, timedelta

class RateLimiter:
    """จำกัดจำนวนคำขอต่อผู้ใช้"""
    def __init__(self, max_requests: int = 10, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window = timedelta(seconds=window_seconds)
        self.requests = defaultdict(list)
    
    def is_allowed(self, user_id: int) -> bool:
        now = datetime.now()
        # ลบคำขอเก่าที่หมดอายุ
        self.requests[user_id] = [
            req_time for req_time in self.requests[user_id]
            if now - req_time < self.window
        ]
        
        if len(self.requests[user_id]) >= self.max_requests:
            return False
        
        self.requests[user_id].append(now)
        return True

rate_limiter = RateLimiter(max_requests=10, window_seconds=60)

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    
    if not rate_limiter.is_allowed(user_id):
        await update.message.reply_text("⏳ คุณส่งคำขอบ่อยเกินไป กรุณารอสักครู่")
        return
    
    # ... โค้ดเดิมในการรับและตอบข้อความ

วิธี Deploy ขึ้น Server จริง

หลังจากทดสอบบนเครื่อง local สำเร็จแล้ว คุณสามารถ Deploy ขึ้น Production ได้หลายวิธี:

# Dockerfile สำหรับ Production
FROM python:3.11

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง