home

หน้าแรก
 Computer >> บทช่วยสอนคอมพิวเตอร์ >  >> การเขียนโปรแกรม >> Redis

สร้าง AI Chatbot ด้วย GPT, Redis และ Python: คู่มือฉบับสมบูรณ์

สถาปัตยกรรมแอปพลิเคชัน

โครงสร้างระดับสูง (High-level architecture):

  • Client/UI: React 18 ใช้ WebSocket ติดต่อแบ็กเอนด์
  • AI Model: GPT-J-6B (6B parameters) ผ่าน Hugging Face Inference API (ฟรีสำหรับการใช้งานพื้นฐาน)
  • Redis: ใช้ Redis JSON เก็บประวัติแชท, Redis Streams เป็นคิวข้อความระหว่างเว็บเซิร์ฟเวอร์และ worker
  • Chat API: FastAPI + WebSocket สำหรับการสื่อสาร real-time แบบสองทาง

การตั้งค่าสภาพแวดล้อมพัฒนา

# โครงสร้างโปรเจกต์
mkdir fullstack-ai-chatbot
cd fullstack-ai-chatbot
mkdir client server
git init
touch .gitignore

สร้าง Chat Server ด้วย FastAPI และ WebSocket

ตั้งค่า Python Environment

cd server
python3.8 -m venv env
source env/bin/activate
pip install fastapi uuid uvicorn gunicorn websockets python-dotenv aioredis

สร้างไฟล์ .env กำหนด APP_ENV=development

FastAPI Server Setup

# main.py
from fastapi import FastAPI, Request
import uvicorn, os
from dotenv import load_dotenv
load_dotenv()

api = FastAPI()

@api.get("/test")
async def root():
    return {"msg": "API is Online"}

if __name__ == "__main__":
    if os.environ.get('APP_ENV') == "development":
        uvicorn.run("main:api", host="0.0.0.0", port=3500, workers=4, reload=True)

เพิ่ม Routes (Routes/Chat)

# src/routes/chat.py
from fastapi import APIRouter, WebSocket, Request, Depends, HTTPException
import uuid
from ..socket.connection import ConnectionManager
from ..socket.utils import get_token

chat = APIRouter()
manager = ConnectionManager()

@chat.post("/token")
async def token_generator(name: str, request: Request):
    if name == "":
        raise HTTPException(status_code=400, detail={"loc": "name", "msg": "Enter a valid name"})
    token = str(uuid.uuid4())
    return {"name": name, "token": token}

@chat.post("/refresh_token")
async def refresh_token(request: Request):
    return None

@chat.websocket("/chat")
async def websocket_endpoint(websocket: WebSocket, token: str = Depends(get_token)):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.send_personal_message(f"Response: Simulating response from the GPT service", websocket)
    except WebSocketDisconnect:
        manager.disconnect(websocket)

Connection Manager

# src/socket/connection.py
from fastapi import WebSocket
from typing import List

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []
    
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
    
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
    
    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

Dependency Injection สำหรับ Token Validation

# src/socket/utils.py
from fastapi import WebSocket, status, Query
from typing import Optional

async def get_token(websocket: WebSocket, token: Optional[str] = Query(None)):
    if token is None or token == "":
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
    return token

สร้างระบบ Real-time ด้วย Redis

Redis และ Distributed Messaging Queues

แยก worker environment จาก web server: web server ใส่ข้อความเข้า message queue (producer), worker ดึงข้อความ เรียก Hugging Face API, ใส่ผลลัพธ์เข้า response queue

เชื่อมต่อ Redis Cluster ใน Python

# src/redis/config.py
import os, aioredis
from dotenv import load_dotenv
from rejson import Client
load_dotenv()

class Redis:
    def __init__(self):
        self.REDIS_URL = os.environ['REDIS_URL']
        self.REDIS_PASSWORD = os.environ['REDIS_PASSWORD']
        self.REDIS_USER = os.environ['REDIS_USER']
        self.connection_url = f"redis://{self.REDIS_USER}:{self.REDIS_PASSWORD}@{self.REDIS_URL}"
        self.REDIS_HOST = os.environ['REDIS_HOST']
        self.REDIS_PORT = os.environ['REDIS_PORT']
    
    async def create_connection(self):
        self.connection = aioredis.from_url(self.connection_url, db=0)
        return self.connection
    
    def create_rejson_connection(self):
        self.redisJson = Client(host=self.REDIS_HOST, port=self.REDIS_PORT,
                                decode_responses=True, username=self.REDIS_USER,
                                password=self.REDIS_PASSWORD)
        return self.redisJson

ทำงานกับ Redis Streams (Producer)

# src/redis/producer.py
from .config import Redis

class Producer:
    def __init__(self, redis_client):
        self.redis_client = redis_client
    
    async def add_to_stream(self, data: dict, stream_channel):
        try:
            msg_id = await self.redis_client.xadd(name=stream_channel, id="*", fields=data)
            return msg_id
        except Exception as e:
            print(f"Error: {e}")

สร้างโมเดลข้อมูลแชท (Pydantic)

# src/schema/chat.py
from datetime import datetime
from pydantic import BaseModel
from typing import List
import uuid

class Message(BaseModel):
    id: str = str(uuid.uuid4())
    msg: str
    timestamp: str = str(datetime.now())

class Chat(BaseModel):
    token: str
    messages: List[Message] = []
    name: str
    session_start: str = str(datetime.now())

ใช้ Redis JSON เก็บ Session

# ใน /token endpoint
json_client = redis.create_rejson_connection()
chat_session = Chat(token=token, messages=[], name=name)
json_client.jsonset(str(token), Path.rootPath(), chat_session.dict())
redis_client = await redis.create_connection()
await redis_client.expire(str(token), 3600)  # 60 นาที

อัปเดต Token Dependency ตรวจสอบ Redis

# src/socket/utils.py
from ..redis.config import Redis

async def get_token(websocket: WebSocket, token: Optional[str] = Query(None)):
    if token is None or token == "":
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
    redis_client = await redis.create_connection()
    isexists = await redis_client.exists(token)
    if isexists == 1:
        return token
    else:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason="Session expired")

เพิ่มสติปัญญาให้ Chatbot ด้วย AI Model

เชื่อมต่อ Hugging Face Inference API

# worker/src/model/gptj.py
import os, requests, json
from dotenv import load_dotenv
load_dotenv()

class GPT:
    def __init__(self):
        self.url = os.environ.get('MODEL_URL')
        self.headers = {"Authorization": f"Bearer {os.environ.get('HUGGINFACE_INFERENCE_TOKEN')}"}
        self.payload = {"inputs": "", "parameters": {"return_full_text": False, "use_cache": True, "max_new_tokens": 25}}
    
    def query(self, input: str) -> str:
        self.payload["inputs"] = f"Human: {input} Bot:"
        data = json.dumps(self.payload)
        response = requests.request("POST", self.url, headers=self.headers, data=data)
        data = json.loads(response.content.decode("utf-8"))
        text = data[0]['generated_text']
        return str(text.split("Human:")[0]).strip("\n").strip()

จำลอง Short-term Memory ด้วย Redis JSON

ดึงประวัติ 4 ข้อความล่าสุด รวมกับข้อความใหม่ ส่งไปยังโมเดล เก็บคำตอบกลับเข้า Redis

# worker/src/redis/cache.py
from .config import Redis
from rejson import Path

class Cache:
    def __init__(self, json_client):
        self.json_client = json_client
    
    async def get_chat_history(self, token: str):
        return self.json_client.jsonget(str(token), Path.rootPath())
    
    async def add_message_to_cache(self, token: str, source: str, message_data: dict):
        if source == "human":
            message_data['msg'] = "Human: " + message_data['msg']
        elif source == "bot":
            message_data['msg'] = "Bot: " + message_data['msg']
        self.json_client.jsonarrappend(str(token), Path('.messages'), message_data)

Stream Consumer (Worker)

# worker/src/redis/stream.py
class StreamConsumer:
    def __init__(self, redis_client):
        self.redis_client = redis_client
    
    async def consume_stream(self, count: int, block: int, stream_channel):
        return await self.redis_client.xread(streams={stream_channel: '0-0'}, count=count, block=block)
    
    async def delete_message(self, stream_channel, message_id):
        await self.redis_client.xdel(stream_channel, message_id)

Worker loop: ดึงข้อความจาก message_channel → เรียก GPT → ใส่คำตอบเข้า response_channel → ลบข้อความที่ประมวลผลแล้ว

ส่งคำตอบกลับ Client (Server)

WebSocket endpoint ฟัง response_channel ตรวจสอบ token ตรงกัน → ส่งคำตอบผ่าน WebSocket

Refresh Token Endpoint

@chat.get("/refresh_token")
async def refresh_token(request: Request, token: str):
    json_client = redis.create_rejson_connection()
    cache = Cache(json_client)
    data = await cache.get_chat_history(token)
    if data is None:
        raise HTTPException(status_code=400, detail="Session expired")
    return data

สรุป

สร้างระบบแชทบอทครบวงจร: Token-based session, Redis Streams สำหรับ message queue, Redis JSON สำหรับประวัติ, Worker แยกประมวลผล AI, WebSocket real-time โค้ดเต็มที่ GitHub