? 100줄 미만의 Python으로 챗봇을 구축하고 단일 명령으로 클라우드에서 서버 없이 호스팅하고 싶으신가요?

이 가이드에서는 DBOS 및 LangChain을 사용하여 대화형 LLM 기반 챗봇을 구축하고 이를 DBOS Cloud에 서버리스로 배포하는 방법을 보여줍니다.

여기에서 라이브 챗봇을 보실 수 있습니다.

채팅 외에도 이 봇은 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()

채팅 처리

자, 채팅하자!
먼저 각 채팅 요청을 처리하는 엔드포인트를 작성하겠습니다.

이 끝점은 다음 세 단계로 구성된 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초에 한 번씩 실행됩니다.

psutil을 사용하여 이 프로세스의 CPU 소비를 추적합니다.
각 요청의 전체 기간을 기록하여 벽시계 시간을 추적합니다.

앱을 처음 시작하면 HTTP 서버에서 CPU가 약간 소모되는 것을 볼 수 있습니다.
그러나 채팅을 시작하면 각 채팅이 최대 10ms의 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을 출력합니다. 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로 무엇을 구축할 수 있는지 알고 싶으십니까? 다른 예제 애플리케이션을 살펴보세요.