? Хотите создать чат-бота из

В этом руководстве показано, как создать интерактивного чат-бота на базе LLM с помощью DBOS и LangChain и бессерверно развернуть его в облаке DBOS.

Вы можете увидеть чат-бота вживую здесь.

Помимо чата, этот бот отображает количество процессорного времени и время настенных часов, затраченное на ваши запросы.
Во время общения вы быстро заметите, что, хотя ваши запросы могут занимать много времени, они потребляют очень мало процессора.
Это потому, что они проводят большую часть своего времени в ожидании ответа от LLM.
Этот пробел объясняет, почему DBOS в 50 раз более экономична, чем другие бессерверные платформы для рабочих нагрузок ИИ, поскольку DBOS выставляет счет только за фактически потребляемое вами процессорное время, в то время как другие платформы выставляют счет за общую продолжительность запроса.

Весь исходный код доступен на GitHub.

Импортируйте и инициализируйте приложение

Начнем с импорта и инициализации DBOS.
Мы также настроим FastAPI для обслуживания 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)

Настройка Лангчейна

Далее давайте настроим Langchain.
Мы будем использовать Langchain для ответа на каждое сообщение чата, используя модель OpenAI gpt-3.5-turbo.
Мы настроим LangChain для хранения истории сообщений в Postgres, чтобы она сохранялась при перезапуске приложения.

Для интереса давайте также поручим нашему чат-боту говорить как пират.

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. Используйте LangChain, чтобы запросить у LLM ответ на сообщение чата.
3. Сохраните ответ в Postgres.

Он также записывает общую продолжительность каждого запроса в буфере в памяти.

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}

Далее давайте напишем функцию, которая фактически запрашивает LangChain для каждого нового сообщения.
Он использует ваше имя пользователя в качестве идентификатора потока, поэтому разные пользователи могут вести разные темы разговора.

Мы аннотируем эту функцию с помощью @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.transaction() для доступа к управляемому соединению с базой данных 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]

Кроме того, давайте обслуживаем интерфейс приложения из HTML-файла с помощью FastAPI.
В рабочей среде мы рекомендуем использовать DBOS в первую очередь для серверной части, а внешний интерфейс развернут в другом месте.

@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 секунды времени настенных часов.
Этот пробел объясняет, почему DBOS в 50 раз дешевле, чем другие бессерверные платформы для рабочих нагрузок ИИ, поскольку DBOS выставляет счет только за фактически потребляемое вами процессорное время, в то время как другие платформы выставляют счет за общую продолжительность запроса.

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.
Получите ключ API здесь и настройте способ оплаты для своей учетной записи здесь.
Этот бот использует gpt-3.5-turbo для генерации текста.
Убедитесь, что у вас есть кредиты (~$1), чтобы использовать его.

Установите ключ API как переменную среды:

export OPENAI_API_KEY=

Развертывание в облаке

Чтобы развернуть это приложение в DBOS Cloud, сначала установите интерфейс командной строки DBOS Cloud (требуется Node):

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

Затем клонируйте репозиторий dbos-demo-apps и разверните:

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

Эта команда выводит URL-адрес — посетите его, чтобы увидеть своего чат-бота!
Вы также можете посетить консоль DBOS Cloud, чтобы просмотреть статус и журналы вашего приложения.

Запуск локально

Сначала клонируйте и войдите в репозиторий 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

DBOS требует базы данных Postgres.
Если у вас его еще нет, вы можете запустить его с помощью Docker:

export PGPASSWORD=dbos
python3 start_postgres_docker.py

Затем запустите приложение в виртуальной среде:

pip install -r requirements.txt
dbos migrate
dbos start

Посетите http://localhost:8000, чтобы увидеть своего чат-бота!

Следующие шаги

Узнайте, как DBOS может сделать ваши приложения более масштабируемыми и устойчивыми:

• Используйте устойчивое выполнение для создания устойчивых к сбоям рабочих процессов.
• Используйте очереди для корректного управления ограничениями скорости API.
• Используйте запланированные рабочие процессы для запуска функций через определенные промежутки времени.
• Хотите узнать, что можно создать с помощью DBOS? Изучите другие примеры приложений.