मास्टर एआई चैटबॉट्स:रेडिस और पायथन के साथ एक शक्तिशाली जीपीटी-संचालित बॉट बनाएं - एक व्यापक गाइड

• पायथन, फास्टएपीआई और वेबसॉकेट के साथ एपीआई कैसे बनाएं
• रेडिस के साथ रीयल-टाइम सिस्टम कैसे बनाएं
• रिएक्ट के साथ चैट यूजर इंटरफेस कैसे बनाएं

सामग्री तालिका

धारा 1

• एप्लिकेशन आर्किटेक्चर
• विकास परिवेश कैसे स्थापित करें

  धारा 2

• पायथन, फास्टएपीआई और वेबसॉकेट के साथ चैट सर्वर कैसे बनाएं
  • पायथन पर्यावरण कैसे स्थापित करें
  • FastAPI सर्वर सेटअप
  • एपीआई में रूट कैसे जोड़ें
  • UUID के साथ चैट सेशन टोकन कैसे जेनरेट करें
  • पोस्टमैन के साथ एपीआई का परीक्षण कैसे करें
  • वेबसॉकेट और कनेक्शन प्रबंधक
  • FastAPI में निर्भरता इंजेक्शन

    धारा 3

• रेडिस के साथ रीयल-टाइम सिस्टम कैसे बनाएं
  • रेडिस और वितरित मैसेजिंग कतारें
  • रेडिस क्लाइंट के साथ पायथन में रेडिस क्लस्टर से कैसे जुड़ें
  • रेडिस स्ट्रीम्स के साथ कैसे काम करें
  • चैट डेटा को मॉडल कैसे करें
  • रेडिस JSON के साथ कैसे काम करें
  • टोकन निर्भरता को कैसे अद्यतन करें

    धारा 4

• एआई मॉडल के साथ चैटबॉट्स में इंटेलिजेंस कैसे जोड़ें
  • हगिंगफेस से कैसे शुरुआत करें
  • भाषा मॉडल के साथ कैसे इंटरैक्ट करें
  • एआई मॉडल के लिए अल्पकालिक मेमोरी का अनुकरण कैसे करें
  • संदेश कतार से उपभोक्ता और रीयल-टाइमडीडेटा स्ट्रीम स्ट्रीम करें
  • चैट क्लाइंट को AI रिस्पॉन्स के साथ कैसे अपडेट करें
  • टोकन ताज़ा करें
  • पोस्टमैन में एकाधिक ग्राहकों के साथ चैट का परीक्षण कैसे करें

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

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

GPT-J-6B और हगिंगफेस इन्फेरेंस एपीआई

रेडिस

वेब सॉकेट और चैट एपीआई

विकास परिवेश कैसे स्थापित करें

git init
touch .gitignore

पायथन, फास्टएपीआई और वेबसॉकेट के साथ चैट सर्वर कैसे बनाएं

पायथन पर्यावरण कैसे स्थापित करें

pip install fastapi uuid uvicorn gunicorn WebSockets python-dotenv aioredis

FastAPI सर्वर सेटअप

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

एपीआई में रूट कैसे जोड़ें

import os
from fastapi import APIRouter, FastAPI, WebSocket, Request
chat = APIRouter()
# @route POST /token
# @desc Route to generate chat token
# @access Public
@chat.post("/token")
async def token_generator(request: Request):
 return None
# @route POST /refresh_token
# @desc Route to refresh token
# @access Public
@chat.post("/refresh_token")
async def refresh_token(request: Request):
 return None
# @route Websocket /chat
# @desc Socket for chatbot
# @access Public
@chat.websocket("/chat")
async def websocket_endpoint(websocket: WebSocket = WebSocket):
 return None

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

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 के साथ चैट सेशन टोकन कैसे जेनरेट करें

from fastapi import APIRouter, FastAPI, WebSocket, Request, BackgroundTasks, HTTPException
import uuid
# @route POST /token
# @desc Route generating chat token
# @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

पोस्टमैन के साथ एपीआई का परीक्षण कैसे करें

वेबसॉकेट और कनेक्शन प्रबंधक

from fastapi import WebSocket
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)

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)

FastAPI में निर्भरता इंजेक्शन

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

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)

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 Route to generate chat token
# @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 Route to refresh token
# @access Public
@chat.post("/refresh_token")
async def refresh_token(request: Request):
 return None
# @route Websocket /chat
# @desc Socket for chatbot
# @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 JSON के साथ डेटा कैसे संग्रहीत और पुनर्प्राप्त करें
• कैसे सेट अप करें रेडिस स्ट्रीम एक वेब सर्वर और कार्यकर्ता वातावरण के बीच संदेश कतार के रूप में

रेडिस और वितरित मैसेजिंग कतारें

pip install aiohttp aioredis python-dotenv

रेडिस क्लाइंट के साथ पायथन में रेडिस क्लस्टर से कैसे जुड़ें

export REDIS_URL=<REDIS URL PROVIDED IN REDIS CLOUD>
export REDIS_USER=<REDIS USER IN REDIS CLOUD>
export REDIS_PASSWORD=<DATABASE PASSWORD IN REDIS CLOUD>
export REDIS_HOST=<REDIS HOST IN REDIS CLOUD>
export REDIS_PORT=<REDIS PORT IN REDIS CLOUD>

import os
from dotenv import load_dotenv
import aioredis
load_dotenv()
class Redis():
 def __init__(self):
 """initialize connection """
 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

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())

रेडिस स्ट्रीम के साथ कैसे काम करें

import os
from dotenv import load_dotenv
import aioredis
load_dotenv()
class Redis():
 def __init__(self):
 """initialize connection """
 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

export REDIS_URL=<REDIS URL PROVIDED IN REDIS CLOUD>
export REDIS_USER=<REDIS USER IN REDIS CLOUD>
export REDIS_PASSWORD=<DATABASE PASSWORD IN REDIS CLOUD>
export REDIS_HOST=<REDIS HOST IN REDIS CLOUD>
export REDIS_PORT=<REDIS PORT IN REDIS CLOUD>

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

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)

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

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())

रेडिस JSON के साथ कैसे काम करें

import os
from dotenv import load_dotenv
import aioredis
from rejson import Client
load_dotenv()
class Redis():
 def __init__(self):
 """initialize connection """
 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

@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"})
 # Create new chat session
 json_client = redis.create_rejson_connection()
 chat_session = Chat(
 token=token,
 messages=[],
 name=name
 )
 # Store chat session in redis JSON with the token as key
 json_client.jsonset(str(token), Path.rootPath(), chat_session.dict())
 # Set a timeout for redis data
 redis_client = await redis.create_connection()
 await redis_client.expire(str(token), 3600)
 return chat_session.dict()

टोकन निर्भरता को कैसे अपडेट करें

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

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 Route to generate chat token
# @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"})
 # Create nee chat session
 json_client = redis.create_rejson_connection()
 chat_session = Chat(
 token=token,
 messages=[],
 name=name
 )
 print(chat_session.dict())
 # Store chat session in redis JSON with the token as key
 json_client.jsonset(str(token), Path.rootPath(), chat_session.dict())
 # Set a timeout for redis data
 redis_client = await redis.create_connection()
 await redis_client.expire(str(token), 3600)
 return chat_session.dict()
# @route POST /refresh_token
# @desc Route to refresh token
# @access Public
@chat.post("/refresh_token")
async def refresh_token(request: Request):
 return None
# @route Websocket /chat
# @desc Socket for chat bot
# @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)

एआई मॉडल के साथ चैटबॉट्स में इंटेलिजेंस कैसे जोड़ें

हगिंगफेस के साथ शुरुआत कैसे करें

भाषा मॉडल के साथ कैसे इंटरैक्ट करें

export HUGGINFACE_INFERENCE_TOKEN=<HUGGINGFACE ACCESS TOKEN>
export MODEL_URL=https://api-inference.huggingface.co/models/EleutherAI/gpt-j-6B

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?")

[{'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'}]

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?")

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

एआई मॉडल के लिए अल्पकालिक मेमोरी का अनुकरण कैसे करें

import os
from dotenv import load_dotenv
import aioredis
from rejson import Client
load_dotenv()
class Redis():
 def __init__(self):
 """initialize connection """
 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

export REDIS_URL=<REDIS URL PROVIDED IN REDIS CLOUD>
export REDIS_USER=<REDIS USER IN REDIS CLOUD>
export REDIS_PASSWORD=<DATABASE PASSWORD IN REDIS CLOUD>
export REDIS_HOST=<REDIS HOST IN REDIS CLOUD>
export REDIS_PORT=<REDIS PORT IN REDIS CLOUD>
export HUGGINFACE_INFERENCE_TOKEN=<HUGGINGFACE ACCESS TOKEN>
export MODEL_URL=https://api-inference.huggingface.co/models/EleutherAI/gpt-j-6B

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

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())

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

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

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)

{'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'}

 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)

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)

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())

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())

Stream Consumer and Real-time Data Pull from the Message Queue

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)

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:
 # Get message from stream, and extract token, message data and message id
 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)
 # Create a new message instance and add to cache, specifying the source as human
 msg = Message(msg=message)
 await cache.add_message_to_cache(token=token, source="human", message_data=msg.dict())
 # Get chat history from cache
 data = await cache.get_chat_history(token=token)
 # Clean message input and send to query
 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())
 # Delete messaage from queue after it has been processed
 await consumer.delete_message(stream_channel="message_channel", message_id=message_id)
if __name__ == "__main__":
 asyncio.run(main())

How to Update the Chat Client with the AI Response

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

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:
 # Get message from stream, and extract token, message data and message id
 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]
 # Create a new message instance and add to cache, specifying the source as human
 msg = Message(msg=message)
 await cache.add_message_to_cache(token=token, source="human", message_data=msg.dict())
 # Get chat history from cache
 data = await cache.get_chat_history(token=token)
 # Clean message input and send to query
 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())
 # Delete messaage from queue after it has been processed
 await consumer.delete_message(stream_channel="message_channel", message_id=message_id)
if __name__ == "__main__":
 asyncio.run(main())

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)

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)

Refresh Token

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

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

How to Test the Chat with multiple Clients in Postman

Recap

• Try Redis Cloud free of charge
• Watch this video on the benefits of Redis Cloud over other Redis providers
• Redis Developer Hub - tools, guides, and tutorials about Redis
• RedisInsight Desktop GUI