? क्या आप पायथन की

यह मार्गदर्शिका आपको दिखाती है कि डीबीओएस और लैंगचेन के साथ एक इंटरैक्टिव एलएलएम-संचालित चैटबॉट कैसे बनाया जाए और इसे सर्वर रहित तरीके से डीबीओएस क्लाउड पर कैसे तैनात किया जाए।

आप चैटबॉट को यहां लाइव देख सकते हैं।

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

सभी स्रोत कोड GitHub पर उपलब्ध हैं।

ऐप को आयात करें और प्रारंभ करें

आइए आयात और डीबीओएस आरंभीकरण से शुरुआत करें।
हम HTTP अनुरोधों को पूरा करने के लिए फास्टएपीआई भी स्थापित करेंगे।

import os
import threading
import time
from collections import deque

import psutil
from dbos import DBOS
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from langchain_core.messages import HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import START, MessagesState, StateGraph
from psycopg_pool import ConnectionPool
from pydantic import BaseModel

from .schema import chat_history

app = FastAPI()
dbos = DBOS(fastapi=app)

लैंगचेन की स्थापना

इसके बाद, आइए लैंगचैन स्थापित करें।
हम OpenAI के gpt-3.5-टर्बो मॉडल का उपयोग करके प्रत्येक चैट संदेश का उत्तर देने के लिए लैंगचैन का उपयोग करेंगे।
हम पोस्टग्रेज़ में संदेश इतिहास को संग्रहीत करने के लिए लैंगचेन को कॉन्फ़िगर करेंगे ताकि यह ऐप पुनरारंभ होने पर बना रहे।

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

def create_langchain():
    # We use gpt-3.5-turbo as our model.
    model = ChatOpenAI(model="gpt-3.5-turbo")

    # This prompt instructs the model how to act. We'll tell it to talk like a pirate!
    prompt = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                "You talk like a pirate. Answer all questions to the best of your ability.",
            ),
            MessagesPlaceholder(variable_name="messages"),
        ]
    )

    # This function tells LangChain to invoke our model with our prompt.
    def call_model(state: MessagesState):
        chain = prompt | model
        response = chain.invoke(state)
        return {"messages": response}

    # Create a checkpointer LangChain can use to store message history in Postgres.
    db = DBOS.config["database"]
    connection_string = f"postgresql://{db['username']}:{db['password']}@{db['hostname']}:{db['port']}/{db['app_db_name']}"
    pool = ConnectionPool(connection_string)
    checkpointer = PostgresSaver(pool)

    # Finally, construct and return the graph LangChain uses to respond to each message.
    # This chatbot uses a simple one-node graph that just calls the model.
    graph = StateGraph(state_schema=MessagesState)
    graph.add_node("model", call_model)
    graph.add_edge(START, "model")
    return graph.compile(checkpointer=checkpointer)


chain = create_langchain()

चैट संभालना

अब, चलो चैट करें!
हम पहले वह समापन बिंदु लिखेंगे जो प्रत्येक चैट अनुरोध को संभालता है।

यह समापन बिंदु तीन चरणों वाला एक DBOS वर्कफ़्लो है:

1. आने वाले चैट संदेश को Postgres में संग्रहीत करें।
2. चैट संदेश का जवाब देने के लिए एलएलएम से पूछताछ करने के लिए लैंगचेन का उपयोग करें।
3. प्रतिक्रिया को पोस्टग्रेज़ में संग्रहीत करें।

यह इन-मेमोरी बफ़र में प्रत्येक अनुरोध की कुल अवधि को भी रिकॉर्ड करता है।

class ChatSchema(BaseModel):
    message: str
    username: str


@app.post("/chat")
@DBOS.workflow()
def chat_workflow(chat: ChatSchema):
    start_time = time.time()
    insert_chat(chat.username, chat.message, True)
    response = query_model(chat.message, chat.username)
    insert_chat(chat.username, response, False)
    elapsed_time = time.time() - start_time
    wallclock_times_buffer.append((time.time(), elapsed_time))
    return {"content": response, "isUser": True}

इसके बाद, आइए उस फ़ंक्शन को लिखें जो वास्तव में प्रत्येक नए संदेश के लिए लैंगचेन से पूछताछ करता है।
यह आपके उपयोगकर्ता नाम को थ्रेड आईडी के रूप में उपयोग करता है ताकि विभिन्न उपयोगकर्ताओं के बीच बातचीत के अलग-अलग थ्रेड हो सकें।

हम इस फ़ंक्शन को हमारे चैट वर्कफ़्लो में एक चरण के रूप में चिह्नित करने के लिए @DBOS.step() के साथ एनोटेट करते हैं।

@DBOS.step()
def query_model(message: str, username: str) -> str:
    config = {"configurable": {"thread_id": username}}
    input_messages = [HumanMessage(message)]
    output = chain.invoke({"messages": input_messages}, config)
    return output["messages"][-1].content

हमें एक इतिहास समापन बिंदु की भी आवश्यकता है जो किसी विशेष उपयोगकर्ता के लिए डेटाबेस से सभी पिछली चैट को पुनर्प्राप्त करता है।

जब हम चैटबॉट शुरू करते हैं तो यह फ़ंक्शन कॉल किया जाता है ताकि यह आपका चैट इतिहास प्रदर्शित कर सके।

@app.get("/history/{username}")
def history_endpoint(username: str):
    return get_chats(username)

फिर, आइए उन कार्यों को लिखने के लिए SQLAlchemy का उपयोग करें जो डेटाबेस से चैट लिखते हैं और पढ़ते हैं।
हम DBOS के प्रबंधित डेटाबेस कनेक्शन तक पहुंचने के लिए इन कार्यों को @DBOS.transaction() के साथ एनोटेट करते हैं।

@DBOS.transaction()
def insert_chat(username: str, content: str, is_user: bool):
    DBOS.sql_session.execute(
        chat_history.insert().values(
            username=username, content=content, is_user=is_user
        )
    )


@DBOS.transaction()
def get_chats(username: str):
    stmt = (
        chat_history.select()
        .where(chat_history.c.username == username)
        .order_by(chat_history.c.created_at.asc())
    )
    result = DBOS.sql_session.execute(stmt)
    return [{"content": row.content, "isUser": row.is_user} for row in result]

इसके अतिरिक्त, आइए फास्टएपीआई का उपयोग करके एक HTML फ़ाइल से ऐप के फ्रंटएंड की सेवा करें।
उत्पादन में, हम मुख्य रूप से बैकएंड के लिए डीबीओएस का उपयोग करने की सलाह देते हैं, जिसमें आपका फ्रंटएंड कहीं और तैनात होता है।

@app.get("/")
def frontend():
    with open(os.path.join("html", "app.html")) as file:
        html = file.read()
    return HTMLResponse(html)

ट्रैकिंग ऐप उपयोग

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

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

जब आप पहली बार ऐप शुरू करते हैं, तो आप HTTP सर्वर से कुछ छोटी अवशिष्ट सीपीयू खपत देखेंगे।
हालाँकि, जैसे ही आप चैट करना शुरू करते हैं, आप तुरंत देखेंगे कि प्रत्येक चैट में केवल ~10 एमएस सीपीयू समय लगता है, लेकिन 1-2 सेकंड का वॉल-क्लॉक समय लगता है।
यह अंतर बताता है कि एआई वर्कलोड के लिए डीबीओएस अन्य सर्वर रहित प्लेटफार्मों की तुलना में 50 गुना सस्ता क्यों है - क्योंकि डीबीओएस केवल आपके द्वारा वास्तव में उपभोग किए गए सीपीयू समय के लिए बिल करता है, जबकि अन्य प्लेटफॉर्म कुल अनुरोध अवधि के लिए बिल करते हैं।

last_cpu_time_ms = 0
cpu_times_buffer = deque()
wallclock_times_buffer = deque()


def update_cpu_usage():
    while True:
        time.sleep(1)
        global last_cpu_time_ms
        # Every second, record CPU time consumed by this process
        # in the last second.
        process = psutil.Process()
        cpu_times = process.cpu_times()
        cpu_time = cpu_times.system   cpu_times.user
        time_consumed = cpu_time - last_cpu_time_ms
        if last_cpu_time_ms > 0:
            cpu_times_buffer.append((time.time(), time_consumed))
        last_cpu_time_ms = cpu_time
        # We only track usage in the last minute, so
        # pop measurements more than 60 seconds old.
        for buf in [cpu_times_buffer, wallclock_times_buffer]:
            while buf and time.time() - buf[0][0] > 60:
                buf.popleft()


threading.Thread(target=update_cpu_usage).start()


@app.get("/times")
def times_endpoint():
    return {
        "cpu_time": sum([t for _, t in cpu_times_buffer]),
        "wall_clock_time": sum([t for _, t in wallclock_times_buffer]),
    }

खुद कोशिश करना!

एक OpenAI खाता बनाना

इस ऐप को चलाने के लिए, आपको एक OpenAI डेवलपर खाते की आवश्यकता है।
यहां एक एपीआई कुंजी प्राप्त करें और यहां अपने खाते के लिए एक भुगतान विधि सेट करें।
यह बॉट टेक्स्ट जेनरेशन के लिए gpt-3.5-टर्बो का उपयोग करता है।
सुनिश्चित करें कि आपके पास इसका उपयोग करने के लिए कुछ क्रेडिट (~$1) हैं।

अपनी एपीआई कुंजी को एक पर्यावरण चर के रूप में सेट करें:

export OPENAI_API_KEY=

बादल पर तैनाती

इस ऐप को डीबीओएस क्लाउड पर तैनात करने के लिए, पहले डीबीओएस क्लाउड सीएलआई इंस्टॉल करें (नोड की आवश्यकता है):

npm i -g @dbos-inc/dbos-cloud

फिर dbos-डेमो-ऐप्स रिपॉजिटरी को क्लोन करें और तैनात करें:

git clone https://github.com/dbos-inc/dbos-demo-apps.git
cd python/chatbot
dbos-cloud app deploy

यह कमांड एक यूआरएल आउटपुट करता है—अपना चैटबॉट देखने के लिए इस पर जाएं!
आप अपने ऐप की स्थिति और लॉग देखने के लिए डीबीओएस क्लाउड कंसोल पर भी जा सकते हैं।

स्थानीय रूप से चल रहा है

सबसे पहले, क्लोन करें और dbos-demo-apps रिपॉजिटरी में प्रवेश करें:

git clone https://github.com/dbos-inc/dbos-demo-apps.git
cd python/chatbot

फिर एक आभासी वातावरण बनाएं:

python3 -m venv .venv
source .venv/bin/activate

डीबीओएस को एक पोस्टग्रेज डेटाबेस की आवश्यकता है।
यदि आपके पास पहले से कोई नहीं है, तो आप डॉकर के साथ एक शुरू कर सकते हैं:

export PGPASSWORD=dbos
python3 start_postgres_docker.py

फिर ऐप को वर्चुअल वातावरण में चलाएं:

pip install -r requirements.txt
dbos migrate
dbos start

अपना चैटबॉट देखने के लिए http://localhost:8000 पर जाएं!

अगले कदम

देखें कि डीबीओएस आपके एप्लिकेशन को कैसे अधिक स्केलेबल और लचीला बना सकता है:

• क्रैशप्रूफ वर्कफ़्लो लिखने के लिए टिकाऊ निष्पादन का उपयोग करें।
• एपीआई दर सीमा को शानदार ढंग से प्रबंधित करने के लिए कतारों का उपयोग करें।
• अपने कार्यों को आवर्ती अंतराल पर चलाने के लिए निर्धारित वर्कफ़्लो का उपयोग करें।
• क्या आप सीखना चाहते हैं कि आप डीबीओएस के साथ क्या बना सकते हैं? अन्य उदाहरण अनुप्रयोगों का अन्वेषण करें।