? 100 行未満の Python でチャットボットを構築し、1 つのコマンドでクラウド上でサーバーレスでホストしたいですか?

このガイドでは、DBOS と LangChain を使用して対話型の LLM を利用したチャットボットを構築し、それをサーバーレスで DBOS クラウドにデプロイする方法を説明します。

ここでチャットボットのライブを見ることができます。

このボットは、チャットに加えて、リクエストによって消費された CPU 時間と実時間の両方を表示します。
チャットすると、リクエストに時間がかかる場合でも、CPU 消費量がほとんどないことがすぐにわかります。
それは、彼らはほとんどの時間をアイドル状態で LLM の応答を待って費やしているためです。
このギャップにより、DBOS が AI ワークロードに対して他のサーバーレス プラットフォームよりも 50 倍コスト効率が高い理由が説明されます。他のプラットフォームが総リクエスト時間に対して課金するのに対し、DBOS は実際に消費した CPU 時間に対してのみ課金されるからです。

すべてのソース コードは GitHub で入手できます。

アプリをインポートして初期化する

インポートと DBOS の初期化から始めましょう。
また、HTTP リクエストを処理するために FastAPI をセットアップします。

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 に保存されるように LangChain を構成し、アプリの再起動後もメッセージ履歴が保持されるようにします。

楽しみのために、チャットボットに海賊のように話すように指示してみましょう。

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

チャットの処理

さあ、チャットしましょう!
まず、各チャット リクエストを処理するエンドポイントを作成します。

このエンドポイントは、3 つのステップからなる 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 にクエリを実行する関数を作成しましょう。
ユーザー名をスレッド ID として使用するため、ユーザーごとに異なるスレッドで会話を行うことができます。

この関数に @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]

さらに、FastAPI を使用して HTML ファイルからアプリのフロントエンドを提供しましょう。
運用環境では、主にバックエンドに DBOS を使用し、フロントエンドは別の場所にデプロイすることをお勧めします。

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

アプリの使用状況の追跡

最後に、リクエストによって消費された CPU 時間と実測時間を追跡するコードを作成して、アプリの UI にそれらのメトリクスを表示できるようにしましょう。
このコードはバックグラウンド スレッドで 1 秒に 1 回実行されます。

psutil を使用して、このプロセスの CPU 消費量を追跡します。
各リクエストのエンドツーエンドの継続時間を記録することで実時間を追跡します。

初めてアプリを起動すると、HTTP サーバーによるわずかな残りの CPU 消費に気づくでしょう。
ただし、チャットを開始すると、各チャットが消費する CPU 時間は最大 10 ミリ秒のみで、実時間では 1 ～ 2 秒であることがすぐにわかります。
このギャップは、DBOS が AI ワークロードに対して他のサーバーレス プラットフォームよりも 50 倍安い理由を説明しています。他のプラットフォームがリクエストの合計時間に対して請求するのに対し、DBOS は実際に消費した CPU 時間に対してのみ請求されるからです。

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 クラウドにデプロイするには、まず DBOS クラウド CLI をインストールします (ノードが必要):

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 を出力します。チャットボットを確認するには URL にアクセスしてください。
DBOS Cloud Console にアクセスして、アプリのステータスとログを確認することもできます。

ローカルで実行する

まず、クローンを作成して 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 で何を構築できるかを知りたいですか?他のサンプル アプリケーションを探索してください。