？想要用

本指南向您展示如何使用 DBOS 和 LangChain 构建一个由 LLM 驱动的交互式聊天机器人，并将其无服务器部署到 DBOS Cloud。

您可以在这里看到聊天机器人。

除了聊天之外，该机器人还会显示您的请求所消耗的 CPU 时间和挂钟时间。
当您聊天时，您很快就会注意到，虽然您的请求可能需要很长时间，但它们消耗的 CPU 很少。
那是因为他们大部分时间都在闲置等待 LLM 的回复。
这一差距解释了为什么 DBOS 的 AI 工作负载成本效益比其他无服务器平台高 50 倍，因为 DBOS 仅按您实际消耗的 CPU 时间计费，而其他平台则按总请求持续时间计费。

所有源代码均可在 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。
我们将使用 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以获取每条新消息的函数。
它使用您的用户名作为线程 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.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]

此外，让我们使用 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 中显示这些指标。
此代码在后台线程中每秒运行一次。

我们使用 psutil 跟踪该进程的 CPU 消耗。
我们通过记录每个请求的端到端持续时间来跟踪挂钟时间。

当您第一次启动应用程序时，您会注意到 HTTP 服务器有一些少量的剩余 CPU 消耗。
然而，当您开始聊天时，您很快就会发现每次聊天仅消耗约 10 毫秒的 CPU 时间，但消耗 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 Cloud，请首先安装 DBOS Cloud CLI（需要 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云控制台查看应用程序的状态和日志。

本地运行

首先，克隆并进入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 构建什么？探索其他示例应用程序。