Bài toán thực tế: Ba nền tảng, ba bộ code riêng biệt
Sáu tháng trước, team mình nhận yêu cầu deploy chatbot AI cho một khách hàng doanh nghiệp. Họ dùng Slack nội bộ, khách hàng B2C của họ dùng Telegram, còn team dev lại dùng Discord. Ba nền tảng, ba đối tượng khác nhau — và mình đã làm điều sai lầm nhất có thể: viết ba bot riêng biệt.
Hậu quả rõ sau đúng 2 tuần: ba codebase không đồng bộ, fix bug ở Telegram xong quên update Slack, logic AI response mỗi chỗ một kiểu. Mỗi lần khách hàng muốn đổi prompt hay thêm tính năng là phải vào ba chỗ sửa. Maintenance nightmare thực sự.
Không riêng gì team mình — pattern này lặp lại ở nhiều dự án chatbot. Cách tiếp cận “mỗi nền tảng một project riêng” có vẻ nhanh lúc đầu, nhưng 2-3 tháng sau là technical debt tích lũy đủ để kéo cả team xuống.
Tại sao tích hợp đa nền tảng lại khó?
Vấn đề cốt lõi là mỗi nền tảng có:
- Authentication riêng: Telegram dùng Bot Token, Slack dùng OAuth + Signing Secret, Discord dùng Bot Token + Application ID
- Event model khác nhau: Telegram dùng polling hoặc webhook, Slack dùng Event API với HTTP endpoint, Discord dùng WebSocket gateway
- Message format không tương thích: Markdown của Telegram khác Slack Block Kit khác Discord Embeds
- Rate limit khác nhau: Telegram giới hạn 30 messages/giây/bot, Slack có Tier 1-4 API limits, Discord có per-guild rate limits
Cố gắng abstract tất cả vào một interface chung mà chưa hiểu rõ từng platform? Kết quả thường là over-engineer từ đầu. Hoặc tệ hơn: một abstraction rò rỉ khắp nơi, không dùng được trong thực tế.
Ba cách giải quyết — và vấn đề của từng cách
Cách 1: Bot framework có sẵn (Botpress, Rasa, Microsoft Bot Framework)
Mình đã thử Botpress và bỏ sau 3 ngày. Khi cần inject LLM tùy chỉnh — như GPT-4 với system prompt riêng, hoặc model local qua Ollama — các framework này thêm quá nhiều abstraction layer. Debug response lỗi từ AI rất khó vì phải trace qua 5-6 layer. Rule-based dialog tree thì ổn. AI chatbot cần control behavior ở prompt level? Tránh xa.
Cách 2: Shared library — một package Python dùng chung
Tốt hơn cách 1, nhưng vẫn có vấn đề cấu trúc: ba entry point riêng biệt, ba server/process chạy độc lập. Cập nhật AI model hay đổi conversation logic vẫn phải deploy ba lần. Với team nhỏ, overhead đó tích lũy nhanh hơn bạn tưởng.
Cách 3: Unified adapter pattern — một core, nhiều adapter
Sau khi refactor lại từ đầu, đây là approach mình settle với. Core idea đơn giản: tách AI logic khỏi platform integration hoàn toàn. Một process, một codebase, nhiều platform.
Kiến trúc thực tế: Unified Adapter Pattern
Cấu trúc project của mình trông như thế này:
ai-chatbot/
├── core/
│ ├── ai_handler.py # Toàn bộ logic AI (OpenAI/Gemini/Ollama)
│ ├── conversation.py # Quản lý conversation history
│ └── message.py # Unified message model
├── adapters/
│ ├── telegram_adapter.py
│ ├── slack_adapter.py
│ └── discord_adapter.py
├── main.py # Entry point — chạy tất cả adapters
└── requirements.txtQuan trọng nhất là Unified Message Model — một dataclass chuẩn mà tất cả adapter phải convert về:
# core/message.py
from dataclasses import dataclass
from typing import Optional
@dataclass
class UnifiedMessage:
platform: str # "telegram", "slack", "discord"
user_id: str # Platform-specific user ID
chat_id: str # Channel/chat để reply về
text: str # Nội dung tin nhắn
username: Optional[str] = None
reply_to_message_id: Optional[str] = NoneAI handler nhận vào UnifiedMessage và trả về plain text. Adapter lo việc format text đó theo chuẩn từng nền tảng:
# core/ai_handler.py
from openai import AsyncOpenAI
from .conversation import ConversationManager
from .message import UnifiedMessage
client = AsyncOpenAI()
conversation_mgr = ConversationManager()
async def process_message(msg: UnifiedMessage) -> str:
# Lấy conversation history theo user_id (cross-platform)
history = conversation_mgr.get_history(msg.user_id)
history.append({"role": "user", "content": msg.text})
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "Bạn là AI assistant hữu ích."},
*history
],
max_tokens=1000
)
reply = response.choices[0].message.content
conversation_mgr.add_reply(msg.user_id, reply)
return replyTelegram Adapter
# adapters/telegram_adapter.py
from telegram import Update
from telegram.ext import Application, MessageHandler, filters
from core.message import UnifiedMessage
from core.ai_handler import process_message
async def handle_message(update: Update, context):
msg = UnifiedMessage(
platform="telegram",
user_id=str(update.effective_user.id),
chat_id=str(update.effective_chat.id),
text=update.message.text,
username=update.effective_user.username
)
reply = await process_message(msg)
await update.message.reply_text(reply)
def create_telegram_app(token: str):
app = Application.builder().token(token).build()
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
return appDiscord Adapter
# adapters/discord_adapter.py
import discord
from core.message import UnifiedMessage
from core.ai_handler import process_message
client = discord.Client(intents=discord.Intents.default())
@client.event
async def on_message(message: discord.Message):
if message.author == client.user:
return
if not client.user.mentioned_in(message):
return
text = message.content.replace(f'<@{client.user.id}>', '').strip()
msg = UnifiedMessage(
platform="discord",
user_id=str(message.author.id),
chat_id=str(message.channel.id),
text=text,
username=message.author.name
)
async with message.channel.typing():
reply = await process_message(msg)
await message.reply(reply)Chạy tất cả trong một process
# main.py
import asyncio
from adapters.telegram_adapter import create_telegram_app
from adapters.discord_adapter import client as discord_client
import os
async def main():
telegram_app = create_telegram_app(os.getenv("TELEGRAM_BOT_TOKEN"))
# Chạy cả hai adapter đồng thời
await asyncio.gather(
telegram_app.run_polling(),
discord_client.start(os.getenv("DISCORD_BOT_TOKEN"))
)
if __name__ == "__main__":
asyncio.run(main())
Những vấn đề phát sinh khi chạy production
Sau 6 tháng production, kiến trúc này hoạt động tốt — nhưng có vài vấn đề chỉ lộ ra khi deploy thật:
Conversation history cross-platform: Cùng một người dùng cả Telegram lẫn Discord — hai user_id khác nhau, hai history riêng biệt. Không cần build hệ thống user linking phức tạp: dùng platform:user_id làm key là đủ.
Discord và asyncio conflict: discord.py có event loop riêng, đôi khi conflict với asyncio.gather. Cần dùng discord.Client với setup_hook cẩn thận hoặc chạy Discord trong thread riêng với asyncio.run_coroutine_threadsafe.
Rate limiting AI API: Peak hours, cả ba platform đổ message vào cùng lúc — OpenAI API throttle ngay. Thêm semaphore vào ai_handler là cách nhanh nhất để kiểm soát:
# Trong ai_handler.py
import asyncio
_semaphore = asyncio.Semaphore(5) # Tối đa 5 concurrent AI calls
async def process_message(msg: UnifiedMessage) -> str:
async with _semaphore:
# ... OpenAI call ở đâyKết quả sau 6 tháng
Giờ khi khách hàng yêu cầu đổi system prompt hay cập nhật model, mình sửa một chỗ trong ai_handler.py và deploy một lần. Bug fix cũng vậy — không còn cảnh “fix Telegram xong quên Slack” nữa.
Một lưu ý thực tế về Slack: adapter này phức tạp hơn hai cái kia vì Slack yêu cầu xử lý event verification và dùng slack-bolt với HTTP server riêng. Slack chạy tốt nhất trên port riêng (3000 là convention phổ biến) với reverse proxy nginx — đừng cố nhét vào cùng event loop với Telegram/Discord.
400 dòng Python cho ba nền tảng — nhỏ hơn tổng code của một platform nếu viết riêng lẻ. Nếu project của bạn cần scale ra nhiều channel, đầu tư vào architecture từ đầu rẻ hơn nhiều so với refactor lúc codebase đã to.
