? Vous souhaitez créer un chatbot en moins de 100 lignes de Python et l'héberger sans serveur sur le cloud avec une seule commande ?

Ce guide vous montre comment créer un chatbot interactif alimenté par LLM avec DBOS et LangChain et le déployer sans serveur sur DBOS Cloud.

Vous pouvez voir le chatbot en direct ici.

En plus de discuter, ce bot affiche à la fois la quantité de temps CPU et le temps d'horloge murale consommés par vos requêtes.
En discutant, vous remarquerez rapidement que même si vos requêtes peuvent prendre beaucoup de temps, elles consomment très peu de CPU.
C'est parce qu'ils passent la plupart de leur temps à attendre la réponse du LLM.
Cet écart explique pourquoi DBOS est 50 fois plus rentable que les autres plates-formes sans serveur pour les charges de travail d'IA, car DBOS facture uniquement le temps CPU que vous consommez réellement, tandis que d'autres plates-formes facturent la durée totale de la requête.

Tout le code source est disponible sur GitHub.

Importer et initialiser l'application

Commençons par les importations et l'initialisation de DBOS.
Nous allons également configurer FastAPI pour répondre aux requêtes 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)

Configuration de LangChain

Ensuite, configurons Langchain.
Nous utiliserons Langchain pour répondre à chaque message de discussion en utilisant le modèle gpt-3.5-turbo d'OpenAI.
Nous allons configurer LangChain pour stocker l'historique des messages dans Postgres afin qu'il persiste lors des redémarrages de l'application.

Pour nous amuser, demandons également à notre chatbot de parler comme un pirate.

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

Gestion des discussions

Maintenant, discutons !
Nous allons d’abord écrire le point de terminaison qui gère chaque demande de chat.

Ce point de terminaison est un workflow DBOS en trois étapes :

1. Stockez le message de discussion entrant dans Postgres.
2. Utilisez LangChain pour interroger le LLM afin de répondre au message de discussion.
3. Stockez la réponse dans Postgres.

Il enregistre également la durée totale de chaque requête dans un tampon en mémoire.

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}

Ensuite, écrivons la fonction qui interroge réellement LangChain pour chaque nouveau message.
Il utilise votre nom d'utilisateur comme identifiant de fil de discussion afin que différents utilisateurs puissent avoir différents fils de conversation.

Nous annotons cette fonction avec @DBOS.step() pour la marquer comme une étape de notre flux de travail de discussion.

@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

Nous avons également besoin d'un point de terminaison d'historique qui récupère toutes les discussions passées de la base de données pour un utilisateur particulier.

Cette fonction est appelée lorsque nous démarrons le chatbot afin qu'il puisse afficher l'historique de vos discussions.

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

Ensuite, utilisons SQLAlchemy pour écrire les fonctions qui écrivent et lisent les discussions dans la base de données.
Nous annotons ces fonctions avec @DBOS.transaction() pour accéder à la connexion à la base de données gérée de DBOS.

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

De plus, servons l'interface de l'application à partir d'un fichier HTML à l'aide de FastAPI.
En production, nous vous recommandons d'utiliser DBOS principalement pour le backend, votre frontend étant déployé ailleurs.

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

Suivi de l'utilisation des applications

Enfin, écrivons du code pour suivre le temps CPU et le temps d'horloge murale consommés par vos requêtes afin que nous puissions afficher ces métriques dans l'interface utilisateur de l'application.
Ce code s'exécute une fois par seconde dans un thread en arrière-plan.

Nous suivons la consommation CPU de ce processus à l'aide de psutil.
Nous suivons l'heure murale en enregistrant la durée de bout en bout de chaque demande.

Lorsque vous démarrez l'application pour la première fois, vous remarquerez une petite consommation résiduelle de processeur du serveur HTTP.
Cependant, lorsque vous commencez à discuter, vous constaterez rapidement que chaque discussion ne consomme qu'environ 10 ms de temps CPU, mais 1 à 2 secondes de temps d'horloge murale.
Cet écart explique pourquoi DBOS est 50 fois moins cher que les autres plates-formes sans serveur pour les charges de travail d'IA, car DBOS facture uniquement le temps CPU que vous consommez réellement, tandis que d'autres plates-formes facturent la durée totale de la requête.

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

Essayez-le vous-même !

Créer un compte OpenAI

Pour exécuter cette application, vous avez besoin d'un compte développeur OpenAI.
Obtenez une clé API ici et configurez un mode de paiement pour votre compte ici.
Ce bot utilise gpt-3.5-turbo pour la génération de texte.
Assurez-vous d'avoir des crédits (~ 1 $) pour l'utiliser.

Définissez votre clé API comme variable d'environnement :

export OPENAI_API_KEY=

Déploiement vers le cloud

Pour déployer cette application sur DBOS Cloud, installez d'abord DBOS Cloud CLI (nécessite Node) :

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

Clonez ensuite le référentiel dbos-demo-apps et déployez :

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

Cette commande génère une URL : visitez-la pour voir votre chatbot !
Vous pouvez également visiter la console DBOS Cloud pour voir l'état et les journaux de votre application.

Exécution locale

Tout d'abord, clonez et entrez dans le référentiel dbos-demo-apps :

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

Créez ensuite un environnement virtuel :

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

DBOS nécessite une base de données Postgres.
Si vous n'en avez pas déjà un, vous pouvez en démarrer un avec Docker :

export PGPASSWORD=dbos
python3 start_postgres_docker.py

Exécutez ensuite l'application dans l'environnement virtuel :

pip install -r requirements.txt
dbos migrate
dbos start

Visitez http://localhost:8000 pour voir votre chatbot !

Prochaines étapes

Découvrez comment DBOS peut rendre vos applications plus évolutives et plus résilientes :

• Utilisez une exécution durable pour écrire des flux de travail résistants aux pannes.
• Utilisez les files d'attente pour gérer efficacement les limites de débit de l'API.
• Utilisez des flux de travail planifiés pour exécuter vos fonctions à intervalles récurrents.
• Vous voulez savoir ce que vous pouvez créer avec DBOS ? Découvrez d'autres exemples d'applications.