؟ هل ترغب في إنشاء برنامج chatbot في أقل من 100 سطر من لغة Python واستضافته بدون خادم على السحابة باستخدام أمر واحد؟

يوضح لك هذا الدليل كيفية إنشاء روبوت دردشة تفاعلي مدعوم من LLM باستخدام DBOS وLangChain ونشره بدون خادم على DBOS Cloud.

يمكنك مشاهدة برنامج الدردشة الحية هنا.

بالإضافة إلى الدردشة، يعرض هذا الروبوت مقدار وقت وحدة المعالجة المركزية ووقت ساعة الحائط التي تستهلكها طلباتك.
أثناء الدردشة، ستلاحظ بسرعة أنه على الرغم من أن طلباتك قد تستغرق وقتًا طويلاً، إلا أنها تستهلك القليل جدًا من وحدة المعالجة المركزية.
وذلك لأنهم يقضون معظم وقتهم خاملين في انتظار رد 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.
سنستخدم Langchain للرد على كل رسالة دردشة باستخدام نموذج OpenAI's 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

نحتاج أيضًا إلى نقطة نهاية للسجل تقوم باسترداد جميع الدردشات السابقة من قاعدة البيانات لمستخدم معين.

يتم استدعاء هذه الوظيفة عندما نبدأ تشغيل برنامج chatbot حتى يتمكن من عرض سجل الدردشة الخاص بك.

@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 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 — قم بزيارته لرؤية برنامج الدردشة الآلي الخاص بك!
يمكنك أيضًا زيارة 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؟ استكشف أمثلة التطبيقات الأخرى.