AI চ্যাটবট আয়ত্ত করুন: Redis এবং Python দিয়ে একটি শক্তিশালী GPT-চালিত বট বানান – একটি সার্বিক গাইড

লেখক: Stephen Sanwo

একটি কাজ করা ফুল-স্ট্যাক অ্যাপ্লিকেশন বানাতে, অনেক মুভিং পার্টস সম্পর্কে চিন্তা করতে হয়। এবং আপনাকে অনেক ডিসিশন নিতে হবে যা আপনার অ্যাপের সাফল্যের জন্য Криটিক্যাল।

উদাহরণস্বরূপ, আপনি কোন ল্যাঙ্গুয়েজ ব্যবহার করবেন এবং কোন প্ল্যাটফর্মে ডিপ্লয় করবেন? আপনি কি একটি কন্টেইনারাইজড সফটওয়্যার সার্ভারে ডিপ্লয় করবেন, নাকি সার্ভারলেস ফাংশন ব্যবহার করবেন ব্যাকএন্ড হ্যান্ডল করতে? আপনি কি থার্ড-পার্টি API ব্যবহার করবেন আপনার অ্যাপ্লিকেশনের জটিল অংশগুলো হ্যান্ডল করতে, যেন অথেনটিকেশন বা পেমেন্ট? আপনি ডেটা কোথায় সংরক্ষণ করবেন?

এই সব ছাড়াও, আপনাকে ইউজার ইন্টারফেস, ডিজাইন এবং আপনার অ্যাপ্লিকেশনের ইউজেবিলিটি সম্পর্কেও চিন্তা করতে হবে, এবং আরও অনেক কিছু।

এই কারণেই জটিল বড় অ্যাপ্লিকেশনগুলোর জন্য একটি মাল্টি-ফাংশনাল ডেভেলপমেন্ট টিমের সহযোগিতা প্রয়োজন অ্যাপ বানাতে।

ফুল-স্ট্যাক অ্যাপ্লিকেশন ডেভেলপমেন্ট শিখার সেরা উপায়গুলোর একটি হলো এইচ-টু-ইন্ড ডেভেলপমেন্ট প্রসেস কভার করে প্রজেক্ট বানানো। আপনি আর্কিটেকচার ডিজাইন, API সার্ভিস ডেভেলপমেন্ট, ইউজার ইন্টারফেস ডেভেলপমেন্ট, এবং অবশেষে আপনার অ্যাপ্লিকেশন ডিপ্লয় করার প্রক্রিয়া থেকে যাবেন।

এই টিউটোরিয়াল আপনাকে একটি AI চ্যাটবট বানানোর প্রক্রিয়ায় নিয়ে যাবে যাতে আপনি এই কনসেপ্টগুলো গভীরে শিখতে পারেন।

আমরা যে কিছু টপিক কভার করব:

• Python, FastAPI, এবং WebSockets দিয়ে API কীভাবে বানাবেন
• Redis দিয়ে রিয়েল-টাইম সিস্টেম কীভাবে বানাবেন
• React দিয়ে চ্যাট ইউজার ইন্টারফেস কীভাবে বানাবেন

গুরুত্বপূর্ণ নোট: এটি একটি ইন্টারমিডিয়েট ফুল-স্ট্যাক সফটওয়্যার ডেভেলপমেন্ট প্রজেক্ট যেটি কিছু বেসিক Python এবং JavaScript নলেজ চায়।

আমি প্রজেক্টটি সেকশনগুলোর মধ্যে সাবধানে ভাগ করেছি যাতে আপনি সহজে সেই ফেজটি বেছে নিতে পারেন যা আপনার জন্য গুরুত্বপূর্ণ, যদি আপনি পুরো অ্যাপ্লিকেশন কোড করতে না চান।

পূর্ণ রিপোজিটরি ডাউনলোড করুন আমার GitHub-এ থেকে এখানে।

সารানী

সেকশন ১

• অ্যাপ্লিকেশন আর্কিটেকচার
• ডেভেলপমেন্ট এনভায়রনমেন্ট সেট আপ কীভাবে করবেন

সেকশন ২

• Python, FastAPI, এবং WebSockets দিয়ে একটি চ্যাট সার্ভার কীভাবে বানাবেন
  • Python এনভায়রনমেন্ট সেট আপ কীভাবে করবেন
  • FastAPI সার্ভার সেটআপ
  • API-তে রুট যোগ কীভাবে করবেন
  • UUID দিয়ে চ্যাট সেশন টোকেন জেনারেট কীভাবে করবেন
  • Postman দিয়ে API টেস্ট কীভাবে করবেন
  • WebSockets এবং কানেকশন ম্যানেজার
  • FastAPI-এ ডিপেন্ডেন্সি ইনজেকশন

সেকশন ৩

• Redis দিয়ে রিয়েল-টাইম সিস্টেম কীভাবে বানাবেন
  • Redis এবং ডিস্ট্রিবিউটেড মেসেজিং কিউ
  • Python-এ Redis ক্লায়েন্ট দিয়ে Redis ক্লাস্টারের সাথে কানেক্ট কীভাবে করবেন
  • Redis Streams দিয়ে কীভাবে কাজ করবেন
  • চ্যাট ডেটা মডেল কীভাবে করবেন
  • Redis JSON দিয়ে কীভাবে কাজ করবেন
  • টোকেন ডিপেন্ডেন্সি আপডেট কীভাবে করবেন

সেকশন ৪

• AI মডেল দিয়ে চ্যাটবটে ইন্টেলিজেন্স যোগ কীভাবে করবেন
  • Huggingface দিয়ে শুরু কীভাবে করবেন
  • ল্যাঙ্গুয়েজ মডেলের সাথে ইন্টারঅ্যাক্ট কীভাবে করবেন
  • AI মডেলের জন্য শর্ট-টার্ম মেমরি সিমুলেট কীভাবে করবেন
  • স্ট্রিম কনসিউমার এবং মেসেজ কিউ থেকে রিয়েল-টাইম ডেটা পুল
  • AI রেসপন্স দিয়ে চ্যাট ক্লায়েন্ট আপডেট কীভাবে করবেন
  • রিফ্রেশ টোকেন
  • Postman-এ মাল্টিপল ক্লায়েন্ট দিয়ে চ্যাট টেস্ট কীভাবে করবেন

অ্যাপ্লিকেশন আর্কিটেকচার

একটি সমাধান আর্কিটেকচার স্কেচ করলে আপনার অ্যাপ্লিকেশন, আপনি যে টুলগুলো ব্যবহার করতে চান, এবং কম্পোনেন্টগুলো একে অপরের সাথে কীভাবে যোগাযোগ করবে সেটির একটি হাই-লেভেল ওভারভিউ দেয়।

আমি নিচে একটি সিম্পল আর্কিটেকচার আঁকেছি draw.io ব্যবহার করে:

ফুলস্ট্যাক চ্যাটবট আর্কিটেকচার

আর্কিটেকচারের বিভিন্ন অংশগুলো আরও বিস্তারিত জানি:

ক্লায়েন্ট/ইউজার ইন্টারফেস

আমরা React ভার্সন ১৮ ব্যবহার করব ইউজার ইন্টারফেস বানাতে। চ্যাট UI ব্যাকএন্ডের সাথে WebSockets এর মাধ্যমে যোগাযোগ করবে।

GPT-J-6B এবং Huggingface Inference API

GPT-J-6B একটি জেনারেটিভ ল্যাঙ্গুয়েজ মডেল যেটি ৬ বিলিয়ন প্যারামিটার দিয়ে ট্রেন করা হয়েছে এবং কিছু টাস্কে 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 কানেকশন পর্যাপ্ত নয় রিয়েল-টাইম bi-directional যোগাযোগ নিশ্চিত করতে ক্লায়েন্ট এবং সার্ভারের মধ্যে।

আমরা চ্যাট সার্ভারের জন্য FastAPI ব্যবহার করব, কারণ এটি আমাদের ইউজ কেসের জন্য একটি ফাস্ট এবং মডার্ন Python সার্ভার প্রদান করে। FastAPI ডকুমেন্টেশন চেক আউট করুন WebSockets সম্পর্কে আরও জানতে।

ডেভেলপমেন্ট এনভায়রনমেন্ট সেট আপ কীভাবে করবেন

আপনি আপনার পছন্দের OS ব্যবহার করতে পারেন এই অ্যাপ বানাতে – আমি বর্তমানে MacOS এবং Visual Studio Code ব্যবহার করছি। শুধু নিশ্চিত করুন আপনার পাস Python এবং NodeJs ইনস্টলড আছে।

প্রজেক্ট স্ট্রাকচার সেট আপ করতে, fullstack-ai-chatbot নামে একটি ফোল্ডার তৈরি করুন। তারপর প্রজেক্টের মধ্যে দুটি ফোল্ডার তৈরি করুন client এবং server নামে। সার্ভারটি ব্যাকএন্ডের কোড রাখবে, আর ক্লায়েন্টটি ফ্রন্টএন্ডের কোড রাখবে।

পরবর্তী প্রজেক্ট ডিরেক্টরির মধ্যে, প্রজেক্ট ফোল্ডারের রুটে একটি Git রিপোজিটরি ইনিশিয়ালাইজ করুন "git init" কমান্ড ব্যবহার করে। তারপর একটি .gitignore ফাইল তৈরি করুন "touch .gitignore" ব্যবহার করে:

git init
touch .gitignore

পরবর্তী সেকশনে, আমরা FastAPI এবং Python ব্যবহার করে আমাদের চ্যাট ওয়েব সার্ভার বানাবো।

Python, FastAPI এবং WebSockets দিয়ে একটি চ্যাট সার্ভার কীভাবে বানাবেন

এই সেকশনে, আমরা FastAPI ব্যবহার করে চ্যাট সার্ভার বানাবো যাতে ইউজারের সাথে যোগাযোগ করতে পারে। আমরা WebSockets ব্যবহার করব bi-directional যোগাযোগ নিশ্চিত করতে ক্লায়েন্ট এবং সার্ভারের মধ্যে যাতে আমরা রিয়েল-টাইমে ইউজারকে রেসপন্স পাঠাতে পারি।

Python এনভায়রনমেন্ট সেট আপ কীভাবে করবেন

আমাদের সার্ভার শুরু করতে, আমাদের Python এনভায়রনমেন্ট সেট আপ করতে হবে। VS Code-এ প্রজেক্ট ফোল্ডার খুলুন, এবং টার্মিনাল খুলুন।

প্রজেক্ট রুট থেকে, server ডিরেক্টরিতে 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 হিসেবে ইনিশিয়ালাইজ করি। তারপর load_dotenv ইম্পোর্ট করি python-dotenv লাইব্রেরি থেকে, এবং এটি ইনিশিয়ালাইজ করি .env ফাইল থেকে ভেরিয়েবলগুলো লোড করতে।

তারপর আমরা একটি সিম্পল টেস্ট রুট তৈরি করি API টেস্ট করতে। টেস্ট রুট একটি সিম্পল JSON রেসপন্স রিটার্ন করবে যা আমাদের বলে API অনলাইন।

শেষে, আমরা ডেভেলপমেন্ট সার্ভার সেট আপ করি uvicorn.run ব্যবহার করে এবং প্রয়োজনীয় আর্গুমেন্টগুলো দিয়ে। API পোর্ট 3500-এ চলবে।

শেষে, সার্ভার টার্মিনালে python main.py দিয়ে চালান। একবার আপনি Application startup complete দেখেন টার্মিনালে, ব্রাউজারে 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-র সাথে কানেক্ট করুন। প্রথমে আমাদের chat from src.chat ইম্পোর্ট করতে হবে আমাদের main.py ফাইলে। তারপর আমরা রাউটার ইনক্লুড করব লITERALLY 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 ব্যবহার করব আমাদের চ্যাট এন্ডপয়েন্টের জন্য ডাইনামিক রুট তৈরি করতে। যেহেতু এটি একটি পাবলিকলি অবেলেবল এন্ডপয়েন্ট, আমাদের JWT এবং অথেনটিকেশনের বিস্তারিতে যেতে হবে না।

আপনি যদি শুরুতে 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-র ডিফল্ট swagger ডক্স WebSockets সাপোর্ট করে না)।

Postman-এ, আপনার ডেভেলপমেন্ট এনভায়রনমেন্টের জন্য একটি কালেকশন তৈরি করুন এবং একটি POST রিকোয়েস্ট পাঠান localhost:3500/token এ name কোয়েরি প্যারামিটার হিসেবে স্পেসিফাই করে এবং এটিকে একটি ভ্যালু দিয়ে। আপনাকে নিচের রেসপন্স পাওয়া উচিত:

টোকেন জেনারেটর 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 নেয়, আমরা নতুন websocket কানেকশন ম্যানেজারে যোগ করি এবং একটি while True লুপ চালাই, যাতে সকেট খোলা থাকে। ছাড়া যখন সকেট ডিসকানেক্ট হয়।

কানেকশন খোলা থাকার সময়, আমরা ক্লায়েন্ট দ্বারা পাঠানো যেকোনো মেসেজ রিসিভ করি websocket.receive_text() দিয়ে এবং টার্মিনালে প্রিন্ট করি এখনও।

তারপর আমরা একটি হার্ড-কোডেড রেসপন্স পাঠাই ক্লায়েন্টকে এখনও। চূড়ান্তভাবে ক্লায়েন্ট থেকে পাওয়া মেসেজ AI মডেলে পাঠানো হবে, এবং ক্লায়েন্টকে পাঠানো রেসপন্স হবে AI মডেল থেকে পাওয়া রেসপন্স।

Postman-এ, আমরা এই এন্ডপয়েন্ট টেস্ট করতে পারি একটি নতুন WebSocket রিকোয়েস্ট তৈরি করে, এবং WebSocket এন্ডপয়েন্ট localhost:3500/chat-এ কানেক্ট করে।

যখন আপনি কানেক্ট ক্লিক করবেন, Messages pane দেখাবে যে API ক্লায়েন্ট URL-এর সাথে কানেক্টেড, এবং একটি সকেট খোলা।

এটি টেস্ট করতে, "Hello Bot" মেসেজ পাঠান চ্যাট সার্ভারে, এবং আপনাকে একটি তাত্ক্ষণিক টেস্ট রেসপন্স "Response: Simulating response from the GPT service" পাওয়া উচিত নিচের মতঃ

Postman চ্যাট টেস্ট

FastAPI-এ ডিপেন্ডেন্সি ইনজেকশন

দুই ভিন্ন ক্লায়েন্ট সেশন পার্থক্য করতে এবং চ্যাট সেশন লিমিট করতে, আমরা একটি টাইমড টোকেন ব্যবহার করব, যেটি কোয়েরি প্যারামিটার হিসেবে পাস করা হবে WebSocket কানেকশনে।

socket ফোল্ডারে, utils.py নামে একটি ফাইল তৈরি করুন তারপর নিচের কোড যোগ করুন:

from fastapi import WebSocket, status, Query
from typing import Optional

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

get_token ফাংশন একটি WebSocket এবং টোকেন নেয়, তারপর চেক করে টোকেন None বা null কিনা।

যদি এই কেস হয়, ফাংশন একটি পলিসি ভায়োলেশন স্ট্যাটাস রিটার্ন করে এবং যদি উপলব্ধ থাকে, ফাংশন শুধু টোকেন রিটার্ন করে। আমরা চূড়ান্তভাবে এই ফাংশনকে আরও টোকেন ভ্যালিডেশন দিয়ে এক্সটেন্ড করব।

এই ফাংশন কনসাম করতে, আমরা এটি ইনজেক্ট করব /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)

এখন আপনি /chat এন্ডপয়েন্টে কানেক্ট করার চেষ্টা করলে Postman-এ, আপনাকে একটি 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 দিয়ে রিয়েল-টাইম সিস্টেম কীভাবে বানাবেন

আমাদের অ্যাপ্লিকেশন বর্তমানে কোনো স্টেট সংরক্ষণ করে না, এবং ইউজারদের আইডেন্টিফাই করার বা চ্যাট ডেটা সংরক্ষণ ও রিট্রিভ করার কোনো উপায় নেই। আমরা চ্যাট সেশনের সময় ক্লায়েন্টকে একটি হার্ড-কোডেড রেসপন্স রিটার্ন করছি।

টিউটোরিয়ালের এই অংশে, আমরা নিচের বিষয়গুলো কভার করব:

• Redis ক্লাস্টার-এ Python-এ কীভাবে কানেক্ট করবেন এবং Redis ক্লায়েন্ট সেট আপ করবেন
• Redis JSON দিয়ে ডেটা কীভাবে সংরক্ষণ ও রিট্রিভ করবেন
• Redis Streams কীভাবে সেট আপ করবেন মেসেজ কিউ হিসেবে একটি ওয়েব সার্ভার ও ওয়ার্কার এনভায়রনমেন্টের মধ্যে

Redis এবং ডিস্ট্রিবিউটেড মেসেজিং কিউ

Redis একটি ওপেন সোর্স ইন-মেমরি ডেটা স্টোর যা আপনি ডাটাবেস, ক্যাশ, মেসেজ ব্রোকার, এবং স্ট্রিমিং ইঞ্জিন হিসেবে ব্যবহার করতে পারেন। এটি অনেক ডেটা স্ট্রাকচার সাপোর্ট করে এবং রিয়েল-টাইম ক্যাপাবিলিটি সহ ডিস্ট্রিবিউটেড অ্যাপ্লিকেশনের জন্য একটি পরফেক্ট সমাধান।

Redis Enterprise Cloud Redis দ্বারা প্রদত্ত একটি ফুলি ম্যানেজড ক্লাউড সার্ভিস যা আমাদের Redis ক্লাস্টার ডিপ্লয় করতে সাহায্য করে অসীম স্কেলে ইনফ্রাস্ট্রাকচারের চিন্তা ছাড়া।

এই টিউটোরিয়ালের জন্য আমরা একটি ফ্রি Redis Enterprise Cloud ইন্সট্যান্স ব্যবহার করব। আপনি Redis Cloud-এ ফ্রি শুরু করতে পারেন এখানে এবং এই টিউটোরিয়াল ফলো করুন একটি Redis ডাটাবেস এবং Redis Insight সেট আপ করতে, Redis-র সাথে ইন্টারঅ্যাক্ট করার জন্য একটি GUI।

একবার আপনার Redis ডাটাবেস সেট আপ হয়ে গেলে, প্রজেক্ট রুটে (server ফোল্ডারের বাইরে) worker নামে একটি নতুন ফোল্ডার তৈরি করুন।

আমরা আমাদের ওয়ার্কার এনভায়রনমেন্টকে ওয়েব সার্ভার থেকে আলাদা করব যাতে যখন ক্লায়েন্ট আমাদের WebSocket-এ একটি মেসেজ পাঠায়, ওয়েব সার্ভারকে থার্ড-পার্টি সার্ভিসের রিকোয়েস্ট হ্যান্ডল করতে না হয়। এছাড়া, রিসোর্সগুলো ফ্রি হয়ে যাবে অন্যান্য ইউজারদের জন্য।

Inference API-র সাথে ব্যাকগ্রাউন্ড যোগাযোগ এই ওয়ার্কার সার্ভিস দ্বারা হ্যান্ডল করা হয়, Redis-এর মাধ্যমে।

সব কানেক্টেড ক্লায়েন্ট থেকে রিকোয়েস্টগুলো মেসেজ কিউতে অ্যাপেন্ড হয় (প্রডিউসার), যখন ওয়ার্কার মেসেজগুলো কনসাম করে, ইনফরেন্স API-তে রিকোয়েস্ট পাঠায়, এবং রেসপন্স একটি রেসপন্স কিউতে অ্যাপেন্ড করে।

একবার API রেসপন্স পেলে, সেটি ক্লায়েন্টকে terug পাঠায়।

প্রডিউসার এবং কনসিউমারের মধ্যে যাত্রার সময়, ক্লায়েন্ট একাধিক মেসেজ পাঠাতে পারে, এবং এই মেসেজগুলো কিউয়াপ হবে এবং ক্রমানুসারে রেসপন্স করা হবে।

আদর্শভাবে, আমরা এই ওয়ার্কারকে একটি সম্পূর্ণ আলাদা সার্ভারে চড়িয়ে রাখতে পারতাম, তার নিজস্ব এনভায়রনমেন্টে, কিন্তু এখনো, আমরা আমাদের লোকাল মেশিনে এর নিজস্ব Python এনভায়রনমেন্ট তৈরি করব।

আপনি কীভাবে চিন্তা করছেন – আমাদের কেন একটি ওয়ার্কার দরকার? একটি সিনেরিও কল্পনা করুন যেখানে ওয়েব সার্ভারও থার্ড-পার্টি সার্ভিসের রিকোয়েস্ট তৈরি করে। এর মানে সকেট কানেকশনের সময় থার্ড পার্টি সার্ভিস থেকে রেসপন্স এর জন্য অপেক্ষা করার সময়, সার্ভার ব্লকড থাকে এবং রিসোর্সগুলো জড়িত থাকে যতক্ষণ না API থেকে রেসপন্স পাওয়া যায়।

আপনি এটি ট্রাই করতে পারেন একটি র‍্যান্ডম স্লিপ time.sleep(10) তৈরি করে হার্ড-কোডেড রেসপন্স পাঠানোর আগে, এবং একটি নতুন মেসেজ পাঠিয়ে। তারপর একটি ভিন্ন টোকেন দিয়ে একটি নতুন Postman সেশনে কানেক্ট করার চেষ্টা করুন।

আপনি লক্ষ্য করবেন যে চ্যাট সেশন কানেক্ট হবে না যতক্ষণ না র‍্যান্ডম স্লিপ টাইম আউট হয়।

আমরা অ্যাসিঙ্ক্রোনাস টেকনিক এবং ওয়ার্কার পুল ব্যবহার করতে পারি একটি আরও প্রোডাকশন-ফোকাসড সার্ভার সেট-আপে, কিন্তু সেটাও পর্যাপ্ত নয় যেহেতু সিমাল্টানিয়াস ইউজারদের সংখ্যা বড়ে।

চূড়ান্তভাবে, আমরা আমাদের ওয়েব সার্ভার রিসোর্স জড়িয়ে রাখতে চাই না, Redis ব্যবহার করে আমাদের চ্যাট API এবং থার্ড-পার্টি API-এর যোগাযোগ ব্রোকার করতে।

পরবর্তী একটি নতুন টার্মিনাল খুলুন, worker ফোল্ডারে cd করুন, এবং একটি নতুন Python ভার্চুয়াল এনভায়রনমেন্ট তৈরি ও এক্টিভেট করুন পার্ট ১-এ আমরা যা করেছিলাম সেমili।

পরবর্তী, নিচের ডিপেন্ডেন্সিগুলো ইনস্টল করুন:

pip install aiohttp aioredis python-dotenv

Python-এ Redis ক্লায়েন্ট দিয়ে Redis ক্লাস্টারের সাথে কানেক্ট কীভাবে করবেন

আমরা aioredis ক্লায়েন্ট ব্যবহার করব Redis ডাটাবেসের সাথে কানেক্ট করার জন্য। আমরা requests লাইব্রেরিও ব্যবহার করব Huggingface inference API-তে রিকোয়েস্ট পাঠাতে।

.env, এবং main.py নামে দুটি ফাইল তৈরি করুন। তারপর src নামে একটি ফোল্ডার তৈরি করুন। আরো, redis নামে একটি ফোল্ডার তৈরি করুন এবং config.py নামে একটি নতুন ফাইল যোগ করুন।

.env ফাইলে, নিচের কোড যোগ করুন – এবং নিশ্চিত করুন আপনি ফিল্ডগুলো আপডেট করেছেন আপনার Redis ক্লাস্টারে প্রদত্ত ক্রেডেনশিয়ালস দিয়ে।

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>

config.py-এ নিচের Redis ক্লাস যোগ করুন:

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

আমরা একটি Redis অবজেক্ট তৈরি করি এবং এনভায়রনমেন্ট ভেরিয়েবল থেকে প্রয়োজনীয় প্যারামিটার ইনিশিয়ালাইজ করি। তারপর একটি অ্যাসিঙ্ক্রোনাস মেথড create_connection তৈরি করি Redis কানেকশন তৈরি করতে এবং aioredis মেথড from_url থেকে পাওয়া কানেকশন পুল রিটার্ন করতে।

পরবর্তী, 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):
        """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

.env ফাইলে, Redis ক্রেডেনশিয়ালসও যোগ করুন:

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>

শেষে, 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-ওয়ার টেক্সট ডেটা পাঠাচ্ছি, কিন্তু আমাদের চ্যাট ডেটাকে শুধু টেক্সট চেয়ে বেশি ইনফো হোল্ড করতে হবে। আমাদের চ্যাট কখন পাঠানো হয়েছে সেটি টাইমস্ট্যাম্প করতে হবে, প্রতিটি মেসেজের জন্য একটি আইডি তৈরি করতে হবে, এবং চ্যাট সেশন সম্পর্কে ডেটা কালেক্ট করতে হবে, তারপর এই ডেটা 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 ব্যবহার করে একটি চ্যাট আইডি অন-দ-ফ্লাই তৈরি করে। এই Message ক্লাস ইনিশিয়ালাইজ করার সময় আমাদের শুধু মেসেজ টেক্সট দিতে হবে।

Redis JSON দিয়ে কীভাবে কাজ করবেন

আমাদের চ্যাট হিস্টরি সংরক্ষণ করতে Redis JSON-এর ক্যাপাবিলিটি ব্যবহার করতে, আমাদের Redis Labs দ্বারা প্রদত্ত rejson ইনস্টল করতে হবে।

টার্মিনালে, server-এ cd করুন এবং rejson ইনস্টল করুন pip install rejson দিয়ে। তারপর আপনার Redis ক্লাসকে server.src.redis.config.py-এ আপডেট করুন create_rejson_connection মেথড ইনক্লুড করে:

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

আমরা create_rejson_connection মেথড যোগ করছি rejson Client দিয়ে Redis-র সাথে কানেক্ট করতে। এটি আমাদের JSON ডেটা Redis-এ তৈরি ও ম্যানিপুলেট করার মেথড দেয়, যেগুলো 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 ক্লায়েন্ট ব্যবহার করে (rejson টাইমআউট ইমপ্লিমেন্ট করে না)। এর মানে ৬০ মিনিট পর চ্যাট সেশন ডেটা হারিয়ে যাবে।

এটি প্রয়োজন কারণ আমরা ইউজারদের অথেনটিকেট করছি না, এবং আমরা একটি ডিফাইনড পিরিয়ডের পর চ্যাট ডেটা ডাম্প করতে চাই। এই স্টেপ অপশনাল, এবং আপনাকে এটি ইনক্লুড করতে হবেনা।

পরবর্তী, Postman-এ, যখন আপনি একটি নতুন টোকেন তৈরি করতে POST রিকোয়েস্ট পাঠাবেন, আপনাকে নিচের মত একটি স্ট্রাকচার্ড রেসপন্স मिलবে। আপনি Redis Insight-এও আপনার চ্যাট ডেটা দেখতে পারেন টোকেন হিসেবে JSON কি এবং ডেটা হিসেবে ভ্যালু।

আপডেটেড টোকেন জেনারেটর

টোকেন ডিপেন্ডেন্সি আপডেট কীভাবে করবেন

এখন যেহেতু আমাদের একটি টোকেন জেনারেট ও সংরক্ষিত হচ্ছে, এটি get_token ডিপেন্ডেন্সি আপডেট করার একটি ভালো সময় আমাদের /chat WebSocket-এ। আমরা এটি করব একটি ভ্যালিড টোকেন চেক করার জন্য চ্যাট সেশন শুরু করার আগে।

server.src.socket.utils.py-এ get_token ফাংশন আপডেট করুন যাতে চেক করে টোকেন Redis ইন্সট্যান্সে আছে কিনা। যদি থাকে তবে আমরা টোকেন রিটার্ন করি, যার মানে সকেট কানেকশন ভ্যালিড। যদি না থাকে, তবে কানেকশন বন্ধ করি।

/token-এর 의해 তৈরি টোকেন ৬০ মিনিট পর আর অস্তিত্বে থাকবে না। তাই আমরা ফ্রন্টএন্ডে কিছু সিম্পল লজিক রাখতে পারি ইউজারকে নতুন টোকেন জেনারেট করতে রিডাইরেক্ট করতে যদি চ্যাট শুরু করার চেষ্টায় এরর রেসপন্স জেনারেট হয়।

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 এন্ডপয়েন্টে POST রিকোয়েস্ট পাঠালে জেনারেট হওয়া টোকেনটি কপি করুন (বা একটি নতুন রিকোয়েস্ট তৈরি করুন) এবং এটিকে /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 মডেল। এটি একটি জেনারেটিভ ল্যাঙ্গুয়েজ মডেল যেটি ৬ বিলিয়ন প্যারামিটার দিয়ে ট্রেন করা হয়েছে।

Huggingface আমাদের একটি অন-ডিমান্ড লিমিটেড API দিয়েছে এই মডেলের সাথে কানেক্ট করার জন্য প্রায় বিনামূল্যে।

Huggingface-এ শুরু করতে, একটি ফ্রি অ্যাকাউন্ট তৈরি করুন। আপনার সেটিংসে, একটি নতুন অ্যাক্সেস টোকেন জেনারেট করুন। ৩০k টোকেন পর্যন্ত, Huggingface ইনফরেন্স API-এ ফ্রি অ্যাক্সেস দেয়।

আপনার API ইউজেজ মনিটর করুন এখানে। নিশ্চিত করুন আপনি এই টোকেনটি নিরাপদ রাখছেন এবং পাবলিকলি এক্সপোজ করছেন না।

নোট: আমরা HTTP কানেকশন ব্যবহার করব API-র সাথে যোগাযোগ করতে কারণ আমরা একটি ফ্রি অ্যাকাউন্ট ব্যবহার করছি। কিন্তু PRO Huggingface অ্যাকাউন্ট WebSockets সাপোর্ট করে স্ট্রিমিং, প্যারাললিজম এবং ব্যাচ জবস দেখুন।

এটি মডেল এবং আমাদের চ্যাট অ্যাপ্লিকেশনের মধ্যে রেসপন্স টাইাম সগনিফিক্যান্টলি উন্নত করতে পারে, এবং আমি আশা করি এই মেথডটি একটি ফলো-আপ আর্টিকেলে কভার করব।

ল্যাঙ্গুয়েজ মডেলের সাথে ইন্টারঅ্যাক্ট কীভাবে করবেন

প্রথমে, আমরা Huggingface কানেকশন ক্রেডেনশিয়ালস .env ফাইলে আমাদের worker ডিরেক্টরির মধ্যে যোগ করি।

export HUGGINFACE_INFERENCE_TOKEN=<HUGGINGFACE ACCESS TOKEN>
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):
        """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

আপনার .env ফাইল নিচের মত দেখতে হবে:

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

পরবর্তী, 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 দ্বারা প্রদত্ত নতুন মেসেজকে মেসেজ অ্যারেতে অ্যাপেন্ড করে।

লক্ষ্য করুন যে মেসেজ অ্যারে অ্যাক্সেস করতে, আমাদের .messages আর্গুমেন্ট হিসেবে দিতে হবে Path-এ। আপনার মেসেজ ডেটার যদি ভিন্ন/নেস্টেড স্ট্রাকচার থাকে, শুধু সেই অ্যারেতে পাথ দিন যাতে আপনি নতুন ডেটা অ্যাপেন্ড করতে চান।

এই মেথড টেস্ট করতে, 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)

আমরা একটি হার্ড-কোডেড মেসেজ ক্যাশে পাঠাচ্ছি, এবং ক্যাশ থেকে চ্যাট হিস্টরি পাচ্ছি। যখন আপনি worker ডিরেক্টরির মধ্যে টার্মিনালে 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 মডেলে পাঠাতে, এবং ইনপুট আপডেট করতে শেষ ৪টি মেসেজ দিয়ে যা ক্লায়েন্ট এবং মডেলের মধ্যে পাঠানো হয়েছে।

প্রথম আমাদের 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
    )

তারপর worker ডিরেক্টরিতে main.py-এ main ফাংশন আপডেট করুন, এবং python main.py চালান Redis ডাটাবেসে নতুন রেজাল্ট দেখতে।

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 ফাংশন আপডেট করতে হবে নতুন মেসেজ ক্যাশে যোগ করতে, ক্যাশ থেকে পূর্বের ৪টি মেসেজ পড়তে, এবং তারপর query মেথড ব্যবহার করে মডেলে API কল করতে। এর পেলোড থাকবে শেষ ৪টি মেসেজের একটি কম্পোজিট স্ট্রিং।

আপনি সবসময় হিস্টরিতে কতগুলো মেসেজ এক্সট্রাক্ট করতে চান সেটি টিউন করতে পারেন, কিন্তু আমি মনে করি ৪টি মেসেজ একটি ডেমোর জন্য খুব ভালো নম্বর।

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

উপরের কোডে, আমরা ক্যাশে নতুন মেসেজ ডেটা যোগ করি। এই মেসেজ চূড়ান্তভাবে মেসেজ কিউ থেকে আসবে। পরবর্তি আমরা ক্যাশ থেকে চ্যাট হিস্টরি পাই, যেটি এখন সাম্প্রতিক যোগ করা ডেটাও ইনক্লুড করবে।

লক্ষ্য করুন যে আমরা একই হার্ড-কোডেড টোকেন ব্যবহার করছি ক্যাশে যোগ করতে এবং ক্যাশ থেকে পেতে, অস্থায়ীভাবে শুধু এটি টেস্ট আউট করতে।

পরবর্তী, আমরা ক্যাশ ডেটা ট্রিম করি এবং শুধু শেষ ৪টি আইটেম এক্সট্রাক্ট করি। তারপর আমরা ইনপুট ডেটা কনসোলিডেট করি msg এক্সট্রাক্ট করে একটি লিস্টে এবং একসাথে জয়েন করি একটি খালি স্ট্রিং-এ।

শেষে, আমরা বট রেসপন্সের জন্য একটি নতুন Message ইন্সট্যান্স তৈরি করি এবং রেসপন্স ক্যাশে যোগ করি source হিসেবে "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 ফাইল নিচের মতো আপডেট করুন একটি while লুপ দিয়ে যাতে মেসেজ চ্যানেলের সাথে কানেকশন জীবিত থাকে, এইভাবে:

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)
                    # একটি নতুন মেসেজ ইন্সট্যান্স তৈরি করুন এবং ক্যাশে যোগ করুন, source হিসেবে human স্পেসিফাই করে
                    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)
                    # মেসেজ ইনপুট ক্লিন করুন এবং 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())
                    # মেসেজ কিউ থেকে ডিলিট করুন প্রসেস হওয়ার পর
                    await consumer.delete_message(stream_channel="message_channel", message_id=message_id)

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

এটি কয়েকটু বড় আপডেট, তাই ধাপে ধাপে নিই:

আমরা একটি while True লুপ ব্যবহার করি যাতে ওয়ার্কার অনলাইন থাকতে পারে মেসেজ কিউ থেকে মেসেজ শুনতে।

পরবর্তী, আমরা message_channel থেকে নতুন মেসেজ অপেক্ষা করি আমাদের consume_stream মেথড কল করে। যদি কিউতে একটি মেসেজ থাকে, আমরা message_id, token, এবং মেসেজ এক্সট্রাক্ট করি। তারপর Message ক্লাসের একটি নতুন ইন্সট্যান্স তৈরি করি, মেসেজ ক্যাশে যোগ করি, এবং তারপর শেষ ৪টি মেসেজ পাই। এটিকে GPT মডেল query মেথডের ইনপুট হিসেবে সেট করি।

একবার রেসপন্স পেলে, আমরা তারপর রেসপন্স ক্যাশে যোগ করি add_message_to_cache মেথড ব্যবহার করে, তারপর মেসেজ কিউ থেকে ডিলিট করি।

AI রেসপন্স দিয়ে চ্যাট ক্লায়েন্ট আপডেট কীভাবে করবেন

এতক্ষন, আমরা ক্লায়েন্ট থেকে একটি চ্যাট মেসেজ পাঠাচ্ছি message_channel-এ (যেটি ওয়ার্কার দ্বারা রিসিভ হয় যা 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 ফাংশন আপডেট করুন প্রডিউসার ইনিশিয়ালাইজ করতে, একটি স্ট্রিম ডেটা তৈরি করতে, এবং রেসপন্স একটি response_channel-এ পাঠাতে add_to_stream মেথড ব্যবহার করে:

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]
                    # একটি নতুন মেসেজ ইন্সট্যান্স তৈরি করুন এবং ক্যাশে যোগ করুন, source হিসেবে human স্পেসিফাই করে
                    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)
                    # মেসেজ ইনপুট ক্লিন করুন এবং 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())
                    # মেসেজ কিউ থেকে ডিলিট করুন প্রসেস হওয়ার পর
                    await consumer.delete_message(stream_channel="message_channel", message_id=message_id)

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

পরবর্তী, আমাদের ক্লায়েন্টকে জানাতে হবে যখন আমরা ওয়ার্কার থেকে রেসপন্স পাই /chat সকেট এন্ডপয়েন্টে। আমরা এটি করব রেসপন্স স্ট্রিম শোনা দিয়ে। আমাদের এখানে একটি while লুপ ইনক্লুড করার প্রয়োজন নেই কারণ সকেট শোনবে যতক্ষণ কানেকশন খোলা থাকে।

লক্ষ্য করুন যে আমাদের ও চেক করতে হবে রেসপন্স কোন ক্লায়েন্টের জন্য লজিক যোগ করে চেক করে যে কানেক্টেড টোকেন রেসপন্সের টোকেনের সমান কিনা। তারপর রেসপন্স কিউ-এ মেসেজ ডিলিট করি একবার পড়া হলে।

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)

রিফ্রেশ টোকেন

শেষে, আমাদের /refresh_token এন্ডপয়েন্ট আপডেট করতে হবে চ্যাট হিস্টরি Redis ডাটাবেস থেকে পেতে আমাদের Cache ক্লাস ব্যবহার করে।

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 রিকোয়েস্ট পাঠাব, এন্ডপয়েন্ট Redis ডাটাবেস থেকে ডেটা ফেচ করবে।

যদি টোকেন টাইম আউট না হয়ে থাকে, ডেটা ইউজারকে পাঠানো হবে। অথবা ৪০০ রেসপন্স পাঠাবে যদি টোকেন না পাওয়া যায়।

Postman-এ মাল্টিপল ক্লায়েন্ট দিয়ে চ্যাট টেস্ট কীভাবে করবেন

শেষে, আমরা চ্যাট সিস্টেম টেস্ট করব Postman-এ মাল্টিপল চ্যাট সেশন তৈরি করে, Postman-এ মাল্টিপল ক্লায়েন্ট কানেক্ট করে, এবং ক্লায়েন্টগুলোর ওপর বটের সাথে চ্যাট করে।

শেষে, আমরা ক্লায়েন্টদের চ্যাট হিস্টরি পেতে চেষ্টা করব এবং আশা করব একটি প্রপার রেসপন্স পাওয়া।

চ্যাট টেস্ট রেজাল্ট

রিক্যাপ

আমরা যা চ্যাট সিস্টেমে অর্জন করেছি তার একটি কুইক রিক্যাপ নিই। চ্যাট ক্লায়েন্ট প্রতিটি চ্যাট সেশনের জন্য একটি টোকেন তৈরি করে একটি ক্লায়েন্টের সাথে। এই টোকেনটি প্রতিটি ক্লায়েন্ট আইডেন্টিফাই করতে ব্যবহৃত হয়, এবং ওয়েব সার্ভারে কানেক্টেড ক্লায়েন্টরা দ্বারা পাঠানো প্রতিটি মেসেজ একটি Redis চ্যানেলে (message_channel) কিউ হয়, টোকেন দিয়ে আইডেন্টিফাইড হয়ে।

আমাদের ওয়ার্কার এনভায়রনমেন্ট এই চ্যানেল থেকে পড়ে। এটিকে কোনো ধারণা নেই ক্লায়েন্ট কে (সেটা একটি ইউনিক টোকেন ছাড়া) এবং কিউ-এ মেসেজটি ব্যবহার করে Huggingface inference API-তে রিকোয়েস্ট পাঠায়।

যখন এটি একটি রেসপন্স পায়, রেসপন্স একটি রেসপন্স চ্যানেলে যোগ হয় এবং চ্যাট হিস্টরি আপডেট হয়। ক্লায়েন্ট response_channel শুনছে সেটি তাত্ক্ষণিক রেসপন্স ক্লায়েন্টকে পাঠায় একবার এটি তার টোকেন সহ একটি রেসপন্স পাওয়ার পর।

যদি সকেট এখনও খোলা থাকে, এই রেসপন্স পাঠানো হয়। যদি সকেট বন্ধ হয়ে যায়, আমরা নিশ্চিত যে রেসপন্স সংরক্ষিত আছে কারণ রেসপন্স চ্যাট হিস্টরিতে যোগ করা হয়েছে। ক্লায়েন্ট হিস্টরি পেতে পারে, এমনকি পেজ রিফ্রেশ হলে বা লост কানেকশনের ইভেন্টে।

এতদূর পৌঁছানোর জন্য অভিনন্দন! আপনি একটি কাজ করার চ্যাট সিস্টেম বানাতে পেরেছেন।

ফলো-আপ আর্টিকেলগুলোর মধ্যে, আমি ফোকাস করব চ্যাট ইউজার ইন্টারফেস বানাতে ক্লায়েন্টের জন্য, ইউনিট এবং ফাংশনাল টেস্ট তৈরি করতে, আমাদের ওয়ার্কার এনভায়রনমেন্ট ফাইন-টিউন করতে ফাস্টার রেসপন্স টাইমের জন্য WebSockets এবং অ্যাসিঙ্ক্রোনাস রিকোয়েস্টস দিয়ে, এবং চূড়ান্তভাবে চ্যাট অ্যাপ্লিকেশন AWS-এ ডিপ্লয় করতে।

এই আর্টিকেল Python, React, Huggingface, Redis, এবং আরও টুলস দিয়ে ফুল-স্ট্যাক ইন্টেলিজেন্ট চ্যাটবট বানানোর একটি সিরিজের অংশ। আপনি পূর্ণ সিরিজ ফলো করতে পারেন আমার ব্লগে: blog.stephensanwo.dev - AI ChatBot Series**

আপনি পূর্ণ রিপোজিটরি ডাউনলোড করতে পারেন আমার GitHub রিপোজিটরি থেকে

আমি এই টিউটোরিয়াল Redis-এর সহযোগিতায় লিখেছি। Redis-এ শুরু করতে সাহায্য দরকার? নিচের রিসোর্সগুলো ট্রাই করুন:

• Redis Cloud বিনামূল্যে ট্রাই করুন
• এই ভিডিও দেখুন Redis Cloud-এর সুবিধা অন্যান্য Redis প্রোভাইডারদের চেয়ে
• Redis Developer Hub - Redis সম্পর্কে টুলস, গাইডস, এবং টিউটোরিয়াল
• RedisInsight ডেস্কটপ GUI