Bot trợ lý AI đa nền tảng này cho phép người dùng tương tác trực tiếp trên Telegram để sử dụng đồng thời nhiều mô hình ngôn ngữ lớn như GPT-4, Claude và Gemini. Thiết kế hướng đến nhu cầu thực tế: giảm thiểu chuyển đổi ứng dụng, tối ưu hóa hiệu suất công việc thông qua giao diện chat quen thuộc.
Kiến trúc hệ thống
Hệ thống áp dụng mô hình lập trình bất đồng bộ (async/await) dựa trên aiogram 3.x, tận dụng khả năng xử lý song song để đáp ứng nhiều yêu cầu cùng lúc mà không gây nghẽn luồng. Mỗi mô hình AI được đóng gói thành một lớp riêng biệt tuân thủ giao diện chung:
from abc import ABC, abstractmethod
class BaseAIService(ABC):
@abstractmethod
async def generate(self, prompt: str, history: list = None) -> dict:
pass
class OpenAIService(BaseAIService):
def __init__(self, api_key: str):
self.client = AsyncOpenAI(api_key=api_key)
async def generate(self, prompt: str, history: list = None) -> dict:
messages = [{"role": "system", "content": "Bạn là trợ lý AI"}]
if history:
messages.extend(history)
messages.append({"role": "user", "content": prompt})
response = await self.client.chat.completions.create(
model="gpt-4-turbo",
messages=messages,
temperature=0.7
)
return {
"content": response.choices[0].message.content,
"usage": response.usage.total_tokens
}
class GeminiService(BaseAIService):
def __init__(self, api_key: str):
genai.configure(api_key=api_key)
self.model = genai.GenerativeModel('gemini-pro')
async def generate(self, prompt: str, history: list = None) -> dict:
context = "Bạn là trợ lý AI.\n"
if history:
for msg in history:
context += f"{msg['role']}: {msg['content']}\n"
context += f"user: {prompt}"
response = await self.model.generate_content_async(context)
return {
"content": response.text,
"usage": response.usage_metadata.total_token_count if hasattr(response, 'usage_metadata') else 0
}
Quản lý phiên hội thoại
Mỗi người dùng duy trì một phiên trò chuyện riêng, lịch sử được lưu trữ dưới dạng JSON trong SQLite. Hệ thống tự động cắt bớt lịch sử khi vượt quá giới hạn token để đảm bảo hiệu suất:
import json
import aiosqlite
class SessionManager:
def __init__(self, db_path: str = "sessions.db"):
self.db_path = db_path
async def init_db(self):
async with aiosqlite.connect(self.db_path) as db:
await db.execute("""
CREATE TABLE IF NOT EXISTS sessions (
user_id INTEGER PRIMARY KEY,
active_model TEXT DEFAULT 'gpt-4',
history TEXT DEFAULT '[]',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
await db.commit()
async def get_session(self, user_id: int) -> dict:
async with aiosqlite.connect(self.db_path) as db:
cursor = await db.execute(
"SELECT active_model, history FROM sessions WHERE user_id = ?",
(user_id,)
)
row = await cursor.fetchone()
if row:
return {
"model": row[0],
"history": json.loads(row[1])
}
return {"model": "gpt-4", "history": []}
async def update_session(self, user_id: int, model: str, new_history: list, max_turns: int = 8):
# Giữ lại chỉ N lượt hội thoại gần nhất
trimmed_history = new_history[-(max_turns * 2):] if len(new_history) > max_turns * 2 else new_history
async with aiosqlite.connect(self.db_path) as db:
await db.execute("""
INSERT INTO sessions (user_id, active_model, history, updated_at)
VALUES (?, ?, ?, CURRENT_TIMESTAMP)
ON CONFLICT(user_id) DO UPDATE SET
active_model = excluded.active_model,
history = excluded.history,
updated_at = CURRENT_TIMESTAMP
""", (user_id, model, json.dumps(trimmed_history)))
await db.commit()
Xử lý tin nhắn dài và định dạng code
Telegram giới hạn 4096 ký tự mỗi tin nhắn. Bot tự động chia nhỏ nội dung dài, đồng thời bảo toàn cấu trúc Markdown để hiển thị code đẹp mắt:
import re
from typing import List
async def send_chunked_message(bot, chat_id: int, text: str, chunk_size: int = 3900):
"""Chia nhỏ và gửi tin nhắn dài"""
chunks = []
current_pos = 0
while current_pos < len(text):
if len(text) - current_pos <= chunk_size:
chunks.append(text[current_pos:])
break
# Tìm điểm ngắt tự nhiên gần nhất
split_point = text.rfind('\n\n', current_pos, current_pos + chunk_size)
if split_point == -1:
split_point = text.rfind('\n', current_pos, current_pos + chunk_size)
if split_point == -1:
split_point = current_pos + chunk_size
chunks.append(text[current_pos:split_point])
current_pos = split_point + 1
for i, chunk in enumerate(chunks):
suffix = f" (tiếp {i+1}/{len(chunks)})" if len(chunks) > 1 and i < len(chunks)-1 else ""
try:
await bot.send_message(
chat_id=chat_id,
text=chunk + suffix,
parse_mode="MarkdownV2",
disable_web_page_preview=True
)
except Exception:
# Fallback sang plain text nếu Markdown lỗi
await bot.send_message(chat_id=chat_id, text=chunk + suffix)
Cấu hình và triển khai
Sử dụng file .env để quản lý biến môi trường nhạy cảm:
# .env
TELEGRAM_TOKEN=1234567890:ABCdefGHIjklMnOprSTUvWxyZ
OPENAI_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
GEMINI_KEY=AIzaSyxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
CLAUDE_KEY=sk-ant-xxxxxxxxxxxxxxxxxxxxxxxxxxxxx
Khởi tạo bot chính với router xử lý lệnh:
from aiogram import Bot, Dispatcher
from aiogram.filters import Command
from aiogram.types import Message
dp = Dispatcher()
bot = Bot(token=os.getenv("TELEGRAM_TOKEN"))
@dp.message(Command("switch"))
async def switch_model(message: Message):
keyboard = InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="GPT-4", callback_data="model_gpt4")],
[InlineKeyboardButton(text="Gemini", callback_data="model_gemini")],
[InlineKeyboardButton(text="Claude", callback_data="model_claude")]
])
await message.answer("Chọn mô hình AI:", reply_markup=keyboard)
@dp.message()
async def handle_message(message: Message):
session_mgr = SessionManager()
session = await session_mgr.get_session(message.from_user.id)
# Khởi tạo dịch vụ AI tương ứng
ai_service = get_ai_service(session["model"])
status_msg = await message.reply("🧠 Đang xử lý...")
try:
result = await ai_service.generate(message.text, session["history"])
# Cập nhật lịch sử
updated_history = session["history"] + [
{"role": "user", "content": message.text},
{"role": "assistant", "content": result["content"]}
]
await session_mgr.update_session(message.from_user.id, session["model"], updated_history)
await status_msg.delete()
await send_chunked_message(bot, message.chat.id, result["content"])
except Exception as e:
await status_msg.edit_text(f"Lỗi: {str(e)}")
if __name__ == "__main__":
asyncio.run(dp.start_polling(bot))
Xử lý sự cố thường gặp
- Lỗi kết nối API: Kiểm tra firewall, proxy, hoặc thêm retry logic với backoff
- Tin nhắn quá dài: Triển khai hàm chia nhỏ và fallback về plain text
- Code hiển thị sai: Escape ký tự đặc biệt trong MarkdownV2:
_ * [ ] ( ) ~ ` > # + - = | { } . ! - Hiệu năng chậm: Thêm index cho cột user_id trong database, sử dụng connection pool
Hệ thống có thể mở rộng bằng cách thêm các mô hình mới chỉ cần implement interface BaseAIService và đăng ký vào factory. Kiến trúc phân tầng giúp dễ dàng thay thế thành phần hoặc chuyển sang nền tảng khác như Discord hay Slack.