AI चैटबॉट में महारत: Redis और Python के साथ शक्तिशाली GPT-संचालित बॉट बनाएं - एक व्यापक गाइड

द्वारा स्टीफन सैनवो

एक वर्किंग फुल-स्टैक एप्लिकेशन बनाने के लिए, विचार करने के लिए बहुत सारे चलते हुए हिस्से हैं। और आपको कई निर्णय लेने होंगे जो आपके ऐप की सफलता के लिए महत्वपूर्ण होंगे।

उदाहरण के लिए, आप किस भाषा का उपयोग करेंगे और किस प्लेटफॉर्म पर डिप्लॉय करेंगे? क्या आप एक सर्वर पर कंटेनराइज़्ड सॉफ्टवेयर डिप्लॉय करने जा रहे हैं, या बैकएंड को संभालने के लिए सर्वरलेस फंक्शन का उपयोग करेंगे? क्या आप अपने एप्लिकेशन के जटिल हिस्सों को संभालने के लिए थर्ड-पार्टी API का उपयोग करने की योजना बना रहे हैं, जैसे ऑथेंटिकेशन या पेमेंट्स? आप डेटा कहां स्टोर करेंगे?

इस सब के अलावा, आपको यूजर इंटरफेस, डिज़ाइन और अपने एप्लिकेशन की उपयोगिता के बारे में भी सोचना होगा, और भी बहुत कुछ।

यही कारण है कि जटिल बड़े एप्लिकेशन के लिए एक बहुआयामी डेवलपमेंट टीम की आवश्यकता होती है जो ऐप बनाने के लिए सहयोग करती है।

फुल स्टैक एप्लिकेशन विकसित करना सीखने के सर्वोत्तम तरीकों में से एक है ऐसे प्रोजेक्ट बनाना जो एंड-टू-एंड डेवलपमेंट प्रक्रिया को कवर करते हैं। आप आर्किटेक्चर डिजाइन करने, API सेवाएं विकसित करने, यूजर इंटरफेस विकसित करने, और अंत में अपने एप्लिकेशन को डिप्लॉय करने की प्रक्रिया से गुजरेंगे।

इसलिए यह ट्यूटोरियल आपको इन अवधारणाओं को गहराई से सीखने में मदद करने के लिए एक AI चैटबॉट बनाने की प्रक्रिया के माध्यम से ले जाएगा।

हम जिन विषयों को कवर करेंगे उनमें शामिल हैं:

• Python, FastAPI, और WebSockets के साथ API कैसे बनाएं
• Redis के साथ रियल-टाइम सिस्टम कैसे बनाएं
• React के साथ चैट यूजर इंटरफेस कैसे बनाएं

महत्वपूर्ण नोट: यह एक इंटरमीडिएट फुल स्टैक सॉफ्टवेयर डेवलपमेंट प्रोजेक्ट है जिसमें बेसिक Python और JavaScript ज्ञान की आवश्यकता होती है।

मैंने प्रोजेक्ट को ध्यान से सेक्शन में विभाजित किया है ताकि आप आसानी से उस चरण का चयन कर सकें जो आपके लिए महत्वपूर्ण है यदि आप फुल एप्लिकेशन कोड नहीं करना चाहते हैं।

आप मेरे GitHub पर पूरा रिपॉजिटरी डाउनलोड कर सकते हैं।

विषय सूची

सेक्शन 1

• एप्लिकेशन आर्किटेक्चर
• डेवलपमेंट एनवायरनमेंट कैसे सेट करें

सेक्शन 2

• Python, FastAPI, और WebSockets के साथ चैट सर्वर कैसे बनाएं
  • Python एनवायरनमेंट कैसे सेट करें
  • FastAPI सर्वर सेटअप
  • API में रूट्स कैसे जोड़ें
  • UUID के साथ चैट सेशन टोकन कैसे जनरेट करें
  • Postman के साथ API कैसे टेस्ट करें
  • WebSockets और कनेक्शन मैनेजर
  • FastAPI में डिपेंडेंसी इंजेक्शन

सेक्शन 3

• Redis के साथ रियल-टाइम सिस्टम कैसे बनाएं
  • Redis और डिस्ट्रिब्यूटेड मैसेजिंग क्यू
  • Redis क्लाइंट के साथ Python में Redis क्लस्टर से कैसे कनेक्ट करें
  • Redis Streams के साथ कैसे काम करें
  • चैट डेटा को कैसे मॉडल करें
  • Redis JSON के साथ कैसे काम करें
  • टोकन डिपेंडेंसी कैसे अपडेट करें

सेक्शन 4

• AI मॉडल के साथ चैटबॉट में इंटेलिजेंस कैसे जोड़ें
  • Huggingface के साथ कैसे शुरू करें
  • लैंग्वेज मॉडल के साथ कैसे इंटरैक्ट करें
  • AI मॉडल के लिए शॉर्ट-टर्म मेमोरी कैसे सिमुलेट करें
  • स्ट्रीम कंज्यूमर और मैसेज क्यू से रियल-टाइम डेटा पुल
  • AI प्रतिक्रिया के साथ चैट क्लाइंट कैसे अपडेट करें
  • रिफ्रेश टोकन
  • Postman में मल्टीपल क्लाइंट्स के साथ चैट कैसे टेस्ट करें

एप्लिकेशन आर्किटेक्चर

एक समाधान आर्किटेक्चर को स्केच करने से आपको अपने एप्लिकेशन का हाई-लेवल ओवरव्यू मिलता है, जिन टूल्स का आप उपयोग करने का इरादा रखते हैं, और घटक एक दूसरे के साथ कैसे संवाद करेंगे।

मैंने draw.io का उपयोग करके नीचे एक सरल आर्किटेक्चर बनाया है:

फुलस्टैक चैटबॉट आर्किटेक्चर

आइए आर्किटेक्चर के विभिन्न भागों को विस्तार से जानें:

क्लाइंट/यूजर इंटरफेस

हम यूजर इंटरफेस बनाने के लिए React वर्जन 18 का उपयोग करेंगे। चैट UI WebSockets के माध्यम से बैकएंड के साथ संवाद करेगा।

GPT-J-6B और Huggingface Inference API

GPT-J-6B एक जेनेरेटिव लैंग्वेज मॉडल है जिसे 6 बिलियन पैरामीटर्स के साथ प्रशिक्षित किया गया था और कुछ कार्यों पर OpenAI के GPT-3 के करीब प्रदर्शन करता है।

मैंने GPT-J-6B का उपयोग करना चुना क्योंकि यह एक ओपन-सोर्स मॉडल है और सरल उपयोग के मामलों के लिए पेड टोकन की आवश्यकता नहीं होती है।

Huggingface हमें इस मॉडल के साथ जुड़ने के लिए एक ऑन-डिमांड API भी प्रदान करता है जो बहुत हद तक मुफ़्त है। आप GPT-J-6B और Hugging Face Inference API के बारे में अधिक पढ़ सकते हैं।

Redis

जब हम GPT को प्रॉम्प्ट भेजते हैं, तो हमें प्रॉम्प्ट को स्टोर करने और प्रतिक्रिया को आसानी से पुनः प्राप्त करने का एक तरीका चाहिए। हम चैट डेटा को स्टोर करने के लिए Redis JSON का उपयोग करेंगे और Huggingface Inference API के साथ रियल-टाइम संचार को संभालने के लिए Redis Streams का उपयोग करेंगे।

Redis एक इन-मेमोरी की-वैल्यू स्टोर है जो JSON-जैसे डेटा की सुपर-फास्ट फेचिंग और स्टोरिंग को सक्षम बनाता है। इस ट्यूटोरियल के लिए, हम टेस्टिंग उद्देश्यों के लिए Redis Enterprise द्वारा प्रदान किए गए एक मैनेज्ड फ्री Redis स्टोरेज का उपयोग करेंगे।

WebSockets और चैट API

क्लाइंट और सर्वर के बीच रियल-टाइम में संदेश भेजने के लिए, हमें एक सॉकेट कनेक्शन खोलने की आवश्यकता है। ऐसा इसलिए है क्योंकि एक HTTP कनेक्शन क्लाइंट और सर्वर के बीच रियल-टाइम द्विदिश संचार सुनिश्चित करने के लिए पर्याप्त नहीं होगा।

हम चैट सर्वर के लिए FastAPI का उपयोग करेंगे, क्योंकि यह हमारे उपयोग के लिए एक तेज़ और आधुनिक Python सर्वर प्रदान करता है। WebSockets के बारे में अधिक जानने के लिए FastAPI डॉक्यूमेंटेशन देखें।

डेवलपमेंट एनवायरनमेंट कैसे सेट करें

आप इस ऐप को बनाने के लिए अपने वांछित OS का उपयोग कर सकते हैं - मैं वर्तमान में macOS और Visual Studio Code का उपयोग कर रहा हूं। बस सुनिश्चित करें कि Python और NodeJs इंस्टॉल हैं।

प्रोजेक्ट संरचना सेट करने के लिए, fullstack-ai-chatbot नामक एक फोल्डर बनाएं। फिर प्रोजेक्ट के भीतर client और server नामक दो फोल्डर बनाएं। सर्वर में बैकएंड का कोड होगा, जबकि क्लाइंट में फ्रंटएंड का कोड होगा।

इसके बाद प्रोजेक्ट डायरेक्टरी के भीतर, प्रोजेक्ट फोल्डर के रूट में "git init" कमांड का उपयोग करके एक Git रिपॉजिटरी इनिशियलाइज़ करें। फिर "touch .gitignore" का उपयोग करके एक .gitignore फाइल बनाएं:

git init
touch .gitignore

अगले सेक्शन में, हम FastAPI और Python का उपयोग करके अपना चैट वेब सर्वर बनाएंगे।

Python, FastAPI और WebSockets के साथ चैट सर्वर कैसे बनाएं

इस सेक्शन में, हम FastAPI का उपयोग करके चैट सर्वर बनाएंगे जो यूजर के साथ संवाद करेगा। हम WebSockets का उपयोग करेंगे ताकि क्लाइंट और सर्वर के बीच द्विदिश संचार सुनिश्चित हो सके ताकि हम यूजर को रियल-टाइम में प्रतिक्रियाएं भेज सकें।

Python एनवायरनमेंट कैसे सेट करें

अपना सर्वर शुरू करने के लिए, हमें अपना Python एनवायरनमेंट सेट करना होगा। VS Code में प्रोजेक्ट फोल्डर खोलें, और टर्मिनल खोलें।

प्रोजेक्ट रूट से, सर्वर डायरेक्टरी में cd करें और python3.8 -m venv env चलाएं। यह हमारे Python प्रोजेक्ट के लिए एक वर्चुअल एनवायरनमेंट बनाएगा, जिसका नाम env होगा। वर्चुअल एनवायरनमेंट को सक्रिय करने के लिए, source env/bin/activate चलाएं।

इसके बाद, अपने Python एनवायरनमेंट में कुछ लाइब्रेरी इंस्टॉल करें।

pip install fastapi uuid uvicorn gunicorn websockets python-dotenv aioredis

अगला, टर्मिनल में touch .env चलाकर एक एनवायरनमेंट फाइल बनाएं। हम .env फाइल के भीतर अपने ऐप वेरिएबल्स और सीक्रेट वेरिएबल्स को परिभाषित करेंगे।

अपना ऐप एनवायरनमेंट वेरिएबल जोड़ें और इसे "development" पर सेट करें जैसे: export APP_ENV=development। अगला, हम FastAPI सर्वर के साथ एक डेवलपमेंट सर्वर सेट करेंगे।

FastAPI सर्वर सेटअप

सर्वर डायरेक्टरी के रूट पर, main.py नामक एक नई फाइल बनाएं फिर डेवलपमेंट सर्वर के लिए नीचे दिया गया कोड पेस्ट करें:

from fastapi import FastAPI, Request
import uvicorn
import 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)
    else:
        pass

सबसे पहले हम FastAPI इम्पोर्ट करते हैं और इसे api के रूप में इनिशियलाइज़ करते हैं। फिर हम python-dotenv लाइब्रेरी से load_dotenv इम्पोर्ट करते हैं, और इसे .env फाइल से वेरिएबल लोड करने के लिए इनिशियलाइज़ करते हैं।

फिर हम API को टेस्ट करने के लिए एक साधारण टेस्ट रूट बनाते हैं। टेस्ट रूट एक साधारण JSON प्रतिक्रिया लौटाएगा जो हमें बताता है कि API ऑनलाइन है।

अंत में, हम uvicorn.run का उपयोग करके और आवश्यक आर्गुमेंट्स प्रदान करके डेवलपमेंट सर्वर सेट करते हैं। API पोर्ट 3500 पर चलेगा।

अंत में, टर्मिनल में python main.py के साथ सर्वर चलाएं। एक बार जब आप टर्मिनल में Application startup complete देखते हैं, तो अपने ब्राउज़र पर URL https://localhost:3500/test पर नेविगेट करें, और आपको इस तरह एक वेब पेज मिलना चाहिए:

API टेस्ट पेज

API में रूट्स कैसे जोड़ें

इस सेक्शन में, हम अपने API में रूट्स जोड़ेंगे। src नामक एक नया फोल्डर बनाएं। यह वह डायरेक्टरी है जहां हमारा सारा API कोड रहेगा।

routes नामक एक सबफोल्डर बनाएं, फोल्डर में cd करें, chat.py नामक एक नई फाइल बनाएं और फिर नीचे दिया गया कोड जोड़ें:

import os
from fastapi import APIRouter, FastAPI, WebSocket, Request

chat = APIRouter()

# @route POST /token
# @desc चैट टोकन जनरेट करने का रूट
# @access Public
@chat.post("/token")
async def token_generator(request: Request):
    return None

# @route POST /refresh_token
# @desc टोकन रिफ्रेश करने का रूट
# @access Public
@chat.post("/refresh_token")
async def refresh_token(request: Request):
    return None

# @route Websocket /chat
# @desc चैटबॉट के लिए सॉकेट
# @access Public
@chat.websocket("/chat")
async def websocket_endpoint(websocket: WebSocket = WebSocket):
    return None

हमने तीन एंडपॉइंट बनाए:

• /token यूजर को चैट सेशन तक पहुंच के लिए एक सेशन टोकन जारी करेगा। चूंकि चैट ऐप सार्वजनिक रूप से खुला होगा, हम ऑथेंटिकेशन के बारे में चिंता नहीं करना चाहते और इसे सरल रखना चाहते हैं - लेकिन हमें अभी भी प्रत्येक यूनिक यूजर सेशन की पहचान करने का एक तरीका चाहिए।
• /refresh_token यूजर के लिए सेशन हिस्ट्री प्राप्त करेगा यदि कनेक्शन खो जाता है, जब तक टोकन अभी भी सक्रिय है और एक्सपायर नहीं हुआ है।
• /chat क्लाइंट और सर्वर के बीच संदेश भेजने के लिए एक WebSocket खोलेगा।

अब, चैट रूट को हमारे मुख्य API से कनेक्ट करें। सबसे पहले हमें अपने main.py फाइल के भीतर src.chat से chat इम्पोर्ट करने की आवश्यकता है। फिर हम include_router मेथड को इनिशियलाइज़्ड FastAPI क्लास पर कॉल करके और chat को आर्गुमेंट के रूप में पास करके राउटर को शामिल करेंगे।

अपने api.py कोड को नीचे दिखाए अनुसार अपडेट करें:

from fastapi import FastAPI, Request
import uvicorn
import os
from dotenv import load_dotenv
from routes.chat import chat

load_dotenv()
api = FastAPI()
api.include_router(chat)

@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)
    else:
        pass

UUID के साथ चैट सेशन टोकन कैसे जनरेट करें

यूजर टोकन जनरेट करने के लिए हम अपने चैट एंडपॉइंट के लिए डायनामिक रूट बनाने के लिए uuid4 का उपयोग करेंगे। चूंकि यह एक सार्वजनिक रूप से उपलब्ध एंडपॉइंट है, हमें JWTs और ऑथेंटिकेशन के बारे में विस्तार से जाने की आवश्यकता नहीं होगी।

यदि आपने शुरुआत में uuid इंस्टॉल नहीं किया था, तो pip install uuid चलाएं। अगला chat.py में, UUID इम्पोर्ट करें, और /token रूट को नीचे दिए गए कोड से अपडेट करें:

from fastapi import APIRouter, FastAPI, WebSocket, Request, BackgroundTasks, HTTPException
import uuid

# @route POST /token
# @desc चैट टोकन जनरेट करने का रूट
# @access Public
@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())
    data = {"name": name, "token": token}
    return data

ऊपर दिए गए कोड में, क्लाइंट अपना नाम प्रदान करता है, जो आवश्यक है। हम यह सुनिश्चित करने के लिए एक त्वरित जांच करते हैं कि नाम फ़ील्ड खाली नहीं है, फिर uuid4 का उपयोग करके एक टोकन जनरेट करते हैं।

सेशन डेटा नाम और टोकन के लिए एक साधारण डिक्शनरी है। अंततः हमें इस सेशन डेटा को पर्सिस्ट करने और टाइमआउट सेट करने की आवश्यकता होगी, लेकिन अभी के लिए हम इसे केवल क्लाइंट को लौटाते हैं।

Postman के साथ API कैसे टेस्ट करें

क्योंकि हम एक WebSocket एंडपॉइंट टेस्ट करेंगे, हमें Postman जैसे टूल का उपयोग करने की आवश्यकता है जो इसकी अनुमति देता है (क्योंकि FastAPI पर डिफ़ॉल्ट स्वैगर डॉक्स WebSockets को सपोर्ट नहीं करते)।

Postman में, अपने डेवलपमेंट एनवायरनमेंट के लिए एक कलेक्शन बनाएं और localhost:3500/token पर एक POST रिक्वेस्ट भेजें जिसमें नाम को क्वेरी पैरामीटर के रूप में निर्दिष्ट करें और इसे एक वैल्यू पास करें। आपको नीचे दिखाए अनुसार प्रतिक्रिया मिलनी चाहिए:

टोकन जनरेटर Postman

WebSockets और कनेक्शन मैनेजर

src रूट में, socket नामक एक नया फोल्डर बनाएं और connection.py नामक एक फाइल जोड़ें। इस फाइल में, हम वह क्लास परिभाषित करेंगे जो हमारे WebSockets से कनेक्शन को नियंत्रित करती है, और कनेक्ट और डिस्कनेक्ट करने के लिए सभी हेल्पर मेथड्स।

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)

ConnectionManager क्लास को active_connections एट्रिब्यूट के साथ इनिशियलाइज़ किया जाता है जो एक्टिव कनेक्शन की एक सूची है।

फिर एसिंक्रोनस connect मेथड एक WebSocket स्वीकार करेगा और इसे एक्टिव कनेक्शन की सूची में जोड़ देगा, जबकि disconnect मेथड Websocket को एक्टिव कनेक्शन की सूची से हटा देगा।

अंत में, send_personal_message मेथड एक संदेश और वह Websocket लेगा जिसे हम संदेश भेजना चाहते हैं और एसिंक्रोनस रूप से संदेश भेजेगा।

WebSockets एक बहुत व्यापक विषय है और हमने यहां केवल सतह को खरोंचा है। हालांकि यह कई कनेक्शन बनाने और उन कनेक्शन को संदेशों को एसिंक्रोनस रूप से संभालने के लिए पर्याप्त होना चाहिए।

आप FastAPI WebSockets और सॉकेट प्रोग्रामिंग के बारे में अधिक पढ़ सकते हैं।

ConnectionManager का उपयोग करने के लिए, इसे src.routes.chat.py के भीतर इम्पोर्ट और इनिशियलाइज़ करें, और /chat WebSocket रूट को नीचे दिए गए कोड से अपडेट करें:

from ..socket.connection import ConnectionManager

manager = ConnectionManager()

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

websocket_endpoint फंक्शन में, जो एक WebSocket लेता है, हम नए वेबसॉकेट को कनेक्शन मैनेजर में जोड़ते हैं और एक while True लूप चलाते हैं, ताकि सॉकेट खुला रहे। सिवाय जब सॉकेट डिस्कनेक्ट हो जाता है।

जब कनेक्शन खुला होता है, तो हम websocket.receive_text() के साथ क्लाइंट द्वारा भेजे गए किसी भी संदेश को प्राप्त करते हैं और अभी के लिए उन्हें टर्मिनल पर प्रिंट करते हैं।

फिर हम अभी के लिए क्लाइंट को एक हार्ड-कोडेड प्रतिक्रिया वापस भेजते हैं। अंततः क्लाइंट्स से प्राप्त संदेश को AI मॉडल को भेजा जाएगा, और क्लाइंट को वापस भेजी गई प्रतिक्रिया AI मॉडल से प्रतिक्रिया होगी।

Postman में, हम एक नया WebSocket रिक्वेस्ट बनाकर और WebSocket एंडपॉइंट localhost:3500/chat से कनेक्ट करके इस एंडपॉइंट को टेस्ट कर सकते हैं।

जब आप कनेक्ट पर क्लिक करते हैं, तो मैसेज पेन दिखाएगा कि API क्लाइंट URL से जुड़ा हुआ है, और एक सॉकेट खुला है।

इसे टेस्ट करने के लिए, चैट सर्वर को "Hello Bot" संदेश भेजें, और आपको तत्काल टेस्ट प्रतिक्रिया "Response: Simulating response from the GPT service" मिलनी चाहिए जैसा कि नीचे दिखाया गया है:

Postman चैट टेस्ट

FastAPI में डिपेंडेंसी इंजेक्शन

दो अलग-अलग क्लाइंट सेशन के बीच अंतर करने और चैट सेशन को सीमित करने के लिए, हम एक टाइम्ड टोकन का उपयोग करेंगे, जिसे WebSocket कनेक्शन के लिए क्वेरी पैरामीटर के रूप में पास किया जाएगा।

सॉकेट फोल्डर में, 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

get_token फंक्शन एक WebSocket और टोकन प्राप्त करता है, फिर जांचता है कि टोकन None या खाली है या नहीं।

यदि ऐसा है, तो फंक्शन एक पॉलिसी वायलेशन स्टेटस लौटाता है और यदि उपलब्ध है, तो फंक्शन बस टोकन लौटाता है। हम अंततः इस फंक्शन को अतिरिक्त टोकन वैलिडेशन के साथ विस्तारित करेंगे।

इस फंक्शन को उपभोग करने के लिए, हम इसे /chat रूट में इंजेक्ट करते हैं। FastAPI एक Depends क्लास प्रदान करता है ताकि आसानी से डिपेंडेंसी इंजेक्ट की जा सके, इसलिए हमें डेकोरेटर्स के साथ छेड़छाड़ नहीं करनी पड़ेगी।

/chat रूट को निम्नलिखित में अपडेट करें:

from ..socket.utils import get_token

@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()
            print(data)
            await manager.send_personal_message(f"Response: Simulating response from the GPT service", websocket)
    except WebSocketDisconnect:
        manager.disconnect(websocket)

अब जब आप Postman में /chat एंडपॉइंट से कनेक्ट करने का प्रयास करते हैं, तो आपको 403 त्रुटि मिलेगी। क्वेरी पैरामीटर के रूप में एक टोकन प्रदान करें और अभी के लिए टोकन को कोई भी वैल्यू दें। फिर आप पहले की तरह कनेक्ट करने में सक्षम होना चाहिए, केवल अब कनेक्शन के लिए एक टोकन की आवश्यकता होती है।

टोकन के साथ Postman चैट टेस्ट

इतनी दूर आने के लिए बधाई! आपका chat.py फाइल अब इस तरह दिखना चाहिए:

import os
from fastapi import APIRouter, FastAPI, WebSocket, WebSocketDisconnect, Request, Depends, HTTPException
import uuid
from ..socket.connection import ConnectionManager
from ..socket.utils import get_token

chat = APIRouter()
manager = ConnectionManager()

# @route POST /token
# @desc चैट टोकन जनरेट करने का रूट
# @access Public
@chat.post("/token")
async def token_generator(name: str, request: Request):
    token = str(uuid.uuid4())
    if name == "":
        raise HTTPException(status_code=400, detail={
            "loc": "name",
            "msg": "Enter a valid name"
        })
    data = {"name": name, "token": token}
    return data

# @route POST /refresh_token
# @desc टोकन रिफ्रेश करने का रूट
# @access Public
@chat.post("/refresh_token")
async def refresh_token(request: Request):
    return None

# @route Websocket /chat
# @desc चैटबॉट के लिए सॉकेट
# @access Public
@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()
            print(data)
            await manager.send_personal_message(f"Response: Simulating response from the GPT service", websocket)
    except WebSocketDisconnect:
        manager.disconnect(websocket)

इस ट्यूटोरियल के अगले भाग में, हम अपने एप्लिकेशन की स्थिति को संभालने और क्लाइंट और सर्वर के बीच डेटा पास करने पर ध्यान केंद्रित करेंगे।

Redis के साथ रियल-टाइम सिस्टम कैसे बनाएं

हमारा एप्लिकेशन वर्तमान में कोई स्थिति स्टोर नहीं करता है, और यूजर्स की पहचान करने या चैट डेटा को स्टोर और पुनः प्राप्त करने का कोई तरीका नहीं है। हम चैट सेशन के दौरान क्लाइंट को एक हार्ड-कोडेड प्रतिक्रिया भी लौटा रहे हैं।

ट्यूटोरियल के इस भाग में, हम निम्नलिखित को कवर करेंगे:

• Python में Redis क्लस्टर से कैसे कनेक्ट करें और Redis क्लाइंट सेट करें
• Redis JSON के साथ डेटा कैसे स्टोर और पुनः प्राप्त करें
• वेब सर्वर और वर्कर एनवायरनमेंट के बीच मैसेज क्यू के रूप में Redis Streams कैसे सेट करें

Redis और डिस्ट्रिब्यूटेड मैसेजिंग क्यू

Redis एक ओपन सोर्स इन-मेमोरी डेटा स्टोर है जिसे आप डेटाबेस, कैश, मैसेज ब्रोकर, और स्ट्रीमिंग इंजन के रूप में उपयोग कर सकते हैं। यह कई डेटा स्ट्रक्चर को सपोर्ट करता है और रियल-टाइम क्षमताओं वाले डिस्ट्रिब्यूटेड एप्लिकेशन के लिए एक आदर्श समाधान है।

Redis Enterprise Cloud Redis द्वारा प्रदान की जाने वाली एक फुली मैनेज्ड क्लाउड सर्विस है जो हमें इन्फ्रास्ट्रक्चर की चिंता किए बिना अनंत स्केल पर Redis क्लस्टर डिप्लॉय करने में मदद करती है।

हम इस ट्यूटोरियल के लिए एक फ्री Redis Enterprise Cloud इंस्टेंस का उपयोग करेंगे। आप यहां फ्री में Redis Cloud के साथ शुरुआत कर सकते हैं और Redis डेटाबेस और Redis Insight सेट करने के लिए इस ट्यूटोरियल को फॉलो कर सकते हैं, जो Redis के साथ इंटरैक्ट करने के लिए एक GUI है।

एक बार जब आप अपना Redis डेटाबेस सेट कर लेते हैं, तो प्रोजेक्ट रूट में (सर्वर फोल्डर के बाहर) worker नामक एक नया फोल्डर बनाएं।

हम अपने वर्कर एनवायरनमेंट को वेब सर्वर से अलग करेंगे ताकि जब क्लाइंट हमारे WebSocket पर एक संदेश भेजे, तो वेब सर्वर को थर्ड-पार्टी सर्विस पर रिक्वेस्ट को संभालना न पड़े। साथ ही, अन्य यूजर्स के लिए संसाधन मुक्त हो सकें।

इनफेरेंस API के साथ बैकग्राउंड संचार इस वर्कर सर्विस द्वारा Redis के माध्यम से संभाला जाता है।

सभी कनेक्टेड क्लाइंट्स से रिक्वेस्ट मैसेज क्यू (प्रोड्यूसर) में जोड़े जाते हैं, जबकि वर्कर संदेशों को उपभोग करता है, इनफेरेंस API पर रिक्वेस्ट भेजता है, और प्रतिक्रिया को एक रिस्पॉन्स क्यू में जोड़ता है।

एक बार जब API को प्रतिक्रिया मिल जाती है, तो यह इसे क्लाइंट को वापस भेज देता है।

प्रोड्यूसर और कंज्यूमर के बीच की यात्रा के दौरान, क्लाइंट कई संदेश भेज सकता है, और ये संदेश कतारबद्ध हो जाएंगे और क्रम में प्रतिक्रिया देंगे।

आदर्श रूप से, हम इस वर्कर को पूरी तरह से अलग सर्वर पर, अपने स्वयं के एनवायरनमेंट में चला सकते थे, लेकिन अभी के लिए, हम इसे अपने स्थानीय मशीन पर अपना Python एनवायरनमेंट बनाएंगे।

आप सोच रहे होंगे - हमें वर्कर की आवश्यकता क्यों है? एक ऐसे परिदृश्य की कल्पना करें जहां वेब सर्वर थर्ड-पार्टी सर्विस पर रिक्वेस्ट भी बनाता है। इसका मतलब है कि सॉकेट कनेक्शन के दौरान थर्ड पार्टी सर्विस से प्रतिक्रिया की प्रतीक्षा करते समय, सर्वर ब्लॉक हो जाता है और संसाधन तब तक बंधे रहते हैं जब तक API से प्रतिक्रिया प्राप्त नहीं हो जाती।

आप इसे एक रैंडम स्लीप time.sleep(10) बनाकर हार्ड-कोडेड प्रतिक्रिया भेजने से पहले, और एक नया संदेश भेजकर आज़मा सकते हैं। फिर एक नए Postman सेशन में एक अलग टोकन से कनेक्ट करने का प्रयास करें।

आप देखेंगे कि चैट सेशन तब तक कनेक्ट नहीं होगा जब तक रैंडम स्लीप टाइम आउट नहीं हो जाता।

जबकि हम अधिक प्रोडक्शन-केंद्रित सर्वर सेट-अप में एसिंक्रोनस तकनीकों और वर्कर पूल का उपयोग कर सकते हैं, जैसे-जैसे एक साथ यूजर्स की संख्या बढ़ती है, वह भी पर्याप्त नहीं होगा।

अंततः, हम Redis का उपयोग करके अपने चैट API और थर्ड-पार्टी API के बीच संचार को ब्रोकर करके वेब सर्वर संसाधनों को बांधने से बचना चाहते हैं।

अब एक नया टर्मिनल खोलें, वर्कर फोल्डर में cd करें, और भाग 1 में हमने जो किया था उसके समान एक नया Python वर्चुअल एनवायरनमेंट बनाएं और सक्रिय करें।

अगला, निम्नलिखित डिपेंडेंसी इंस्टॉल करें:

pip install aiohttp aioredis python-dotenv

Redis क्लाइंट के साथ Python में Redis क्लस्टर से कैसे कनेक्ट करें

हम Redis डेटाबेस से जुड़ने के लिए aioredis क्लाइंट का उपयोग करेंगे। हम Huggingface इनफेरेंस API पर रिक्वेस्ट भेजने के लिए requests लाइब्रेरी का भी उपयोग करेंगे।

.env, और main.py नामक दो फाइलें बनाएं। फिर src नामक एक फोल्डर बनाएं। साथ ही, redis नामक एक फोल्डर बनाएं और config.py नामक एक नई फाइल जोड़ें।

.env फाइल में, निम्नलिखित कोड जोड़ें - और सुनिश्चित करें कि आप अपने Redis क्लस्टर में प्रदान किए गए क्रेडेंशियल्स के साथ फ़ील्ड अपडेट करते हैं।

export REDIS_URL=<REDIS CLOUD में प्रदान किया गया REDIS URL>
export REDIS_USER=<REDIS CLOUD में REDIS USER>
export REDIS_PASSWORD=<REDIS CLOUD में डेटाबेस पासवर्ड>
export REDIS_HOST=<REDIS CLOUD में REDIS HOST>
export REDIS_PORT=<REDIS CLOUD में REDIS PORT>

config.py में नीचे Redis क्लास जोड़ें:

import os
from dotenv import load_dotenv
import aioredis

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}"

    async def create_connection(self):
        self.connection = aioredis.from_url(
            self.connection_url, db=0)
        return self.connection

हम एक Redis ऑब्जेक्ट बनाते हैं और एनवायरनमेंट वेरिएबल्स से आवश्यक पैरामीटर्स इनिशियलाइज़ करते हैं। फिर हम एक Redis कनेक्शन बनाने और aioredis मेथड from_url से प्राप्त कनेक्शन पूल को लौटाने के लिए एक एसिंक्रोनस मेथड create_connection बनाते हैं।

अगला, हम नीचे दिए गए कोड को चलाकर main.py में Redis कनेक्शन को टेस्ट करते हैं। यह एक नया Redis कनेक्शन पूल बनाएगा, एक साधारण की "key" सेट करेगा, और इसे एक स्ट्रिंग "value" असाइन करेगा।

from src.redis.config import Redis
import asyncio

async def main():
    redis = Redis()
    redis = await redis.create_connection()
    print(redis)
    await redis.set("key", "value")

if __name__ == "__main__":
    asyncio.run(main())

अब Redis Insight खोलें (यदि आपने इसे डाउनलोड और इंस्टॉल करने के लिए ट्यूटोरियल का पालन किया है) आपको ऐसा कुछ देखना चाहिए:

Redis Insight टेस्ट

Redis Streams के साथ कैसे काम करें

अब जब हमारा वर्कर एनवायरनमेंट सेट हो गया है, तो हम वेब सर्वर पर एक प्रोड्यूसर और वर्कर पर एक कंज्यूमर बना सकते हैं।

सबसे पहले, आइए सर्वर पर फिर से अपना Redis क्लास बनाएं। server.src में redis नामक एक फोल्डर बनाएं और दो फाइलें जोड़ें, config.py और producer.py।

config.py में, वर्कर एनवायरनमेंट के लिए हमने जो किया था वही कोड जोड़ें:

import os
from dotenv import load_dotenv
import aioredis

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}"

    async def create_connection(self):
        self.connection = aioredis.from_url(
            self.connection_url, db=0)
        return self.connection

.env फाइल में, Redis क्रेडेंशियल्स भी जोड़ें:

export REDIS_URL=<REDIS CLOUD में प्रदान किया गया REDIS URL>
export REDIS_USER=<REDIS CLOUD में REDIS USER>
export REDIS_PASSWORD=<REDIS CLOUD में डेटाबेस पासवर्ड>
export REDIS_HOST=<REDIS CLOUD में REDIS HOST>
export REDIS_PORT=<REDIS CLOUD में REDIS PORT>

अंत में, server.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)
            print(f"Message id {msg_id} added to {stream_channel} stream")
            return msg_id
        except Exception as e:
            print(f"Error sending msg to stream => {e}")

हमने एक Producer क्लास बनाई है जिसे Redis क्लाइंट के साथ इनिशियलाइज़ किया गया है। हम इस क्लाइंट का उपयोग add_to_stream मेथड के साथ स्ट्रीम में डेटा जोड़ने के लिए करते हैं, जो डेटा और Redis चैनल नाम लेता है।

स्ट्रीम चैनल में डेटा जोड़ने के लिए Redis कमांड xadd है और इसमें aioredis में हाई-लेवल और लो-लेवल दोनों फंक्शन हैं।

अगला, हमारे नए बने Producer को चलाने के लिए, chat.py और WebSocket /chat एंडपॉइंट को नीचे की तरह अपडेट करें। अपडेटेड चैनल नाम message_channel पर ध्यान दें।

from ..redis.producer import Producer
from ..redis.config import Redis

chat = APIRouter()
manager = ConnectionManager()
redis = Redis()

@chat.websocket("/chat")
async def websocket_endpoint(websocket: WebSocket, token: str = Depends(get_token)):
    await manager.connect(websocket)
    redis_client = await redis.create_connection()
    producer = Producer(redis_client)
    try:
        while True:
            data = await websocket.receive_text()
            print(data)
            stream_data = {}
            stream_data[token] = data
            await producer.add_to_stream(stream_data, "message_channel")
            await manager.send_personal_message(f"Response: Simulating response from the GPT service", websocket)
    except WebSocketDisconnect:
        manager.disconnect(websocket)

अगला, Postman में, एक कनेक्शन बनाएं और Hello कहने वाले कई संदेश भेजें। आपके पास टर्मिनल पर प्रिंटेड स्ट्रीम संदेश होने चाहिए जैसे:

टर्मिनल चैनल मैसेज टेस्ट

Redis Insight में, आप एक नया message_channel बने हुए देखेंगे और क्लाइंट से भेजे गए संदेशों से भरा एक टाइमस्टैम्प्ड क्यू देखेंगे। यह टाइमस्टैम्प्ड क्यू संदेशों के क्रम को बनाए रखने के लिए महत्वपूर्ण है।

Redis Insight चैनल

चैट डेटा को कैसे मॉडल करें

अगला, हम अपने चैट संदेशों के लिए एक मॉडल बनाएंगे। याद रखें कि हम WebSockets पर टेक्स्ट डेटा भेज रहे हैं, लेकिन हमारे चैट डेटा को केवल टेक्स्ट से अधिक जानकारी रखने की आवश्यकता है। हमें चैट कब भेजा गया था इसे टाइमस्टैम्प करने की आवश्यकता है, प्रत्येक संदेश के लिए एक ID बनाएं, और चैट सेशन के बारे में डेटा एकत्र करें, फिर इस डेटा को JSON फॉर्मेट में स्टोर करें।

हम इस JSON डेटा को Redis में स्टोर कर सकते हैं ताकि कनेक्शन खो जाने पर हम चैट हिस्ट्री न खोएं, क्योंकि हमारा WebSocket स्टेट स्टोर नहीं करता।

server.src में schema नामक एक नया फोल्डर बनाएं। फिर server.src.schema में chat.py नामक एक फाइल बनाएं और निम्नलिखित कोड जोड़ें:

from datetime import datetime
from pydantic import BaseModel
from typing import List, Optional
import uuid

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

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

हम चैट डेटा को मॉडल करने के लिए Pydantic के BaseModel क्लास का उपयोग कर रहे हैं। Chat क्लास एक सिंगल चैट सेशन के बारे में डेटा रखेगा। यह टोकन, यूजर का नाम, और datetime.now() का उपयोग करके चैट सेशन शुरू होने के समय के लिए ऑटोमैटिकली जनरेटेड टाइमस्टैम्प स्टोर करेगा।

इस चैट सेशन के भीतर भेजे और प्राप्त किए गए संदेश Message क्लास के साथ स्टोर किए जाते हैं जो uuid4 का उपयोग करके ऑन-द-फ्लाई एक चैट ID बनाता है। इस Message क्लास को इनिशियलाइज़ करते समय हमें केवल संदेश टेक्स्ट प्रदान करने की आवश्यकता होती है।

Redis JSON के साथ कैसे काम करें

हमारे चैट इतिहास को स्टोर करने के लिए Redis JSON की क्षमता का उपयोग करने के लिए, हमें Redis Labs द्वारा प्रदान किया गया rejson इंस्टॉल करना होगा।

टर्मिनल में, server में cd करें और pip install rejson के साथ rejson इंस्टॉल करें। फिर server.src.redis.config.py में अपने Redis क्लास को create_rejson_connection मेथड शामिल करने के लिए अपडेट करें:

import os
from dotenv import load_dotenv
import aioredis
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

हम create_rejson_connection मेथड जोड़ रहे हैं ताकि rejson Client के साथ Redis से कनेक्ट किया जा सके। यह हमें Redis में JSON डेटा बनाने और हेरफेर करने के मेथड देता है, जो aioredis के साथ उपलब्ध नहीं हैं।

अगला, server.src.routes.chat.py में हम /token एंडपॉइंट को एक नया Chat इंस्टेंस बनाने और सेशन डेटा को Redis JSON में स्टोर करने के लिए अपडेट कर सकते हैं:

@chat.post("/token")
async def token_generator(name: str, request: Request):
    token = str(uuid.uuid4())
    if name == "":
        raise HTTPException(status_code=400, detail={
            "loc": "name",
            "msg": "Enter a valid name"
        })
    # नया चैट सेशन बनाएं
    json_client = redis.create_rejson_connection()
    chat_session = Chat(
        token=token,
        messages=[],
        name=name
    )
    # टोकन को की के रूप में उपयोग करके चैट सेशन को Redis JSON में स्टोर करें
    json_client.jsonset(str(token), Path.rootPath(), chat_session.dict())
    # Redis डेटा के लिए टाइमआउट सेट करें
    redis_client = await redis.create_connection()
    await redis_client.expire(str(token), 3600)
    return chat_session.dict()

नोट: चूंकि यह एक डेमो ऐप है, मैं चैट डेटा को Redis में बहुत लंबे समय तक स्टोर नहीं करना चाहता। इसलिए मैंने aioredis क्लाइंट का उपयोग करके टोकन पर 60 मिनट का टाइमआउट जोड़ा है (rejson टाइमआउट लागू नहीं करता)। इसका मतलब है कि 60 मिनट के बाद, चैट सेशन डेटा खो जाएगा।

यह आवश्यक है क्योंकि हम यूजर्स को ऑथेंटिकेट नहीं कर रहे हैं, और हम एक परिभाषित अवधि के बाद चैट डेटा को डंप करना चाहते हैं। यह चरण वैकल्पिक है, और आपको इसे शामिल करने की आवश्यकता नहीं है।

अगला, Postman में, जब आप एक नया टोकन बनाने के लिए POST रिक्वेस्ट भेजते हैं, तो आपको नीचे दिखाए अनुसार एक संरचित प्रतिक्रिया मिलेगी। आप Redis Insight में अपने चैट डेटा को टोकन के साथ JSON की के रूप में और डेटा को वैल्यू के रूप में स्टोर देख सकते हैं।

अपडेटेड टोकन जनरेटर

टोकन डिपेंडेंसी कैसे अपडेट करें

अब जब हमारे पास एक टोकन जनरेट और स्टोर हो रहा है, तो यह हमारे /chat WebSocket में get_token डिपेंडेंसी को अपडेट करने का एक अच्छा समय है। हम ऐसा वैलिड टोकन की जांच करने के लिए करते हैं ताकि चैट सेशन शुरू करने से पहले।

server.src.socket.utils.py में get_token फंक्शन को अपडेट करें ताकि यह जांच सके कि टोकन Redis इंस्टेंस में मौजूद है या नहीं। यदि यह मौजूद है तो हम टोकन लौटाते हैं, जिसका अर्थ है कि सॉकेट कनेक्शन वैलिड है। यदि यह मौजूद नहीं है, तो हम कनेक्शन बंद कर देते हैं।

/token द्वारा बनाया गया टोकन 60 मिनट के बाद मौजूद नहीं रहेगा। इसलिए हम फ्रंटएंड पर कुछ सरल लॉजिक रख सकते हैं ताकि यदि चैट शुरू करने का प्रयास करते समय एक त्रुटि प्रतिक्रिया उत्पन्न होती है तो यूजर को एक नया टोकन जनरेट करने के लिए रीडायरेक्ट किया जाए।

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 not authenticated or expired token")

डिपेंडेंसी को टेस्ट करने के लिए, उस रैंडम टोकन के साथ चैट सेशन से कनेक्ट करें जिसका हम उपयोग कर रहे हैं, और आपको 403 त्रुटि मिलनी चाहिए। (ध्यान दें कि आपको Redis Insight में मैन्युअली टोकन डिलीट करना होगा।)

अब उस टोकन को कॉपी करें जो तब जनरेट हुआ था जब आपने /token एंडपॉइंट पर पोस्ट रिक्वेस्ट भेजी थी (या एक नई रिक्वेस्ट बनाएं) और इसे /chat WebSocket द्वारा आवश्यक टोकन क्वेरी पैरामीटर के वैल्यू के रूप में पेस्ट करें। फिर कनेक्ट करें। आपको एक सफल कनेक्शन मिलना चाहिए।

टोकन के साथ चैट सेशन

इसे सभी को एक साथ लाते हुए, आपका chat.py नीचे दिए गए जैसा दिखना चाहिए।

import os
from fastapi import APIRouter, FastAPI, WebSocket, WebSocketDisconnect, Request, Depends
import uuid
from ..socket.connection import ConnectionManager
from ..socket.utils import get_token
import time
from ..redis.producer import Producer
from ..redis.config import Redis
from ..schema.chat import Chat
from rejson import Path

chat = APIRouter()
manager = ConnectionManager()
redis = Redis()

# @route POST /token
# @desc चैट टोकन जनरेट करने का रूट
# @access Public
@chat.post("/token")
async def token_generator(name: str, request: Request):
    token = str(uuid.uuid4())
    if name == "":
        raise HTTPException(status_code=400, detail={
            "loc": "name",
            "msg": "Enter a valid name"
        })
    # नया चैट सेशन बनाएं
    json_client = redis.create_rejson_connection()
    chat_session = Chat(
        token=token,
        messages=[],
        name=name
    )
    print(chat_session.dict())
    # टोकन को की के रूप में उपयोग करके चैट सेशन को Redis JSON में स्टोर करें
    json_client.jsonset(str(token), Path.rootPath(), chat_session.dict())
    # Redis डेटा के लिए टाइमआउट सेट करें
    redis_client = await redis.create_connection()
    await redis_client.expire(str(token), 3600)
    return chat_session.dict()

# @route POST /refresh_token
# @desc टोकन रिफ्रेश करने का रूट
# @access Public
@chat.post("/refresh_token")
async def refresh_token(request: Request):
    return None

# @route Websocket /chat
# @desc चैट बॉट के लिए सॉकेट
# @access Public
@chat.websocket("/chat")
async def websocket_endpoint(websocket: WebSocket, token: str = Depends(get_token)):
    await manager.connect(websocket)
    redis_client = await redis.create_connection()
    producer = Producer(redis_client)
    json_client = redis.create_rejson_connection()
    try:
        while True:
            data = await websocket.receive_text()
            stream_data = {}
            stream_data[token] = data
            await producer.add_to_stream(stream_data, "message_channel")
            await manager.send_personal_message(f"Response: Simulating response from the GPT service", websocket)
    except WebSocketDisconnect:
        manager.disconnect(websocket)

इतनी दूर पहुंचने पर बहुत बढ़िया! अगले सेक्शन में, हम AI मॉडल के साथ संवाद करने और क्लाइंट, सर्वर, वर्कर, और एक्सटर्नल API के बीच डेटा ट्रांसफर को संभालने पर ध्यान केंद्रित करेंगे।

AI मॉडल के साथ चैटबॉट में इंटेलिजेंस कैसे जोड़ें

इस सेक्शन में, हम ट्रांसफॉर्मर मॉडल के साथ संवाद करने के लिए एक रैपर बनाने, यूजर से प्रॉम्प्ट को बातचीत के प्रारूप में API को भेजने, और अपने चैट एप्लिकेशन के लिए प्रतिक्रियाओं को प्राप्त करने और बदलने पर ध्यान केंद्रित करेंगे।

Huggingface के साथ कैसे शुरू करें

हम Huggingface पर कोई लैंग्वेज मॉडल बनाने या डिप्लॉय नहीं करेंगे। इसके बजाय, हम Huggingface के त्वरित इनफेरेंस API का उपयोग करके प्री-ट्रेंड मॉडल से जुड़ने पर ध्यान केंद्रित करेंगे।

हम जिस मॉडल का उपयोग करेंगे वह EleutherAI द्वारा प्रदान किया गया GPT-J-6B मॉडल है। यह एक जेनेरेटिव लैंग्वेज मॉडल है जिसे 6 बिलियन पैरामीटर्स के साथ प्रशिक्षित किया गया था।

Huggingface हमें इस मॉडल के साथ जुड़ने के लिए एक ऑन-डिमांड लिमिटेड API प्रदान करता है जो बहुत हद तक मुफ़्त है।

Huggingface के साथ शुरू करने के लिए, एक मुफ़्त खाता बनाएं। अपनी सेटिंग्स में, एक नया एक्सेस टोकन जनरेट करें। 30k टोकन तक, Huggingface इनफेरेंस API तक मुफ्त पहुंच प्रदान करता है।

आप यहां अपने API उपयोग की निगरानी कर सकते हैं। सुनिश्चित करें कि आप इस टोकन को सुरक्षित रखें और इसे सार्वजनिक रूप से उजागर न करें।

नोट: हम API के साथ संवाद करने के लिए HTTP कनेक्शन का उपयोग करेंगे क्योंकि हम एक फ्री अकाउंट का उपयोग कर रहे हैं। लेकिन PRO Huggingface अकाउंट WebSockets के साथ स्ट्रीमिंग को सपोर्ट करता है, समानांतरता और बैच जॉब्स देखें।

यह हमारे चैट एप्लिकेशन और मॉडल के बीच प्रतिक्रिया समय को काफी सुधारने में मदद कर सकता है, और मैं उम्मीद करता हूं कि इस विधि को एक फॉलो-अप लेख में कवर करूंगा।

लैंग्वेज मॉडल के साथ कैसे इंटरैक्ट करें

सबसे पहले, हम अपने वर्कर डायरेक्टरी के भीतर .env फाइल में Huggingface कनेक्शन क्रेडेंशियल्स जोड़ते हैं।

export HUGGINFACE_INFERENCE_TOKEN=<HUGGINGFACE एक्सेस टोकन>
export MODEL_URL=https://api-inference.huggingface.co/models/EleutherAI/gpt-j-6B

अगला, worker.src में model नामक एक फोल्डर बनाएं फिर एक फाइल gptj.py जोड़ें। फिर नीचे GPT क्लास जोड़ें:

import os
from dotenv import load_dotenv
import requests
import json

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) -> list:
        self.payload["inputs"] = input
        data = json.dumps(self.payload)
        response = requests.request(
            "POST", self.url, headers=self.headers, data=data)
        print(json.loads(response.content.decode("utf-8")))
        return json.loads(response.content.decode("utf-8"))

if __name__ == "__main__":
    GPT().query("Will artificial intelligence help humanity conquer the universe?")

GPT क्लास को Huggingface मॉडल url, ऑथेंटिकेशन header, और प्रीडिफाइन्ड payload के साथ इनिशियलाइज़ किया जाता है। लेकिन पेलोड इनपुट एक डायनामिक फ़ील्ड है जो query मेथड द्वारा प्रदान किया जाता है और Huggingface एंडपॉइंट पर रिक्वेस्ट भेजने से पहले अपडेट किया जाता है।

अंत में, हम GPT क्लास के इंस्टेंस पर query मेथड चलाकर सीधे इसका टेस्ट करते हैं। टर्मिनल में, python src/model/gptj.py चलाएं, और आपको नीचे जैसी प्रतिक्रिया मिलनी चाहिए (बस ध्यान रखें कि आपकी प्रतिक्रिया निश्चित रूप से इससे अलग होगी):

[{'generated_text': ' (AI) could solve all the problems on this planet? I am of the opinion that in the short term artificial intelligence is much better than human beings, but in the long and distant future human beings will surpass artificial intelligence.\n\nIn the distant'}]

अगला, हम इनपुट में कुछ बदलाव करते हैं ताकि मॉडल के साथ इंटरैक्शन को इनपुट के प्रारूप को बदलकर अधिक संवादात्मक बनाया जा सके।

GPT क्लास को इस तरह अपडेट करें:

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": False,
                "max_new_tokens": 25
            }
        }

    def query(self, input: str) -> list:
        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']
        res = str(text.split("Human:")[0]).strip("\n").strip()
        return res

if __name__ == "__main__":
    GPT().query("Will artificial intelligence help humanity conquer the universe?")

हमने इनपुट को एक स्ट्रिंग लिटरल f"Human: {input} Bot:" के साथ अपडेट किया। मानव इनपुट को स्ट्रिंग में रखा जाता है और बॉट एक प्रतिक्रिया प्रदान करता है। यह इनपुट प्रारूप GPT-J6B को एक संवादात्मक मॉडल में बदल देता है। अन्य परिवर्तन जिन पर आप ध्यान दे सकते हैं उनमें शामिल हैं:

• use_cache: आप इसे False कर सकते हैं यदि आप चाहते हैं कि मॉडल वही इनपुट होने पर एक नई प्रतिक्रिया बनाए। मैं प्रोडक्शन में इसे True छोड़ने का सुझाव देता हूं ताकि यदि कोई यूजर बॉट को बार-बार एक ही संदेश से स्पैम करता है तो आपके फ्री टोकन समाप्त न हों। कैश का उपयोग करने से वास्तव में मॉडल से एक नई प्रतिक्रिया लोड नहीं होती।
• return_full_text: False है, क्योंकि हमें इनपुट वापस करने की आवश्यकता नहीं है - हमारे पास पहले से ही है। जब हमें प्रतिक्रिया मिलती है, तो हम प्रतिक्रिया से "Bot:" और अग्रणी/अनुगामी स्पेस को हटाते हैं और केवल प्रतिक्रिया टेक्स्ट लौटाते हैं।

AI मॉडल के लिए शॉर्ट-टर्म मेमोरी कैसे सिमुलेट करें

हम मॉडल को भेजे जाने वाले प्रत्येक नए इनपुट के लिए, मॉडल के लिए बातचीत के इतिहास को याद रखने का कोई तरीका नहीं है। यह महत्वपूर्ण है यदि हम बातचीत में संदर्भ रखना चाहते हैं।

लेकिन याद रखें कि जैसे-जैसे हम मॉडल को भेजे जाने वाले टोकन की संख्या बढ़ती है, प्रोसेसिंग अधिक महंगी हो जाती है, और प्रतिक्रिया समय भी लंबा होता है।

इसलिए हमें शॉर्ट-टर्म हिस्ट्री को पुनः प्राप्त करने और इसे मॉडल को भेजने का एक तरीका खोजने की आवश्यकता होगी। हमें एक स्वीट स्पॉट भी निकालना होगा - हम मॉडल को कितना ऐतिहासिक डेटा पुनः प्राप्त करना और भेजना चाहते हैं?

चैट इतिहास को संभालने के लिए, हमें अपने JSON डेटाबेस पर वापस गिरने की आवश्यकता है। हम token का उपयोग करके अंतिम चैट डेटा प्राप्त करेंगे, और फिर जब हमें प्रतिक्रिया मिलती है, तो प्रतिक्रिया को JSON डेटाबेस में जोड़ दें।

worker.src.redis.config.py को create_rejson_connection मेथड शामिल करने के लिए अपडेट करें। साथ ही, ऑथेंटिकेशन डेटा के साथ .env फाइल अपडेट करें, और सुनिश्चित करें कि rejson इंस्टॉल है।

आपका worker.src.redis.config.py इस तरह दिखना चाहिए:

import os
from dotenv import load_dotenv
import aioredis
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

जबकि आपकी .env फाइल इस तरह दिखनी चाहिए:

export REDIS_URL=<REDIS CLOUD में प्रदान किया गया REDIS URL>
export REDIS_USER=<REDIS CLOUD में REDIS USER>
export REDIS_PASSWORD=<REDIS CLOUD में डेटाबेस पासवर्ड>
export REDIS_HOST=<REDIS CLOUD में REDIS HOST>
export REDIS_PORT=<REDIS CLOUD में REDIS PORT>
export HUGGINFACE_INFERENCE_TOKEN=<HUGGINGFACE एक्सेस टोकन>
export MODEL_URL=https://api-inference.huggingface.co/models/EleutherAI/gpt-j-6B

अगला, 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):
        data = self.json_client.jsonget(
            str(token), Path.rootPath())
        return data

कैश को एक rejson क्लाइंट के साथ इनिशियलाइज़ किया जाता है, और मेथड get_chat_history उस टोकन के लिए Redis से चैट इतिहास प्राप्त करने के लिए एक टोकन लेता है। सुनिश्चित करें कि आप rejson से Path ऑब्जेक्ट इम्पोर्ट करते हैं।

अगला, worker.main.py को नीचे दिए गए कोड से अपडेट करें:

from src.redis.config import Redis
import asyncio
from src.model.gptj import GPT
from src.redis.cache import Cache

redis = Redis()

async def main():
    json_client = redis.create_rejson_connection()
    data = await Cache(json_client).get_chat_history(token="18196e23-763b-4808-ae84-064348a0daff")
    print(data)

if __name__ == "__main__":
    asyncio.run(main())

मैंने Postman में पिछले टेस्ट से बने एक सैंपल टोकन को हार्ड-कोड किया है। यदि आपके पास कोई टोकन नहीं बना है, तो बस /token पर एक नई रिक्वेस्ट भेजें और टोकन कॉपी करें, फिर टर्मिनल में python main.py चलाएं। आपको टर्मिनल में डेटा ऐसा दिखना चाहिए:

{'token': '18196e23-763b-4808-ae84-064348a0daff', 'messages': [], 'name': 'Stephen', 'session_start': '2022-07-16 13:20:01.092109'}

अगला, हमें अपने Cache क्लास में एक add_message_to_cache मेथड जोड़ने की आवश्यकता है जो एक विशिष्ट टोकन के लिए Redis में संदेश जोड़ता है।

async def add_message_to_cache(self, token: str, message_data: dict):
    self.json_client.jsonarrappend(
        str(token), Path('.messages'), message_data)

jsonarrappend मेथड जो rejson द्वारा प्रदान किया जाता है, नए संदेश को मैसेज ऐरे में जोड़ता है।

नोट करें कि मैसेज ऐरे तक पहुंचने के लिए, हमें Path के आर्गुमेंट के रूप में .messages प्रदान करने की आवश्यकता है। यदि आपके मैसेज डेटा की एक अलग/नेस्टेड संरचना है, तो बस उस ऐरे का पथ प्रदान करें जिसमें आप नया डेटा जोड़ना चाहते हैं।

इस मेथड को टेस्ट करने के लिए, main.py फाइल में main फंक्शन को नीचे दिए गए कोड से अपडेट करें:

async def main():
    json_client = redis.create_rejson_connection()
    await Cache(json_client).add_message_to_cache(token="18196e23-763b-4808-ae84-064348a0daff", message_data={
        "id": "1",
        "msg": "Hello",
        "timestamp": "2022-07-16 13:20:01.092109"
    })
    data = await Cache(json_client).get_chat_history(token="18196e23-763b-4808-ae84-064348a0daff")
    print(data)

हम कैश को एक हार्ड-कोडेड संदेश भेज रहे हैं, और कैश से चैट इतिहास प्राप्त कर रहे हैं। जब आप वर्कर डायरेक्टरी के भीतर टर्मिनल में python main.py चलाते हैं, तो आपको टर्मिनल में ऐसा कुछ प्रिंट होना चाहिए, जिसमें संदेश मैसेज ऐरे में जोड़ा गया हो।

{'token': '18196e23-763b-4808-ae84-064348a0daff', 'messages': [{'id': '1', 'msg': 'Hello', 'timestamp': '2022-07-16 13:20:01.092109'}], 'name': 'Stephen', 'session_start': '2022-07-16 13:20:01.092109'}

अंत में, हमें main फंक्शन को संदेश डेटा को GPT मॉडल को भेजने के लिए अपडेट करने की आवश्यकता है, और अंतिम 4 संदेशों के साथ इनपुट को अपडेट करें जो क्लाइंट और मॉडल के बीच भेजे गए थे।

सबसे पहले आइए अपने add_message_to_cache फंक्शन को एक नए आर्गुमेंट "source" के साथ अपडेट करें जो हमें बताएगा कि संदेश मानव है या बॉट। हम तब इस आर्गु का उपयोग डेटा को कैश में स्टोर करने से पहले "Human:" या "Bot:" टैग जोड़ने के लिए कर सकते हैं।

Cache क्लास में add_message_to_cache मेथड को इस तरह अपडेट करें:

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)

फिर वर्कर डायरेक्टरी में main.py में main फंक्शन को अपडेट करें, और Redis डेटाबेस में नए परिणाम देखने के लिए python main.py चलाएं।

async def main():
    json_client = redis.create_rejson_connection()
    await Cache(json_client).add_message_to_cache(token="18196e23-763b-4808-ae84-064348a0daff", source="human", message_data={
        "id": "1",
        "msg": "Hello",
        "timestamp": "2022-07-16 13:20:01.092109"
    })
    data = await Cache(json_client).get_chat_history(token="18196e23-763b-4808-ae84-064348a0daff")
    print(data)

अगला, हमें main फंक्शन को कैश में नए संदेश जोड़ने, कैश से पिछले 4 संदेश पढ़ने, और फिर query मेथड का उपयोग करके मॉडल पर API कॉल करने के लिए अपडेट करने की आवश्यकता है। इसमें अंतिम 4 संदेशों की एक संयुक्त स्ट्रिंग वाला पेलोड होगा।

आप हमेशा उस इतिहास में संदेशों की संख्या को ट्यून कर सकते हैं जिसे आप निकालना चाहते हैं, लेकिन मुझे लगता है कि डेमो के लिए 4 संदेश एक बहुत अच्छी संख्या है।

worker.src में, schema नामक एक नया फोल्डर बनाएं। फिर chat.py नामक एक नई फाइल बनाएं और chat.py में हमारे मैसेज स्कीमा को इस तरह पेस्ट करें:

from datetime import datetime
from pydantic import BaseModel
from typing import List, Optional
import uuid

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

अगला, main.py फाइल को नीचे की तरह अपडेट करें:

async def main():
    json_client = redis.create_rejson_connection()
    await Cache(json_client).add_message_to_cache(token="18196e23-763b-4808-ae84-064348a0daff", source="human", message_data={
        "id": "3",
        "msg": "I would like to go to the moon to, would you take me?",
        "timestamp": "2022-07-16 13:20:01.092109"
    })
    data = await Cache(json_client).get_chat_history(token="18196e23-763b-4808-ae84-064348a0daff")
    print(data)
    message_data = data['messages'][-4:]
    input = ["" + i['msg'] for i in message_data]
    input = " ".join(input)
    res = GPT().query(input=input)
    msg = Message(
        msg=res
    )
    print(msg)
    await Cache(json_client).add_message_to_cache(token="18196e23-763b-4808-ae84-064348a0daff", source="bot", message_data=msg.dict())

ऊपर दिए गए कोड में, हम कैश में नया मैसेज डेटा जोड़ते हैं। यह संदेश अंततः मैसेज क्यू से आएगा। अगला हम कैश से चैट इतिहास प्राप्त करते हैं, जिसमें अब हमारे द्वारा जोड़ा गया सबसे हालिया डेटा शामिल होगा।

ध्यान दें कि हम कैश में जोड़ने और कैश से प्राप्त करने के लिए एक ही हार्ड-कोडेड टोकन का उपयोग कर रहे हैं, अस्थायी रूप से सिर्फ इसे टेस्ट करने के लिए।

अगला, हम कैश डेटा को ट्रिम करते हैं और केवल अंतिम 4 आइटम निकालते हैं। फिर हम एक लिस्ट में msg निकालकर और इसे एक खाली स्ट्रिंग से जोड़कर इनपुट डेटा को समेकित करते हैं।

अंत में, हम बॉट प्रतिक्रिया के लिए एक नया Message इंस्टेंस बनाते हैं और स्रोत को "bot" निर्दिष्ट करके प्रतिक्रिया को कैश में जोड़ते हैं।

अगला, python main.py को कुछ बार चलाएं, प्रत्येक रन के साथ वांछित अनुसार मानव संदेश और आईडी बदलते रहें। आपके पास मॉडल के साथ एक पूर्ण बातचीत इनपुट और आउटपुट होना चाहिए।

Redis Insight खोलें और आपके पास नीचे जैसा कुछ होना चाहिए:

संवादात्मक चैट

स्ट्रीम कंज्यूमर और मैसेज क्यू से रियल-टाइम डेटा पुल

अगला, हम एक कंज्यूमर बनाना चाहते हैं और अपने worker.main.py को मैसेज क्यू से कनेक्ट करने के लिए अपडेट करना चाहते हैं। हम चाहते हैं कि यह टोकन डेटा को रियल-टाइम में खींचे, क्योंकि हम वर्तमान में टोकन और मैसेज इनपुट को हार्ड-कोड कर रहे हैं।

worker.src.redis में stream.py नामक एक नई फाइल बनाएं। नीचे दिए गए कोड के साथ एक StreamConsumer क्लास जोड़ें:

class StreamConsumer:
    def __init__(self, redis_client):
        self.redis_client = redis_client

    async def consume_stream(self, count: int, block: int, stream_channel):
        response = await self.redis_client.xread(
            streams={stream_channel: '0-0'}, count=count, block=block)
        return response

    async def delete_message(self, stream_channel, message_id):
        await self.redis_client.xdel(stream_channel, message_id)

StreamConsumer क्लास को एक Redis क्लाइंट के साथ इनिशियलाइज़ किया जाता है। consume_stream मेथड मैसेज चैनल से क्यू से एक नया संदेश खींचता है, aioredis द्वारा प्रदान किए गए xread मेथड का उपयोग करके।

अगला, worker.main.py फाइल को एक व्हाइल लूप के साथ अपडेट करें ताकि मैसेज चैनल से कनेक्शन जीवित रहे, जैसे:

from src.redis.config import Redis
import asyncio
from src.model.gptj import GPT
from src.redis.cache import Cache
from src.redis.config import Redis
from src.redis.stream import StreamConsumer
import os
from src.schema.chat import Message

redis = Redis()

async def main():
    json_client = redis.create_rejson_connection()
    redis_client = await redis.create_connection()
    consumer = StreamConsumer(redis_client)
    cache = Cache(json_client)
    print("Stream consumer started")
    print("Stream waiting for new messages")
    while True:
        response = await consumer.consume_stream(stream_channel="message_channel", count=1, block=0)
        if response:
            for stream, messages in response:
                # स्ट्रीम से संदेश प्राप्त करें, और टोकन, मैसेज डेटा और मैसेज आईडी निकालें
                for message in messages:
                    message_id = message[0]
                    token = [k.decode('utf-8') for k, v in message[1].items()][0]
                    message = [v.decode('utf-8') for k, v in message[1].items()][0]
                    print(token)
                    # एक नया मैसेज इंस्टेंस बनाएं और कैश में जोड़ें, स्रोत को मानव के रूप में निर्दिष्ट करें
                    msg = Message(msg=message)
                    await cache.add_message_to_cache(token=token, source="human", message_data=msg.dict())
                    # कैश से चैट इतिहास प्राप्त करें
                    data = await cache.get_chat_history(token=token)
                    # मैसेज इनपुट को साफ करें और क्वेरी को भेजें
                    message_data = data['messages'][-4:]
                    input = ["" + i['msg'] for i in message_data]
                    input = " ".join(input)
                    res = GPT().query(input=input)
                    msg = Message(
                        msg=res
                    )
                    print(msg)
                    await cache.add_message_to_cache(token=token, source="bot", message_data=msg.dict())
                    # संदेश को प्रोसेस होने के बाद क्यू से डिलीट करें
                    await consumer.delete_message(stream_channel="message_channel", message_id=message_id)

if __name__ == "__main__":
    asyncio.run(main())

यह काफी अपडेट है, तो आइए इसे चरण दर चरण लें:

हम while True लूप का उपयोग करते हैं ताकि वर्कर क्यू से संदेश सुनने के लिए ऑनलाइन रह सके।

अगला, हम अपने consume_stream मेथड को कॉल करके message_channel से नए संदेशों की प्रतीक्षा करते हैं। यदि क्यू में कोई संदेश है, तो हम message_id, टोकन, और संदेश निकालते हैं। फिर हम Message क्लास का एक नया इंस्टेंस बनाते हैं, संदेश को कैश में जोड़ते हैं, और फिर अंतिम 4 संदेश प्राप्त करते हैं। हम इसे GPT मॉडल query मेथड के लिए इनपुट के रूप में सेट करते हैं।

एक बार जब हमें प्रतिक्रिया मिल जाती है, तो हम फिर add_message_to_cache मेथड का उपयोग करके प्रतिक्रिया को कैश में जोड़ते हैं, फिर संदेश को क्यू से डिलीट कर देते हैं।

AI प्रतिक्रिया के साथ चैट क्लाइंट को कैसे अपडेट करें

अब तक, हम क्लाइंट से मैसेज चैनल पर एक चैट संदेश भेज रहे हैं (जिसे वर्कर द्वारा प्राप्त किया जाता है जो AI मॉडल को क्वेरी करता है) एक प्रतिक्रिया प्राप्त करने के लिए।

अब, हमें इस प्रतिक्रिया को क्लाइंट को भेजने की आवश्यकता है। जब तक सॉकेट कनेक्शन अभी भी खुला है, क्लाइंट प्रतिक्रिया प्राप्त करने में सक्षम होना चाहिए।

यदि कनेक्शन बंद है, तो क्लाइंट हमेशा refresh_token एंडपॉइंट का उपयोग करके चैट इतिहास से प्रतिक्रिया प्राप्त कर सकता है।

worker.src.redis में producer.py नामक एक नई फाइल बनाएं, और एक Producer क्लास जोड़ें जो हमारे चैट वेब सर्वर पर थी:

class Producer:
    def __init__(self, redis_client):
        self.redis_client = redis_client

    async def add_to_stream(self, data: dict, stream_channel) -> bool:
        msg_id = await self.redis_client.xadd(name=stream_channel, id="*", fields=data)
        print(f"Message id {msg_id} added to {stream_channel} stream")
        return msg_id

अगला, main.py फाइल में, main फंक्शन को प्रोड्यूसर को इनिशियलाइज़ करने, एक स्ट्रीम डेटा बनाने, और add_to_stream मेथड का उपयोग करके प्रतिक्रिया को response_channel पर भेजने के लिए अपडेट करें:

from src.redis.config import Redis
import asyncio
from src.model.gptj import GPT
from src.redis.cache import Cache
from src.redis.config import Redis
from src.redis.stream import StreamConsumer
import os
from src.schema.chat import Message
from src.redis.producer import Producer

redis = Redis()

async def main():
    json_client = redis.create_rejson_connection()
    redis_client = await redis.create_connection()
    consumer = StreamConsumer(redis_client)
    cache = Cache(json_client)
    producer = Producer(redis_client)
    print("Stream consumer started")
    print("Stream waiting for new messages")
    while True:
        response = await consumer.consume_stream(stream_channel="message_channel", count=1, block=0)
        if response:
            for stream, messages in response:
                # स्ट्रीम से संदेश प्राप्त करें, और टोकन, मैसेज डेटा और मैसेज आईडी निकालें
                for message in messages:
                    message_id = message[0]
                    token = [k.decode('utf-8') for k, v in message[1].items()][0]
                    message = [v.decode('utf-8') for k, v in message[1].items()][0]
                    # एक नया मैसेज इंस्टेंस बनाएं और कैश में जोड़ें, स्रोत को मानव के रूप में निर्दिष्ट करें
                    msg = Message(msg=message)
                    await cache.add_message_to_cache(token=token, source="human", message_data=msg.dict())
                    # कैश से चैट इतिहास प्राप्त करें
                    data = await cache.get_chat_history(token=token)
                    # मैसेज इनपुट को साफ करें और क्वेरी को भेजें
                    message_data = data['messages'][-4:]
                    input = ["" + i['msg'] for i in message_data]
                    input = " ".join(input)
                    res = GPT().query(input=input)
                    msg = Message(
                        msg=res
                    )
                    stream_data = {}
                    stream_data[str(token)] = str(msg.dict())
                    await producer.add_to_stream(stream_data, "response_channel")
                    await cache.add_message_to_cache(token=token, source="bot", message_data=msg.dict())
                    # संदेश को प्रोसेस होने के बाद क्यू से डिलीट करें
                    await consumer.delete_message(stream_channel="message_channel", message_id=message_id)

if __name__ == "__main__":
    asyncio.run(main())

अगला, हमें क्लाइंट को यह बताने की आवश्यकता है कि हम /chat सॉकेट एंडपॉइंट में वर्कर से प्रतिक्रियाएं कब प्राप्त करते हैं। हम ऐसा रिस्पॉन्स स्ट्रीम को सुनकर करते हैं। हमें यहां व्हाइल लूप शामिल करने की आवश्यकता नहीं है क्योंकि सॉकेट तब तक सुनेगा जब तक कनेक्शन खुला है।

ध्यान दें कि हमें यह भी जांचना होगा कि प्रतिक्रिया किस क्लाइंट के लिए है, यह जांचकर कि कनेक्टेड टोकन रिस्पॉन्स में टोकन के बराबर है या नहीं। फिर एक बार पढ़े जाने पर रिस्पॉन्स क्यू में संदेश को डिलीट करें।

server.src.redis में stream.py नामक एक नई फाइल बनाएं और हमारे StreamConsumer क्लास को इस तरह जोड़ें:

from .config import Redis

class StreamConsumer:
    def __init__(self, redis_client):
        self.redis_client = redis_client

    async def consume_stream(self, count: int, block: int, stream_channel):
        response = await self.redis_client.xread(
            streams={stream_channel: '0-0'}, count=count, block=block)
        return response

    async def delete_message(self, stream_channel, message_id):
        await self.redis_client.xdel(stream_channel, message_id)

अगला, /chat सॉकेट एंडपॉइंट को इस तरह अपडेट करें:

from ..redis.stream import StreamConsumer

@chat.websocket("/chat")
async def websocket_endpoint(websocket: WebSocket, token: str = Depends(get_token)):
    await manager.connect(websocket)
    redis_client = await redis.create_connection()
    producer = Producer(redis_client)
    json_client = redis.create_rejson_connection()
    consumer = StreamConsumer(redis_client)
    try:
        while True:
            data = await websocket.receive_text()
            stream_data = {}
            stream_data[str(token)] = str(data)
            await producer.add_to_stream(stream_data, "message_channel")
            response = await consumer.consume_stream(stream_channel="response_channel", block=0)
            print(response)
            for stream, messages in response:
                for message in messages:
                    response_token = [k.decode('utf-8') for k, v in message[1].items()][0]
                    if token == response_token:
                        response_message = [v.decode('utf-8') for k, v in message[1].items()][0]
                        print(message[0].decode('utf-8'))
                        print(token)
                        print(response_token)
                        await manager.send_personal_message(response_message, websocket)
                        await consumer.delete_message(stream_channel="response_channel", message_id=message[0].decode('utf-8'))
    except WebSocketDisconnect:
        manager.disconnect(websocket)

रिफ्रेश टोकन

अंत में, हमें अपने Cache क्लास का उपयोग करके Redis डेटाबेस से चैट इतिहास प्राप्त करने के लिए /refresh_token एंडपॉइंट को अपडेट करने की आवश्यकता है।

server.src.redis में, cache.py फाइल जोड़ें और नीचे दिया गया कोड जोड़ें:

from rejson import Path

class Cache:
    def __init__(self, json_client):
        self.json_client = json_client

    async def get_chat_history(self, token: str):
        data = self.json_client.jsonget(
            str(token), Path.rootPath())
        return data

अगला, server.src.routes.chat.py में Cache क्लास इम्पोर्ट करें और /token एंडपॉइंट को नीचे दिए अनुसार अपडेट करें:

from ..redis.cache import Cache

@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 == None:
        raise HTTPException(
            status_code=400,
            detail="Session expired or does not exist"
        )
    else:
        return data

अब, जब हम किसी भी टोकन के साथ /refresh_token एंडपॉइंट पर GET रिक्वेस्ट भेजते हैं, तो एंडपॉइंट हमारे Cache क्लास का उपयोग करके Redis डेटाबेस से डेटा प्राप्त करेगा।

यदि टोकन टाइम आउट नहीं हुआ है, तो डेटा यूजर को भेजा जाएगा। या यदि टोकन नहीं मिलता है तो यह 400 प्रतिक्रिया भेजेगा।

Postman में मल्टीपल क्लाइंट्स के साथ चैट कैसे टेस्ट करें

अंत में, हम Postman में मल्टीपल चैट सेशन बनाकर, Postman में मल्टीपल क्लाइंट्स कनेक्ट करके, और क्लाइंट्स पर बॉट के साथ चैट करके चैट सिस्टम को टेस्ट करेंगे।

अंत में, हम क्लाइंट्स के लिए चैट इतिहास प्राप्त करने का प्रयास करेंगे और उम्मीद है कि उचित प्रतिक्रिया मिलेगी।

सारांश

आइए एक त्वरित सारांश लें कि हमने अपने चैट सिस्टम के साथ क्या हासिल किया है। चैट क्लाइंट प्रत्येक क्लाइंट के साथ चैट सेशन के लिए एक टोकन बनाता है। इस टोकन का उपयोग प्रत्येक क्लाइंट की पहचान करने के लिए किया जाता है, और हमारे वेब सर्वर से जुड़े क्लाइंट्स द्वारा भेजे गए प्रत्येक संदेश को एक Redis चैनल (message_channel) में कतारबद्ध किया जाता है, जिसे टोकन द्वारा पहचाना जाता है।

हमारा वर्कर एनवायरनमेंट इस चैनल से पढ़ता है। इसे इस बात का कोई सुराग नहीं है कि क्लाइंट कौन है (सिवाय इसके कि यह एक यूनिक टोकन है) और क्यू में संदेश का उपयोग Huggingface इनफेरेंस API पर रिक्वेस्ट भेजने के लिए करता है।

जब इसे प्रतिक्रिया मिलती है, तो प्रतिक्रिया को एक रिस्पॉन्स चैनल में जोड़ा जाता है और चैट इतिहास अपडेट किया जाता है। अपने टोकन के साथ प्रतिक्रिया प्राप्त करते ही response_channel को सुन रहा क्लाइंट तुरंत क्लाइंट को प्रतिक्रिया भेजता है।

यदि सॉकेट अभी भी खुला है, तो यह प्रतिक्रिया भेजी जाती है। यदि सॉकेट बंद है, तो हम निश्चित हैं कि प्रतिक्रिया संरक्षित है क्योंकि प्रतिक्रिया चैट इतिहास में जोड़ी गई है। क्लाइंट इतिहास प्राप्त कर सकता है, भले ही पेज रीफ्रेश हो या कनेक्शन खो जाए।

इतनी दूर आने के लिए बधाई! आप एक वर्किंग चैट सिस्टम बनाने में सक्षम हुए हैं।

फॉलो-अप लेखों में, मैं क्लाइंट के लिए चैट यूजर इंटरफेस बनाने, यूनिट और फंक्शनल टेस्ट बनाने, WebSockets और एसिंक्रोनस रिक्वेस्ट के साथ तेज प्रतिक्रिया समय के लिए हमारे वर्कर एनवायरनमेंट को फाइन-ट्यून करने, और अंततः चैट एप्लिकेशन को AWS पर डिप्लॉय करने पर ध्यान केंद्रित करूंगा।

यह लेख Python, React, Huggingface, Redis, और जैसे टूल्स के साथ फुल-स्टैक इंटेलिजेंट चैटबॉट बनाने पर एक श्रृंखला का हिस्सा है। आप मेरे ब्लॉग पर पूरी श्रृंखला का अनुसरण कर सकते हैं: blog.stephensanwo.dev - AI ChatBot Series**

आप मेरे GitHub रिपॉजिटरी पर पूरा रिपॉजिटरी डाउनलोड कर सकते हैं

मैंने यह ट्यूटोरियल Redis के सहयोग से लिखा है। Redis के साथ शुरुआत करने में सहायता चाहिए? निम्नलिखित संसाधनों को आज़माएं:

• Redis Cloud को मुफ़्त में आज़माएं
• इस वीडियो को देखें अन्य Redis प्रोवाइडर्स पर Redis Cloud के लाभों पर
• Redis डेवलपर हब - Redis के बारे में टूल्स, गाइड्स, और ट्यूटोरियल
• RedisInsight डेस्कटॉप GUI

मुफ़्त में कोड करना सीखें। freeCodeCamp का ओपन सोर्स पाठ्यक्रम 40,000 से अधिक लोगों को डेवलपर के रूप में नौकरी पाने में मदद कर चुका है। शुरू करें