Django Gemini ChatBot 연동

Intro

이번엔 지난 시간에 이어 Django 서버에 Langchain을 이용하여 Gemini API 연동한 chatbot을 만드려고 한다.

하지만 chatbot의 통신 방식을 websocket을 연결하는 방식으로 진행하는데 채팅에 대한 테스트를 진행하기 위해서 PostMan을 사용하며 Gemini에서 생성된 데이터를 chunk 단위로 전송하여 통신하고 다음 포스팅에서 서버에서 chunk를 하나의 텍스트로 조합하여 보내도록 변경하는 방식으로 진행한다.

참고

https://medium.com/@m.moshek/streaming-llm-output-with-django-react-and-langchain-tutorial-2963275b4f9c

 

Install

pip install langchain==0.1.11 langchain-community==0.0.26 langchain-openai==0.0.8 channels==4.0.0 daphne==4.1.0 python-dotenv==1.0.1

 

Websocket 요청을 받기 위한 앱 추가

INSTALLED_APPS의 리스트에 daphne 추가 여기서 중요한 점은 staticfiles 이전에 추가해 주어야 한다는 점.

INSTALLED_APPS =[
    ***
    'django.contrib.messages',
    'daphne',
    'django.contrib.staticfiles',
    ***
]

 

Django Channels를 배포하기 위해서는 HTTP요청은 uWSGI프로토콜로 받고 ,WS 요청은 ASGI프로토콜로 받아야하는데 Daphne는 Django Channel를 지원하기 위해 개발되었고 HTTP,HTTP2,WS프로토콜 서버로 HTTP와 WS 요청을 받아 들여 자동으로 어떤 프로토콜으로 처리해야할지 결정해준다.

공식 문서에서 보안상의 이유로 HTTP 요청을 Daphne에서 처리하는게 불안하다면  WSGI 서버 앞에 Daphne를 두도록 설정하라고 한다. 하지만 이번 포스팅에서는 WS를 처리하기 위한 작업만 추가하고 이외의 작업은 추후에 프로젝트를 고도화 하는 단계에서 진행해야할것 같다.

 

ASGI

# WSGI_APPLICATION = 'snowfall.wsgi.application'
ASGI_APPLICATION = "snowfall.asgi.application"

 

웹서버와 연결하기 위한 방법을 변경한다.

 

Views.py

웹소캣을 통해 입력을 받으면 GEMINI API와 연결하여 응답을 생성하고 Response를 보내는 파일을 작성했다.

API Key는 .env 파일을 통해 관리합니다.

langchainAgent앱 하위의 views.py에 아래 코드를 작성한다.

from django.shortcuts import render
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from channels.generic.websocket import AsyncWebsocketConsumer
import json
import os

prompt = ChatPromptTemplate.from_messages([
    ("system", "너는 어떤 언어로 물어보더라도 한국어로 대답해야해"),
    ("user", "{input}")
])


llm = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    google_api_key=os.getenv("GEMINI_API_KEY"),
    temperature=0.7
)
#Langchain의 기본적인 문자열 출력 파서.
output_parser = StrOutputParser()

# Langchain 체인 구성 prompt | llm | output_parser
chain = prompt | llm.with_config({"run_name": "model"}) | output_parser.with_config({"run_name": "Assistant"})


class ChatConsumer(AsyncWebsocketConsumer):

    async def connect(self):
        await self.accept()

    async def disconnect(self, close_code):
        pass

    async def receive(self, text_data):
        text_data_json = {}
        if type (text_data) == json:
            text_data_json = json.loads(text_data)
        else:
            text_data_json["message"] = text_data
        message = text_data_json["message"]

        try:

            async for chunk in chain.astream_events({'input': message}, version="v1", include_names=["Assistant"]):
                if chunk["event"] in ["on_parser_start", "on_parser_stream"]:
                    await self.send(text_data=json.dumps(chunk))

        except Exception as e:
            print(e)

 

해당 코드에 대해 간단하게 설명을 하면 기본적으로 시스템  역할 정의를 한국어로 답하는 봇이고 사용자 입력을 input으로 받아 응답을 생성합니다. 여기서 Gemini를 사용하였는데 해당 api를 사용한 이유는 무료이기 때문에... 사용하였고 이 외에 다른 모델을 사용하고 싶다면

해당 api 공식 문서를 참고하면 될것 같습니다!

CathConsumer Class는 Django Channels의 WebSocket 연결을 처리하기 위한 클래스

클래스 내부의 내용은 클라이언트가 보낸 메시지를 Json으로 보낸 경우 Json 데이터로 파싱하고 일반 텍스트일 경우 일반 텍스트 그대로 사용하는데 보통 이 내용은 FE 개발자와 인터페이스를 맞추면 됩니다. 저는 Json 내용으로 프론트 엔드를 작성해두었지만 postman에서 테스트 해야하기 때문에 일반 텍스트도 받을 수 있게 설정 했습니다.

 

WebSocket Routing

langchainAgent앱 하위에 routing.py와 urls.py 파일을 추가 후 다음과 같은 코드를 작성했다.

해당 파일들은 Django Channels에서 WebSocket 요청을 처리하기 위해 경로를 설정했다.

#routing.py
from django.urls import re_path  
from . import views  
  
websocket_urlpatterns = [  
    re_path(r'^ws/chat/$', views.ChatConsumer.as_asgi()),
]

정규식을 통해 ws/chat/ 경로로 들어오는 webSocket 요청을 CathConsumer가 처리하도록 연결하고 as_asgi()를 사용하여 Django의 비동기 ASGI 환경에 맞도록 설정.

#urls.py
from django.urls import path  
from . import views  
  
  
urlpatterns = [  
    path('ws/chat/', views.ChatConsumer.as_asgi()),  
]

 

ASGI 설정

루트 폴더의 asgi.py 파일을 다음과 같이 변경하여  각 요청에 대한 프로토콜을 설정해준다.

import os  
from django.core.asgi import get_asgi_application  
from channels.routing import ProtocolTypeRouter, URLRouter  
from channels.auth import AuthMiddlewareStack  
from langchainAgent.routing import websocket_urlpatterns  
  
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'snowfall.settings') 
  
application = ProtocolTypeRouter({  
  "http": get_asgi_application(),  
  "websocket": AuthMiddlewareStack(  
        URLRouter(  
            websocket_urlpatterns  
        )  
    ),  
})

이렇게 하고 서버를 실행하면 간단하게 Langchain을 이용한 chatbot을 생성할 수 있습니다. 하지만 메모리 기능이 없고 실시간 데이터를 얻어오는 등의 기능은 추가적인 작업이 필요합니다. 

Postman 테스트

 

 

이렇게 연결 후 대화를 할 수 있는데 응답값이 json 형식으로 오게 된다.

근데 여기서 내용을 보면 불필요한 내용들도 있는데 그 이유는 LangChain 내부의 실행 상태를 여러 단계로 나누어 전달하는데 이 모든것을 전달하기 때문에 불필요한 데이터도 오게 된다.

정상 응답 데이터 전문

{
    "event": "on_parser_stream",
    "name": "Assistant",
    "run_id": "a3a69e43-634a-498e-b3eb-4fbe931e0fce",
    "tags": [
        "seq:step:3"
    ],
    "metadata": {},
    "data": {
        "chunk": " \ubb34\uc5c7\uc744 \ub3c4\uc640\ub4dc\ub9b4\uae4c\uc694?"
    },
    "parent_ids": []
}

위 데이터가 정상 응답 데이터인데 이것만 받기 위해서는

if chunk["event"] in ["on_parser_start", "on_parser_stream"]:
                    await self.send(text_data=json.dumps(chunk))

=>
if chunk["event"] == "on_parser_stream":
            await self.send(text_data=json.dumps(chunk))

이렇게 바꿔주면 된다.

정상  응답 데이터에서 chunk 부분이 응답 값인데 unicode_escape형식으로 인코딩된 한글 문자열이기 때문에 이를 디코딩 해주면 무엇을 도와드릴까요? 가 된다.

이렇게 Django에 Langchain을 연결하여 chatbot을 연동하는 방법을 알아봤다. 기존 작성자 분이 내용 설명을 잘 해주셨기 때문에 크게 어려움 없지 진행할 수 있었고 이제 여기에 살을 붙이는 과정을 진행 해야겠다.
아마 다음 포스팅은 Django 서버 이전에 react를 통한 프론트 엔드 구축에 대한 내용을 먼저 작성하게 될것 같다.