Yups, setelah sebelumnya aku berhasil membuat sebuah chatbot
Telegram berbasis AI sederhana, kali ini setelah ngereview kembali kode semalam (yang hampir seluruhnya digenerate oleh AI), kelliatan berantakan banget. Jadi aku mutusin buat nyusun kembali kode-kode ini agar menjadi lebih rapi dan terstruktur.
Rencananya bukan cuma sekedar refactoring, tapi membuat arsitektur aplikasi yang terstandarisasi, azzeekkk...
Jadi, aku mulai riset, googling, prompting, searching, watching, thinking, mancing kira-kira arsitektur apa yang keren, cocok dan enak dipake buat bikin AI chatbot ini. Ternyata ada banyak banget guys, bjirr baru tahu, aing...
Di atas ini adalah contoh dari sebagian Software Architecture Patterns yang ada, yang lain juga masih banyak sih.
Setelah tahu ada begitu banyak arsitektur desain yang ada dan mencoba mempelajari beberapa, aku mutusin buat di tahap awal ini, sepertinya yang penting refactoring aja dulu, mengingat kurangnya pengalaman untuk nentuin mana arsitektur yang cocok, hehe...
Jadi, aku misahin kode sebelumnya menjadi 2 objek utama, kaya diagram di bawah.
Karena AI Chatbot kita masih sebatas nerusin respon dari LLM melalui Ollama ketika ada chat yang masuk dan diterima oleh bot.
Berdasarkan gambar di atas, kode program kita terbagi menjadi 3 file:
JatiBot.py
from dotenv import load_dotenv
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters, ContextTypes
import os
class JatiBot:
def __init__(self):
load_dotenv()
self.message:str
self.token = os.getenv("TELEGRAM_TOKEN")
self.app = ApplicationBuilder().token(self.token).build()
def splitString(self, string, n):
return [string[i:i+n] for i in range(0, len(string), n)]
async def getMessage(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
self.message = update.message.text
async def sendReply(self, message, update: Update, context: ContextTypes.DEFAULT_TYPE):
max_length = 4096
messages = self.splitString(message, max_length)
for msg in messages:
await update.message.reply_text(msg)
async def start(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text("Halo! Kirim pesan apa pun untuk mulai berbicara.")
def run(self, handleMessage):
self.app.add_handler(CommandHandler("start", self.start))
self.app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handleMessage))
self.app.run_polling()
JatiLlama.py
from dotenv import load_dotenv
import os, json, requests
class JatiLlama:
def __init__(self):
load_dotenv()
self.ollama_host:str = os.getenv('OLLAMA_HOST')
self.ollama_config:dict = {
"model": os.getenv('OLLAMA_MODEL'),
"stream": False
}
self.system_initial:dict
self.conversationsFile = os.getenv('CONVERSATIONS_FILE')
# Set Initial System Role
def setSystemRole(self, content:str):
self.system_initial = {
"role": "system",
"content": content if content else "Anda adalah chatbot yang membantu orang mencari informasi dengan mudah."
}
# Load RoleModel
def loadConversations(self):
with open(self.conversationsFile, "r") as file:
return json.load(file)
# Save RoleModel
def saveConversations(self, conversations:list):
with open(self.conversationsFile, "w") as file:
json.dump(conversations, file, indent=4)
# Send Message to Ollama
def sendMessage(self, message):
reply:dict
try:
data = self.ollama_config
conversations:list = self.loadConversations()
conversations.append({"role": "user", "content": message})
data['messages'] = conversations.copy()
data['messages'].insert(0, self.system_initial)
response = requests.post(f"{self.ollama_host}/api/chat", json=data)
if response.status_code == 200:
reply = response.json()
conversations.append(reply['message'])
self.saveConversations(conversations)
else:
return f"Error {response.status_code}: {response.text}"
except Exception as e:
return f"Error: {e}"
return reply['message']['content']
main.py
import logging
from telegram import Update
from telegram.ext import ContextTypes
from JatiBot import JatiBot
from JatiLlama import JatiLlama
myBot:JatiBot = JatiBot()
myAi:JatiLlama = JatiLlama()
async def handleMessage(update: Update, context: ContextTypes.DEFAULT_TYPE):
message = await myBot.getMessage(update, context)
myAi.setSystemRole("Anda adalah asisten virtual yang dapat membantu mahasiswa menyelesaikan penulisan tugas akhir mereka.")
reply = myAi.sendMessage(message)
await myBot.sendReply(reply, update, context)
logging.basicConfig(level=logging.INFO)
if __name__ == "__main__":
print("Bot berjalan...")
myBot.run(handleMessage)
Walaupun menurutku kode ini masih kurang efisien dan belum sesuai kaidah 'Clean Code' ya, but it's okaylahyaa...
to be continued...