? Möchten Sie einen Chatbot in

Diese Anleitung zeigt Ihnen, wie Sie einen interaktiven LLM-basierten Chatbot mit DBOS und LangChain erstellen und ihn serverlos in der DBOS Cloud bereitstellen.

Hier können Sie den Chatbot live sehen.

Zusätzlich zum Chatten zeigt dieser Bot sowohl die CPU-Zeit als auch die Arbeitszeit an, die Ihre Anfragen verbrauchen.
Während Sie chatten, werden Sie schnell feststellen, dass Ihre Anfragen zwar lange dauern, aber nur sehr wenig CPU verbrauchen.
Das liegt daran, dass sie die meiste Zeit untätig damit verbringen, auf die Antwort des LLM zu warten.
Diese Lücke erklärt, warum DBOS für KI-Workloads 50-mal kosteneffizienter ist als andere serverlose Plattformen – weil DBOS nur die tatsächlich verbrauchte CPU-Zeit in Rechnung stellt, während andere Plattformen die gesamte Anforderungsdauer in Rechnung stellen.

Der gesamte Quellcode ist auf GitHub verfügbar.

Importieren und initialisieren Sie die App

Beginnen wir mit Importen und der Initialisierung von DBOS.
Wir werden auch FastAPI einrichten, um HTTP-Anfragen zu bedienen.

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)

LangChain einrichten

Als nächstes richten wir Langchain ein.
Wir werden Langchain verwenden, um jede Chat-Nachricht mit dem gpt-3.5-turbo-Modell von OpenAI zu beantworten.
Wir konfigurieren LangChain so, dass der Nachrichtenverlauf in Postgres gespeichert wird, sodass er über App-Neustarts hinweg bestehen bleibt.

Zum Spaß weisen wir unseren Chatbot auch an, wie ein Pirat zu sprechen.

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

Umgang mit Chats

Jetzt lasst uns chatten!
Wir schreiben zunächst den Endpunkt, der jede Chat-Anfrage verarbeitet.

Dieser Endpunkt ist ein DBOS-Workflow mit drei Schritten:

1. Speichern Sie die eingehende Chat-Nachricht in Postgres.
2. Verwenden Sie LangChain, um das LLM abzufragen, um auf die Chat-Nachricht zu antworten.
3. Speichern Sie die Antwort in Postgres.

Es zeichnet außerdem die Gesamtdauer jeder Anfrage in einem In-Memory-Puffer auf.

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}

Als nächstes schreiben wir die Funktion, die LangChain tatsächlich nach jeder neuen Nachricht abfragt.
Es verwendet Ihren Benutzernamen als Thread-ID, sodass verschiedene Benutzer unterschiedliche Konversationsthreads führen können.

Wir kommentieren diese Funktion mit @DBOS.step(), um sie als Schritt in unserem Chat-Workflow zu markieren.

@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

Wir benötigen außerdem einen Verlaufsendpunkt, der alle vergangenen Chats für einen bestimmten Benutzer aus der Datenbank abruft.

Diese Funktion wird aufgerufen, wenn wir den Chatbot starten, damit er Ihren Chatverlauf anzeigen kann.

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

Dann verwenden wir SQLAlchemy, um die Funktionen zu schreiben, die Chats in die Datenbank schreiben und Chats aus der Datenbank lesen.
Wir kommentieren diese Funktionen mit @DBOS.transaction(), um auf die verwaltete Datenbankverbindung von DBOS zuzugreifen.

@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]

Außerdem stellen wir das Frontend der App mithilfe von FastAPI aus einer HTML-Datei bereit.
In der Produktion empfehlen wir, DBOS hauptsächlich für das Backend zu verwenden und Ihr Frontend an anderer Stelle bereitzustellen.

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

App-Nutzung verfolgen

Schreiben wir abschließend Code, um die von Ihren Anfragen verbrauchte CPU- und Arbeitszeit zu verfolgen, damit wir diese Metriken in der Benutzeroberfläche der App anzeigen können.
Dieser Code wird einmal pro Sekunde in einem Hintergrundthread ausgeführt.

Wir verfolgen den CPU-Verbrauch dieses Prozesses mit psutil.
Wir verfolgen die Arbeitszeit, indem wir die End-to-End-Dauer jeder Anfrage aufzeichnen.

Wenn Sie die App zum ersten Mal starten, werden Sie einen geringen Rest-CPU-Verbrauch vom HTTP-Server bemerken.
Wenn Sie jedoch mit dem Chatten beginnen, werden Sie schnell feststellen, dass jeder Chat nur ca. 10 ms CPU-Zeit, aber 1–2 Sekunden Arbeitszeit in Anspruch nimmt.
Diese Lücke erklärt, warum DBOS für KI-Workloads 50-mal günstiger ist als andere serverlose Plattformen – weil DBOS nur die tatsächlich verbrauchte CPU-Zeit in Rechnung stellt, während andere Plattformen die gesamte Anforderungsdauer in Rechnung stellen.

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]),
    }

Probieren Sie es selbst aus!

Erstellen eines OpenAI-Kontos

Um diese App auszuführen, benötigen Sie ein OpenAI-Entwicklerkonto.
Besorgen Sie sich hier einen API-Schlüssel und richten Sie hier eine Zahlungsmethode für Ihr Konto ein.
Dieser Bot verwendet gpt-3.5-turbo zur Textgenerierung.
Stellen Sie sicher, dass Sie über ein gewisses Guthaben (ca. 1 $) verfügen, um es zu verwenden.

Legen Sie Ihren API-Schlüssel als Umgebungsvariable fest:

export OPENAI_API_KEY=

Bereitstellung in der Cloud

Um diese App in der DBOS Cloud bereitzustellen, installieren Sie zunächst die DBOS Cloud CLI (erfordert Node):

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

Klonen Sie dann das dbos-demo-apps-Repository und stellen Sie Folgendes bereit:

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

Dieser Befehl gibt eine URL aus – besuchen Sie sie, um Ihren Chatbot zu sehen!
Sie können auch die DBOS Cloud Console besuchen, um den Status und die Protokolle Ihrer App anzuzeigen.

Lokal ausgeführt

Klonen Sie zunächst das dbos-demo-apps-Repository und geben Sie es ein:

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

Dann erstellen Sie eine virtuelle Umgebung:

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

DBOS erfordert eine Postgres-Datenbank.
Wenn Sie noch keines haben, können Sie eines mit Docker starten:

export PGPASSWORD=dbos
python3 start_postgres_docker.py

Führen Sie dann die App in der virtuellen Umgebung aus:

pip install -r requirements.txt
dbos migrate
dbos start

Besuchen Sie http://localhost:8000, um Ihren Chatbot zu sehen!

Nächste Schritte

Sehen Sie sich an, wie DBOS Ihre Anwendungen skalierbarer und robuster machen kann:

• Verwenden Sie dauerhafte Ausführung, um absturzsichere Workflows zu schreiben.
• Verwenden Sie Warteschlangen, um API-Ratenlimits elegant zu verwalten.
• Verwenden Sie geplante Workflows, um Ihre Funktionen in wiederkehrenden Abständen auszuführen.
• Möchten Sie erfahren, was Sie mit DBOS erstellen können? Entdecken Sie weitere Beispielanwendungen.