? ¿Quiere crear un chatbot en

Esta guía le muestra cómo crear un chatbot interactivo con tecnología LLM con DBOS y LangChain e implementarlo sin servidor en DBOS Cloud.

Puedes ver el chatbot en vivo aquí.

Además de chatear, este bot muestra la cantidad de tiempo de CPU y de reloj de pared consumidos por tus solicitudes.
Mientras chateas, notarás rápidamente que, si bien tus solicitudes pueden tardar mucho tiempo, consumen muy poca CPU.
Esto se debe a que pasan la mayor parte del tiempo inactivos esperando que responda el LLM.
Esta brecha explica por qué DBOS es 50 veces más rentable que otras plataformas sin servidor para cargas de trabajo de IA, porque DBOS factura solo por el tiempo de CPU que realmente consume, mientras que otras plataformas facturan por la duración total de la solicitud.

Todo el código fuente está disponible en GitHub.

Importar e inicializar la aplicación

Comencemos con las importaciones y la inicialización de DBOS.
También configuraremos FastAPI para atender solicitudes 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)

Configurando LangChain

A continuación, configuremos Langchain.
Usaremos Langchain para responder cada mensaje de chat usando el modelo gpt-3.5-turbo de OpenAI.
Configuraremos LangChain para almacenar el historial de mensajes en Postgres para que persista después de reiniciar la aplicación.

Para divertirnos, también instruyamos a nuestro chatbot para que hable como un pirata.

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

Manejo de chats

¡Ahora charlemos!
Primero escribiremos el punto final que maneja cada solicitud de chat.

Este punto final es un flujo de trabajo DBOS con tres pasos:

1. Almacenar el mensaje de chat entrante en Postgres.
2. Utilice LangChain para consultar el LLM para responder al mensaje de chat.
3. Almacenar la respuesta en Postgres.

También registra la duración total de cada solicitud en un búfer en memoria.

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}

A continuación, escribamos la función que realmente consulta a LangChain para cada mensaje nuevo.
Utiliza su nombre de usuario como ID del hilo para que diferentes usuarios puedan tener diferentes hilos de conversación.

Anotamos esta función con @DBOS.step() para marcarla como un paso en nuestro flujo de trabajo de chat.

@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

También necesitamos un punto final de historial que recupere todos los chats anteriores de la base de datos para un usuario en particular.

Esta función se llama cuando iniciamos el chatbot para que pueda mostrar tu historial de chat.

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

Luego, usemos SQLAlchemy para escribir las funciones que escriben y leen chats en la base de datos.
Anotamos estas funciones con @DBOS.transaction() para acceder a la conexión de base de datos administrada 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]

Además, proporcionemos la interfaz de la aplicación desde un archivo HTML usando FastAPI.
En producción, recomendamos usar DBOS principalmente para el backend, con el frontend implementado en otro lugar.

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

Seguimiento del uso de la aplicación

Por último, escribamos un código para realizar un seguimiento del tiempo de CPU y del reloj de pared consumido por sus solicitudes para que podamos mostrar esas métricas en la interfaz de usuario de la aplicación.
Este código se ejecuta una vez por segundo en un hilo en segundo plano.

Hacemos un seguimiento del consumo de CPU de este proceso utilizando psutil.
Realizamos un seguimiento del tiempo del reloj de pared registrando la duración de un extremo a otro de cada solicitud.

Cuando inicies la aplicación por primera vez, notarás un pequeño consumo residual de CPU del servidor HTTP.
Sin embargo, cuando empieces a chatear, verás rápidamente que cada chat solo consume ~10 ms de tiempo de CPU, pero entre 1 y 2 segundos de tiempo de reloj de pared.
Esta brecha explica por qué DBOS es 50 veces más barato que otras plataformas sin servidor para cargas de trabajo de IA, porque DBOS factura solo por el tiempo de CPU que realmente consume, mientras que otras plataformas facturan por la duración total de la solicitud.

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

¡Pruébalo tú mismo!

Crear una cuenta OpenAI

Para ejecutar esta aplicación, necesitas una cuenta de desarrollador de OpenAI.
Obtenga una clave API aquí y configure un método de pago para su cuenta aquí.
Este bot usa gpt-3.5-turbo para generar texto.
Asegúrate de tener algunos créditos (~$1) para usarlo.

Establezca su clave API como una variable de entorno:

export OPENAI_API_KEY=

Implementación en la nube

Para implementar esta aplicación en DBOS Cloud, primero instale la CLI de DBOS Cloud (requiere Node):

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

Luego clona el repositorio dbos-demo-apps e implementa:

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

Este comando genera una URL: ¡visítala para ver tu chatbot!
También puede visitar DBOS Cloud Console para ver el estado y los registros de su aplicación.

Ejecutando localmente

Primero, clona e ingresa al repositorio dbos-demo-apps:

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

Luego crea un entorno virtual:

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

DBOS requiere una base de datos Postgres.
Si aún no tienes uno, puedes iniciar uno con Docker:

export PGPASSWORD=dbos
python3 start_postgres_docker.py

Luego ejecuta la aplicación en el entorno virtual:

pip install -r requirements.txt
dbos migrate
dbos start

¡Visita http://localhost:8000 para ver tu chatbot!

Próximos pasos

Vea cómo DBOS puede hacer que sus aplicaciones sean más escalables y resistentes:

• Utilice una ejecución duradera para escribir flujos de trabajo a prueba de fallas.
• Utilice colas para administrar con elegancia los límites de tasa de API.
• Utilice flujos de trabajo programados para ejecutar sus funciones en intervalos recurrentes.
• ¿Quieres saber qué puedes crear con DBOS? Explora otras aplicaciones de ejemplo.