End-to-End 연습5_LLM 대화형 RAG 챗봇 구현하기

이번에는 정해진 시간에 응답이 오는 알림이 아니라, 유저와 실시간으로 대화가 가능하도록 만들어보려고 합니다.

 

우선 텔레그램 봇 라이브러리를 설치 후에

pip install python-telegram-bot

 

봇 실행 파일 bot.py를 만들고

파일에 다음 내용을 작성

 

본격적으로 RAG를 올리기 전에, 실시간 챗봇 테스트를 해보겠습니다.

from telegram import Update
from telegram.ext import ApplicationBuilder, ContextTypes, MessageHandler, filters

TELEGRAM_BOT_TOKEN = "**********"

# 텔레그램 메시지 수신 시 호출되는 함수
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
	user_text = update.message.text
	print(f"유저 질문 수신: {user_text}")

	# 앵무새 답장
	reply_msg = f"[봇 테스트 응답] 보내신 메시지: '{user_text}'"
	await update.message.reply_text(reply_msg)


if __name__ == "__main__":
	print("텔레그램 챗봇 수신 대기 중... (종료: Ctrl+C)")
	app = ApplicationBuilder().token(TELEGRAM_BOT_TOKEN).build()

	# 텍스트 메시지 감지 핸들러 등록
	app.add_handler(
		MessageHandler(filters.TEXT & (~filters.COMMAND), handle_message)
	)

	# 24시간 폴링(대기) 시작
	app.run_polling()

유저의 text를 인식하고 그대로 출력하는 앵무새 봇 구현

-> 문제 없음

 

이제 간단한 RAG를 도입해보겠습니다.

 

앞서 진행한 단계인

①매 시간마다 데이터를 보내 LLM 답변을 받는 알람 서비스와

②실시간 대화 서비스를 더해

 

"실시간으로 사용자의 질문과 함께 서버에 있는 데이터를 LLM으로 보내

돌아오는 신뢰성 있는(데이터를 기반으로 한) 답변을 출력한다"

from datetime import datetime
import urllib.parse
from google import genai
import pandas as pd
from sqlalchemy import create_engine
from telegram import Update
from telegram.ext import ApplicationBuilder, ContextTypes, MessageHandler, filters

GEMINI_API_KEY = ""
client = genai.Client(api_key=GEMINI_API_KEY)

TELEGRAM_BOT_TOKEN = ""

db_password = ""
safe_password = urllib.parse.quote_plus(db_password)
db_endpoint = "database-1.cdyyqwcq2v17.ap-northeast-2.rds.amazonaws.com"
ssl_path = "/home/ubuntu/stock_project/global-bundle.pem"

final_url = f"mysql+pymysql://admin:{safe_password}@{db_endpoint}:3306/yfinance_db"
engine = create_engine(final_url, connect_args={"ssl": {"ca": ssl_path}})


def get_rag_response(user_query):
    # AWS RDS MySQL에서 최신 주가/지표 검색
    query = "SELECT Date, Close FROM samsung_stock ORDER BY Date ASC"
    df = pd.read_sql(query, con=engine)

    df["MA5"] = df["Close"].rolling(window=5).mean()
    df["MA20"] = df["Close"].rolling(window=20).mean()

    latest = df.iloc[-1]
    prev = df.iloc[-2] if len(df) >= 2 else latest

    today_date = pd.to_datetime(latest["Date"]).strftime("%Y-%m-%d")
    close_price = int(latest["Close"])
    prev_close = int(prev["Close"])
    price_change = close_price - prev_close
    change_rate = (
        (price_change / prev_close) * 100 if prev_close != 0 else 0.0
    )
    ma5 = int(latest["MA5"]) if pd.notnull(latest["MA5"]) else close_price
    ma20 = int(latest["MA20"]) if pd.notnull(latest["MA20"]) else close_price

    # DB 검색 데이터를 컨텍스트로 결합
    context_data = f"""
    [DB 최신 주가 정보 - 기준일자: {today_date}]
    - 오늘 종가: {close_price:,}원 (전일 대비 {price_change:+,}원, {change_rate:+.2f}%)
    - 5일 이동평균선(MA5): {ma5:,}원
    - 20일 이동평균선(MA20): {ma20:,}원
    """

    prompt = f"""
    너는 삼성전자 전담 주식 AI 에이전트야. 아래 제공된 [DB 최신 주가 정보]만을 바탕으로 사용자의 질문에 답변해줘.
    DB 데이터와 연관된 정보는 친절하고 명확하게 설명하고, DB에 전혀 없는 내용에 대해서는 솔직하게 잘 모른다고 답해줘.

    {context_data}

    사용자 질문: {user_query}
    """

    # Gemini 호출
    interaction = client.interactions.create(
        model="gemini-3.5-flash", input=prompt
    )
    return interaction.output_text.strip()


# 텔레그램 메시지 수신 시 실행되는 비동기 함수
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_text = update.message.text
    print(f"유저 질문 수신: {user_text}")

    # RAG 파이프라인 수행 (Gemini 응답 받기)
    bot_response = get_rag_response(user_text)

    await update.message.reply_text(bot_response)


if __name__ == "__main__":
    print("RAG 주식 대화형 챗봇 서버 작동 중... (Ctrl+C로 종료)")
    app = ApplicationBuilder().token(TELEGRAM_BOT_TOKEN).build()

    # 텍스트 메시지 수신 핸들러
    app.add_handler(
        MessageHandler(filters.TEXT & (~filters.COMMAND), handle_message)
    )

    app.run_polling()

전달한 데이터(7월 20일의 종가데이터와 등락률)를 바탕으로 7월 19일 정보를 추론한 모습이다