Chainlit快速实现AI对话应用将聊天数据的持久化到postgres关系数据库中

概述

默认情况下，Chainlit 应用不会保留其生成的聊天和元素。即网页一刷新，所有的聊天记录，页面上的所有聊天记录都会消失。但是，存储和利用这些数据的能力可能是您的项目或组织的重要组成部分。

之前写过一篇文章《Chainlit快速实现AI对话应用并将聊天数据的持久化到sqllite本地数据库中》，这个技术方案的优点是，不需要自己在安装数据库，创建表结构等操作，缺点是，只适合用户量比较少的情况。使用postgres数据库可以解决中等规模的用户访问聊天记录访问问题。

教程

1. 安装chainlit依赖

pip install chainlit psycopg2 aiohttp aiofiles sqlalchemy

2. 配置环境变量

在项目根目录下，创建.env文件，内容如下：

OPENAI_BASE_URL="https://dashscope.aliyuncs.com/compatible-mode/v1"
OPENAI_API_KEY="your api_key"
  • 由于国内无法访问open ai的chatgpt，所以需要配置 OPENAI_BASE_URL的代理地址，如果使用国内的LLM大模型接口，可以使用兼容open ai的接口地址

安装postgres 数据库

可以参考这篇文章 《windows 安装PostgresSQL数据库简单教程》，安装postgres数据库后，使用navicat等数据管理工具，创建一个数据库，例如，名为chain_lit的数据库，然后导入一下创建表结构的sql命令：

CREATE TABLE users ("id" UUID PRIMARY KEY,"identifier" TEXT NOT NULL UNIQUE,"metadata" JSONB NOT NULL,"createdAt" TEXT
);CREATE TABLE IF NOT EXISTS threads ("id" UUID PRIMARY KEY,"createdAt" TEXT,"name" TEXT,"userId" UUID,"userIdentifier" TEXT,"tags" TEXT[],"metadata" JSONB,FOREIGN KEY ("userId") REFERENCES users("id") ON DELETE CASCADE
);CREATE TABLE IF NOT EXISTS steps ("id" UUID PRIMARY KEY,"name" TEXT NOT NULL,"type" TEXT NOT NULL,"threadId" UUID NOT NULL,"parentId" UUID,"disableFeedback" BOOLEAN NOT NULL,"streaming" BOOLEAN NOT NULL,"waitForAnswer" BOOLEAN,"isError" BOOLEAN,"metadata" JSONB,"tags" TEXT[],"input" TEXT,"output" TEXT,"createdAt" TEXT,"start" TEXT,"end" TEXT,"generation" JSONB,"showInput" TEXT,"language" TEXT,"indent" INT
);CREATE TABLE IF NOT EXISTS elements ("id" UUID PRIMARY KEY,"threadId" UUID,"type" TEXT,"url" TEXT,"chainlitKey" TEXT,"name" TEXT NOT NULL,"display" TEXT,"objectKey" TEXT,"size" TEXT,"page" INT,"language" TEXT,"forId" UUID,"mime" TEXT
);CREATE TABLE IF NOT EXISTS feedbacks ("id" UUID PRIMARY KEY,"forId" UUID NOT NULL,"threadId" UUID NOT NULL,"value" INT NOT NULL,"comment" TEXT
);

3. 创建代码

在项目艮目录下，创建postgres_client.py文件,代码如下：

from typing import TYPE_CHECKING, Dict, Union, Anyimport psycopg2  # type: ignore
from chainlit.data import BaseStorageClient
from chainlit.logger import logger
from psycopg2.extras import RealDictCursorif TYPE_CHECKING:from psycopg2.extensions import connection, cursorclass PostgresStorageClient(BaseStorageClient):"""Class to enable storage in a PostgreSQL database.parms:host: Hostname or IP address of the PostgreSQL server.dbname: Name of the database to connect to.user: User name used to authenticate.password: Password used to authenticate.port: Port number to connect to (default: 5432)."""def __init__(self, host: str, dbname: str, user: str, password: str, port: int = 5432):try:self.conn: connection = psycopg2.connect(host=host,dbname=dbname,user=user,password=password,port=port)self.cursor: cursor = self.conn.cursor(cursor_factory=RealDictCursor)logger.info("PostgresStorageClient initialized")except Exception as e:logger.warn(f"PostgresStorageClient initialization error: {e}")async def upload_file(self, object_key: str, data: Union[bytes, str], mime: str = 'application/octet-stream',overwrite: bool = True) -> Dict[str, Any]:try:# Assuming the table is called files and has columns id, object_key, data, and mimequery = """INSERT INTO files (object_key, data, mime)VALUES (%s, %s, %s)ON CONFLICT (object_key)DO UPDATE SET data = EXCLUDED.data, mime = EXCLUDED.mime;"""self.cursor.execute(query, (object_key, psycopg2.Binary(data) if isinstance(data, bytes) else data, mime))self.conn.commit()url = f"http://example.com/download/{object_key}"return {"object_key": object_key, "url": url}except Exception as e:logger.warn(f"PostgresStorageClient, upload_file error: {e}")return {}

在项目艮目录下，创建postgres_data.py文件,代码如下：

import json
import ssl
import uuid
from dataclasses import asdict
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
from literalai.helper import utc_nowimport aiofiles
import aiohttp
from chainlit.context import context
from chainlit.data import BaseDataLayer, BaseStorageClient, queue_until_user_message
from chainlit.logger import logger
from chainlit.step import StepDict
from chainlit.types import (Feedback,FeedbackDict,PageInfo,PaginatedResponse,Pagination,ThreadDict,ThreadFilter,
)
from chainlit.user import PersistedUser, User
from sqlalchemy import text
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmakerif TYPE_CHECKING:from chainlit.element import Element, ElementDictfrom chainlit.step import StepDictclass PostgresDataLayer(BaseDataLayer):def __init__(self,conninfo: str,ssl_require: bool = False,storage_provider: Optional[BaseStorageClient] = None,user_thread_limit: Optional[int] = 1000,show_logger: Optional[bool] = False,):self._conninfo = conninfoself.user_thread_limit = user_thread_limitself.show_logger = show_loggerssl_args = {}if ssl_require:# Create an SSL context to require an SSL connectionssl_context = ssl.create_default_context()ssl_context.check_hostname = Falsessl_context.verify_mode = ssl.CERT_NONEssl_args["ssl"] = ssl_contextself.engine: AsyncEngine = create_async_engine(self._conninfo, connect_args=ssl_args)self.async_session = sessionmaker(bind=self.engine, expire_on_commit=False, class_=AsyncSession)  # type: ignoreif storage_provider:self.storage_provider: Optional[BaseStorageClient] = storage_providerif self.show_logger:logger.info("SQLAlchemyDataLayer storage client initialized")else:self.storage_provider = Nonelogger.warn("SQLAlchemyDataLayer storage client is not initialized and elements will not be persisted!")async def build_debug_url(self) -> str:return ""###### SQL Helpers ######async def execute_sql(self, query: str, parameters: dict) -> Union[List[Dict[str, Any]], int, None]:parameterized_query = text(query)async with self.async_session() as session:try:await session.begin()result = await session.execute(parameterized_query, parameters)await session.commit()if result.returns_rows:json_result = [dict(row._mapping) for row in result.fetchall()]clean_json_result = self.clean_result(json_result)return clean_json_resultelse:return result.rowcountexcept SQLAlchemyError as e:await session.rollback()logger.warn(f"An error occurred: {e}")return Noneexcept Exception as e:await session.rollback()logger.warn(f"An unexpected error occurred: {e}")return Noneasync def get_current_timestamp(self) -> str:return utc_now()def clean_result(self, obj):"""Recursively change UUID -> str and serialize dictionaries"""if isinstance(obj, dict):return {k: self.clean_result(v) for k, v in obj.items()}elif isinstance(obj, list):return [self.clean_result(item) for item in obj]elif isinstance(obj, uuid.UUID):return str(obj)return obj###### User ######async def get_user(self, identifier: str) -> Optional[PersistedUser]:if self.show_logger:logger.info(f"SQLAlchemy: get_user, identifier={identifier}")query = "SELECT * FROM users WHERE identifier = :identifier"parameters = {"identifier": identifier}result = await self.execute_sql(query=query, parameters=parameters)if result and isinstance(result, list):user_data = result[0]return PersistedUser(**user_data)return Noneasync def create_user(self, user: User) -> Optional[PersistedUser]:if self.show_logger:logger.info(f"SQLAlchemy: create_user, user_identifier={user.identifier}")existing_user: Optional["PersistedUser"] = await self.get_user(user.identifier)user_dict: Dict[str, Any] = {"identifier": str(user.identifier),"metadata": json.dumps(user.metadata) or {},}if not existing_user:  # create the userif self.show_logger:logger.info("SQLAlchemy: create_user, creating the user")user_dict["id"] = str(uuid.uuid4())user_dict["createdAt"] = await self.get_current_timestamp()query = """INSERT INTO users ("id", "identifier", "createdAt", "metadata") VALUES (:id, :identifier, :createdAt, :metadata)"""await self.execute_sql(query=query, parameters=user_dict)else:  # update the userif self.show_logger:logger.info("SQLAlchemy: update user metadata")query = """UPDATE users SET "metadata" = :metadata WHERE "identifier" = :identifier"""await self.execute_sql(query=query, parameters=user_dict)  # We want to update the metadatareturn await self.get_user(user.identifier)###### Threads ######async def get_thread_author(self, thread_id: str) -> str:if self.show_logger:logger.info(f"SQLAlchemy: get_thread_author, thread_id={thread_id}")query = """SELECT "userIdentifier" FROM threads WHERE "id" = :id"""parameters = {"id": thread_id}result = await self.execute_sql(query=query, parameters=parameters)if isinstance(result, list) and result:author_identifier = result[0].get("userIdentifier")if author_identifier is not None:return author_identifierraise ValueError(f"Author not found for thread_id {thread_id}")async def get_thread(self, thread_id: str) -> Optional[ThreadDict]:if self.show_logger:logger.info(f"SQLAlchemy: get_thread, thread_id={thread_id}")user_threads: Optional[List[ThreadDict]] = await self.get_all_user_threads(thread_id=thread_id)if user_threads:return user_threads[0]else:return Noneasync def update_thread(self,thread_id: str,name: Optional[str] = None,user_id: Optional[str] = None,metadata: Optional[Dict] = None,tags: Optional[List[str]] = None,):if self.show_logger:logger.info(f"SQLAlchemy: update_thread, thread_id={thread_id}")if context.session.user is not None:user_identifier = context.session.user.identifierelse:raise ValueError("User not found in session context")data = {"id": thread_id,"createdAt": (await self.get_current_timestamp() if metadata is None else None),"name": (nameif name is not Noneelse (metadata.get("name") if metadata and "name" in metadata else None)),"userId": user_id,"userIdentifier": user_identifier,"tags": tags,"metadata": json.dumps(metadata) if metadata else None,}parameters = {key: value for key, value in data.items() if value is not None}  # Remove keys with None valuescolumns = ", ".join(f'"{key}"' for key in parameters.keys())values = ", ".join(f":{key}" for key in parameters.keys())updates = ", ".join(f'"{key}" = EXCLUDED."{key}"' for key in parameters.keys() if key != "id")query = f"""INSERT INTO threads ({columns})VALUES ({values})ON CONFLICT ("id") DO UPDATESET {updates};"""await self.execute_sql(query=query, parameters=parameters)async def delete_thread(self, thread_id: str):if self.show_logger:logger.info(f"SQLAlchemy: delete_thread, thread_id={thread_id}")# Delete feedbacks/elements/steps/threadfeedbacks_query = """DELETE FROM feedbacks WHERE "forId" IN (SELECT "id" FROM steps WHERE "threadId" = :id)"""elements_query = """DELETE FROM elements WHERE "threadId" = :id"""steps_query = """DELETE FROM steps WHERE "threadId" = :id"""thread_query = """DELETE FROM threads WHERE "id" = :id"""parameters = {"id": thread_id}await self.execute_sql(query=feedbacks_query, parameters=parameters)await self.execute_sql(query=elements_query, parameters=parameters)await self.execute_sql(query=steps_query, parameters=parameters)await self.execute_sql(query=thread_query, parameters=parameters)async def list_threads(self, pagination: Pagination, filters: ThreadFilter) -> PaginatedResponse:if self.show_logger:logger.info(f"SQLAlchemy: list_threads, pagination={pagination}, filters={filters}")if not filters.userId:raise ValueError("userId is required")all_user_threads: List[ThreadDict] = (await self.get_all_user_threads(user_id=filters.userId) or [])search_keyword = filters.search.lower() if filters.search else Nonefeedback_value = int(filters.feedback) if filters.feedback else Nonefiltered_threads = []for thread in all_user_threads:keyword_match = Truefeedback_match = Trueif search_keyword or feedback_value is not None:if search_keyword:keyword_match = any(search_keyword in step["output"].lower()for step in thread["steps"]if "output" in step)if feedback_value is not None:feedback_match = False  # Assume no match until foundfor step in thread["steps"]:feedback = step.get("feedback")if feedback and feedback.get("value") == feedback_value:feedback_match = Truebreakif keyword_match and feedback_match:filtered_threads.append(thread)start = 0if pagination.cursor:for i, thread in enumerate(filtered_threads):if (thread["id"] == pagination.cursor):  # Find the start index using pagination.cursorstart = i + 1breakend = start + pagination.firstpaginated_threads = filtered_threads[start:end] or []has_next_page = len(filtered_threads) > endstart_cursor = paginated_threads[0]["id"] if paginated_threads else Noneend_cursor = paginated_threads[-1]["id"] if paginated_threads else Nonereturn PaginatedResponse(pageInfo=PageInfo(hasNextPage=has_next_page,startCursor=start_cursor,endCursor=end_cursor,),data=paginated_threads,)###### Steps ######@queue_until_user_message()async def create_step(self, step_dict: "StepDict"):if self.show_logger:logger.info(f"SQLAlchemy: create_step, step_id={step_dict.get('id')}")if not getattr(context.session.user, "id", None):raise ValueError("No authenticated user in context")step_dict["showInput"] = (str(step_dict.get("showInput", "")).lower()if "showInput" in step_dictelse None)parameters = {key: valuefor key, value in step_dict.items()if value is not None and not (isinstance(value, dict) and not value)}parameters["metadata"] = json.dumps(step_dict.get("metadata", {}))parameters["generation"] = json.dumps(step_dict.get("generation", {}))columns = ", ".join(f'"{key}"' for key in parameters.keys())values = ", ".join(f":{key}" for key in parameters.keys())updates = ", ".join(f'"{key}" = :{key}' for key in parameters.keys() if key != "id")query = f"""INSERT INTO steps ({columns})VALUES ({values})ON CONFLICT (id) DO UPDATESET {updates};"""await self.execute_sql(query=query, parameters=parameters)@queue_until_user_message()async def update_step(self, step_dict: "StepDict"):if self.show_logger:logger.info(f"SQLAlchemy: update_step, step_id={step_dict.get('id')}")await self.create_step(step_dict)@queue_until_user_message()async def delete_step(self, step_id: str):if self.show_logger:logger.info(f"SQLAlchemy: delete_step, step_id={step_id}")# Delete feedbacks/elements/stepsfeedbacks_query = """DELETE FROM feedbacks WHERE "forId" = :id"""elements_query = """DELETE FROM elements WHERE "forId" = :id"""steps_query = """DELETE FROM steps WHERE "id" = :id"""parameters = {"id": step_id}await self.execute_sql(query=feedbacks_query, parameters=parameters)await self.execute_sql(query=elements_query, parameters=parameters)await self.execute_sql(query=steps_query, parameters=parameters)###### Feedback ######async def upsert_feedback(self, feedback: Feedback) -> str:if self.show_logger:logger.info(f"SQLAlchemy: upsert_feedback, feedback_id={feedback.id}")feedback.id = feedback.id or str(uuid.uuid4())feedback_dict = asdict(feedback)parameters = {key: value for key, value in feedback_dict.items() if value is not None}columns = ", ".join(f'"{key}"' for key in parameters.keys())values = ", ".join(f":{key}" for key in parameters.keys())updates = ", ".join(f'"{key}" = :{key}' for key in parameters.keys() if key != "id")query = f"""INSERT INTO feedbacks ({columns})VALUES ({values})ON CONFLICT (id) DO UPDATESET {updates};"""await self.execute_sql(query=query, parameters=parameters)return feedback.idasync def delete_feedback(self, feedback_id: str) -> bool:if self.show_logger:logger.info(f"SQLAlchemy: delete_feedback, feedback_id={feedback_id}")query = """DELETE FROM feedbacks WHERE "id" = :feedback_id"""parameters = {"feedback_id": feedback_id}await self.execute_sql(query=query, parameters=parameters)return True###### Elements ######@queue_until_user_message()async def create_element(self, element: "Element"):if self.show_logger:logger.info(f"SQLAlchemy: create_element, element_id = {element.id}")if not getattr(context.session.user, "id", None):raise ValueError("No authenticated user in context")if not self.storage_provider:logger.warn(f"SQLAlchemy: create_element error. No blob_storage_client is configured!")returnif not element.for_id:returncontent: Optional[Union[bytes, str]] = Noneif element.path:async with aiofiles.open(element.path, "rb") as f:content = await f.read()elif element.url:async with aiohttp.ClientSession() as session:async with session.get(element.url) as response:if response.status == 200:content = await response.read()else:content = Noneelif element.content:content = element.contentelse:raise ValueError("Element url, path or content must be provided")if content is None:raise ValueError("Content is None, cannot upload file")context_user = context.session.useruser_folder = getattr(context_user, "id", "unknown")file_object_key = f"{user_folder}/{element.id}" + (f"/{element.name}" if element.name else "")if not element.mime:element.mime = "application/octet-stream"uploaded_file = await self.storage_provider.upload_file(object_key=file_object_key, data=content, mime=element.mime, overwrite=True)if not uploaded_file:raise ValueError("SQLAlchemy Error: create_element, Failed to persist data in storage_provider")element_dict: ElementDict = element.to_dict()element_dict["url"] = uploaded_file.get("url")element_dict["objectKey"] = uploaded_file.get("object_key")element_dict_cleaned = {k: v for k, v in element_dict.items() if v is not None}columns = ", ".join(f'"{column}"' for column in element_dict_cleaned.keys())placeholders = ", ".join(f":{column}" for column in element_dict_cleaned.keys())query = f"INSERT INTO elements ({columns}) VALUES ({placeholders})"await self.execute_sql(query=query, parameters=element_dict_cleaned)@queue_until_user_message()async def delete_element(self, element_id: str, thread_id: Optional[str] = None):if self.show_logger:logger.info(f"SQLAlchemy: delete_element, element_id={element_id}")query = """DELETE FROM elements WHERE "id" = :id"""parameters = {"id": element_id}await self.execute_sql(query=query, parameters=parameters)async def delete_user_session(self, id: str) -> bool:return False  # Not sure why documentation wants thisasync def get_all_user_threads(self, user_id: Optional[str] = None, thread_id: Optional[str] = None) -> Optional[List[ThreadDict]]:"""Fetch all user threads up to self.user_thread_limit, or one thread by id if thread_id is provided."""if self.show_logger:logger.info(f"SQLAlchemy: get_all_user_threads")user_threads_query = """SELECT"id" AS thread_id,"createdAt" AS thread_createdat,"name" AS thread_name,"userId" AS user_id,"userIdentifier" AS user_identifier,"tags" AS thread_tags,"metadata" AS thread_metadataFROM threadsWHERE "userId" = :user_id OR "id" = :thread_idORDER BY "createdAt" DESCLIMIT :limit"""user_threads = await self.execute_sql(query=user_threads_query,parameters={"user_id": user_id,"limit": self.user_thread_limit,"thread_id": thread_id,},)if not isinstance(user_threads, list):return Noneif not user_threads:return []else:thread_ids = ("('"+ "','".join(map(str, [thread["thread_id"] for thread in user_threads]))+ "')")steps_feedbacks_query = f"""SELECTs."id" AS step_id,s."name" AS step_name,s."type" AS step_type,s."threadId" AS step_threadid,s."parentId" AS step_parentid,s."streaming" AS step_streaming,s."waitForAnswer" AS step_waitforanswer,s."isError" AS step_iserror,s."metadata" AS step_metadata,s."tags" AS step_tags,s."input" AS step_input,s."output" AS step_output,s."createdAt" AS step_createdat,s."start" AS step_start,s."end" AS step_end,s."generation" AS step_generation,s."showInput" AS step_showinput,s."language" AS step_language,s."indent" AS step_indent,f."value" AS feedback_value,f."comment" AS feedback_commentFROM steps s LEFT JOIN feedbacks f ON s."id" = f."forId"WHERE s."threadId" IN {thread_ids}ORDER BY s."createdAt" ASC"""steps_feedbacks = await self.execute_sql(query=steps_feedbacks_query, parameters={})elements_query = f"""SELECTe."id" AS element_id,e."threadId" as element_threadid,e."type" AS element_type,e."chainlitKey" AS element_chainlitkey,e."url" AS element_url,e."objectKey" as element_objectkey,e."name" AS element_name,e."display" AS element_display,e."size" AS element_size,e."language" AS element_language,e."page" AS element_page,e."forId" AS element_forid,e."mime" AS element_mimeFROM elements eWHERE e."threadId" IN {thread_ids}"""elements = await self.execute_sql(query=elements_query, parameters={})thread_dicts = {}for thread in user_threads:thread_id = thread["thread_id"]if thread_id is not None:thread_dicts[thread_id] = ThreadDict(id=thread_id,createdAt=thread["thread_createdat"],name=thread["thread_name"],userId=thread["user_id"],userIdentifier=thread["user_identifier"],tags=thread["thread_tags"],metadata=thread["thread_metadata"],steps=[],elements=[],)# Process steps_feedbacks to populate the steps in the corresponding ThreadDictif isinstance(steps_feedbacks, list):for step_feedback in steps_feedbacks:thread_id = step_feedback["step_threadid"]if thread_id is not None:feedback = Noneif step_feedback["feedback_value"] is not None:feedback = FeedbackDict(forId=step_feedback["step_id"],id=step_feedback.get("feedback_id"),value=step_feedback["feedback_value"],comment=step_feedback.get("feedback_comment"),)step_dict = StepDict(id=step_feedback["step_id"],name=step_feedback["step_name"],type=step_feedback["step_type"],threadId=thread_id,parentId=step_feedback.get("step_parentid"),streaming=step_feedback.get("step_streaming", False),waitForAnswer=step_feedback.get("step_waitforanswer"),isError=step_feedback.get("step_iserror"),metadata=(step_feedback["step_metadata"]if step_feedback.get("step_metadata") is not Noneelse {}),tags=step_feedback.get("step_tags"),input=(step_feedback.get("step_input", "")if step_feedback["step_showinput"] == "true"else None),output=step_feedback.get("step_output", ""),createdAt=step_feedback.get("step_createdat"),start=step_feedback.get("step_start"),end=step_feedback.get("step_end"),generation=step_feedback.get("step_generation"),showInput=step_feedback.get("step_showinput"),language=step_feedback.get("step_language"),indent=step_feedback.get("step_indent"),feedback=feedback,)# Append the step to the steps list of the corresponding ThreadDictthread_dicts[thread_id]["steps"].append(step_dict)if isinstance(elements, list):for element in elements:thread_id = element["element_threadid"]if thread_id is not None:element_dict = ElementDict(id=element["element_id"],threadId=thread_id,type=element["element_type"],chainlitKey=element.get("element_chainlitkey"),url=element.get("element_url"),objectKey=element.get("element_objectkey"),name=element["element_name"],display=element["element_display"],size=element.get("element_size"),language=element.get("element_language"),autoPlay=element.get("element_autoPlay"),playerConfig=element.get("element_playerconfig"),page=element.get("element_page"),forId=element.get("element_forid"),mime=element.get("element_mime"),)thread_dicts[thread_id]["elements"].append(element_dict)  # type: ignorereturn list(thread_dicts.values())

在项目根目录下，创建一个app.py的文件，代码如下：

from typing import List, Optionalimport chainlit as cl
import chainlit.data as cl_data
from chainlit.data.sql_alchemy import SQLAlchemyDataLayer
from openai import AsyncOpenAIfrom postgres_client import PostgresStorageClientclient = AsyncOpenAI()thread_history = []  # type: List[cl_data.ThreadDict]
deleted_thread_ids = []  # type: List[str]storage_client = PostgresStorageClient(host="postgres数据库IP", dbname="postgres数据库名称", port=5432, user="postgres数据库账户",password="postgres数据库密码")cl_data._data_layer = SQLAlchemyDataLayer(conninfo="postgresql+asyncpg://username:password@ip:port/dbname",storage_provider=storage_client)@cl.on_chat_start
async def main():content = "你好，我是泰山AI智能客服，有什么可以帮助您吗？"await cl.Message(content).send()@cl.on_message
async def handle_message():# Wait for queue to be flushedawait cl.sleep(1)msg = cl.Message(content="")await msg.send()stream = await client.chat.completions.create(model="qwen-turbo", messages=cl.chat_context.to_openai(), stream=True)async for part in stream:if token := part.choices[0].delta.content or "":await msg.stream_token(token)await msg.update()@cl.password_auth_callback
def auth_callback(username: str, password: str) -> Optional[cl.User]:if (username, password) == ("admin", "admin"):return cl.User(identifier="admin")else:return None@cl.on_chat_resume
async def on_chat_resume():pass
  • 将代码中关于postgres数据库连接信息，修改为自己的即可。

4. 执行命令创建 AUTH_SECRET 鉴权

chainlit create-secret

在这里插入图片描述
复制最后一行代码到.env环境配置文件中

CHAINLIT_AUTH_SECRET="$b?/v0NeJlAU~I5As1WSCa,j8wJ3w%agTyIFlUt4408?mfC*,/wovlfA%3O/751U"
OPENAI_BASE_URL="https://dashscope.aliyuncs.com/compatible-mode/v1"
OPENAI_API_KEY=""

5. 执行服务启动命令

chainlit run app.py -w

6. 启动后效果展示

在这里插入图片描述

  • 现在聊天记录都被保存在服务的sqllite本地数据库中了，只要不重启服务，聊天记录就不会丢失了!

相关文章推荐

《使用 Xinference 部署本地模型》
《Fastgpt接入Whisper本地模型实现语音输入》
《Fastgpt部署和接入使用重排模型bge-reranker》
《Fastgpt部署接入 M3E和chatglm2-m3e文本向量模型》
《Fastgpt 无法启动或启动后无法正常使用的讨论（启动失败、用户未注册等问题这里）》
《vllm推理服务兼容openai服务API》
《vLLM模型推理引擎参数大全》
《解决vllm推理框架内在开启多显卡时报错问题》
《Ollama 在本地快速部署大型语言模型，可进行定制并创建属于您自己的模型》

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/396948.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

实现高亮的全文分页检索

实现高亮的全文分页检索

文章目录 🌞 Sun Frame：SpringBoot 的轻量级开发框架（个人开源项目推荐）🌟 亮点功能📦 spring cloud模块概览常用工具 🔗 更多信息1.sun-club-infra 模块SubjectEsServiceImpl.java1.querySubje…
阅读更多...
大数据-74 Kafka 高级特性 稳定性 - 控制器、可靠性 副本复制、失效副本、副本滞后 多图一篇详解

大数据-74 Kafka 高级特性 稳定性 - 控制器、可靠性 副本复制、失效副本、副本滞后 多图一篇详解

点一下关注吧！！！非常感谢！！持续更新！！！ 目前已经更新到了： Hadoop（已更完）HDFS（已更完）MapReduce（已更完&am…
阅读更多...
【Linux】Linux环境基础开发工具使用之软件包管理（yum）与 Linux编辑器（vim）

【Linux】Linux环境基础开发工具使用之软件包管理（yum）与 Linux编辑器（vim）

目录 一、Linux 软件包管理器 yum1.1 什么是软件包1.2 关于 rzsz1.3 查看软件包1.4 如何安装软件1.5 如何卸载软件1.6 如何更新软件包1.7 yum源更新 二、 Linux编辑器-vim使用2.1 vim的基本概念2.2 vim的基本操作2.3 vim命令模式命令集2.3.1 从命令模式切换为插入模式2.3.2 从插…
阅读更多...
03创建型设计模式——抽象工厂模式

03创建型设计模式——抽象工厂模式

一、抽象工厂模式简介 抽象工厂模式是所有形态的工厂模式中最为抽象和具有一般性的。抽象工厂模式可以向客户端提供一个接口，使得客户端在不必指定产品的具体类型的情况下，能够创建多个产品族的产品对象。例如现实生活中，水果的种类繁多&…
阅读更多...
Vue3+Ts项目中经常遇到导入组件，vscode报无法找到模块xxx，xxx隐式拥有 “any“ 类型解决办法~

Vue3+Ts项目中经常遇到导入组件，vscode报无法找到模块xxx，xxx隐式拥有 “any“ 类型解决办法~

1、报错截图： 2、解决办法：在确保路径正确的情况下，你会在 src 目录下找到一个名为 env.d.ts 的文件（或者类似的名称）。在这个文件中，你可以声明 .vue 文件的模块类型。例如：(这告诉 TypeScript…
阅读更多...
系统内存管理：虚拟内存、内存分段与分页、页表缓存TLB以及Linux内存管理

系统内存管理：虚拟内存、内存分段与分页、页表缓存TLB以及Linux内存管理

虚拟内存 虚拟内存是一种操作系统提供的机制，用于将每个进程分配的独立的虚拟地址空间映射到实际的物理内存地址空间上。通过使用虚拟内存，操作系统可以有效地解决多个应用程序直接操作物理内存可能引发的冲突问题。 在使用虚拟内存的情况下&#xff0…
阅读更多...
IDEA打开持久层的代码很卡，关掉mybatis-plus的插件

IDEA打开持久层的代码很卡，关掉mybatis-plus的插件

不知道大家有没有遇到过打开 mapper 层的页面，然后要切换另外 java 文件的时候很卡，我遇到过卡了好几分钟的，那种继承了 mybatis-plus 的 mapper java文件或者 xml 文件都会，我后来把 mybatis 的插件关掉了，就不会了
阅读更多...
LVS的12种调度算法详解

LVS的12种调度算法详解

1.lvs调度算法类型 1.1静态方法 仅根据算法本身进行调度，不考虑RS的负载情况 1.2动态方法 主要根据每RS当前的负载状态及调度算法进行调度Overheadvalue较小的RS将被调度 1.1lvs静态调度算法 1.1.1RR（轮询算法）： roundrobin 轮…
阅读更多...
Haproxy状态页

Haproxy状态页

HAProxy 的状态页是一个非常有用的监控和诊断工具，它提供了关于 HAProxy 服务器运行状态的详细信息，通过web界面，显示当前HAProxy的运行状态。 状态页通常包含以下关键信息： 前端（Frontend）和后端&#x…
阅读更多...
从TiDB迁移到OceanBase的实践分享

从TiDB迁移到OceanBase的实践分享

本文来自OceanBase热心用户的分享 近期，我们计划将业务数据库从TiDB迁移到OceanBase，但面临的一个主要挑战是如何更平滑的完成这一迁移过程。经过研究，了解到OceanBase提供的OMS数据迁移工具能够支持从TiDB到OceanBase的迁移，并且…
阅读更多...
js构造函数的prototype赋值总结

js构造函数的prototype赋值总结

我们知道通过构造函数的prototype,可以生成让所有实例对象访问的通用属性和方法,下面通过代码来解释这个过程 function Person(name){this.name name; }Person.prototype.sex man我们定义了一个构造函数Person,然后给它的prototype添加了一个sex的属性,下面我们来看看Person…
阅读更多...
OSPF进阶

OSPF进阶

一、LSA详解 Type：LSA的类型（1、2、3、4、5、7类） link-state-ID：链路状态表示符 ADV router：产生该LSA的路由器 age：老化时间 Metric：开销值，一般都为ADV router到达该路由的开…
阅读更多...
人工智能，代跑通，预测模型，模型优化

人工智能，代跑通，预测模型，模型优化

人工智能，代跑通，预测模型，模型优化，增加模块，文章复现，python代做，预测，微调，融合，深度学习，机器学习程序代写，环境调试，…
阅读更多...
STM32-门电路-储存器-寄存器-STM32f1-MCU-GPIO-总线-keil5-点led-寄存器编程

STM32-门电路-储存器-寄存器-STM32f1-MCU-GPIO-总线-keil5-点led-寄存器编程

1、门电路 门电路组成简单加法器： 二进制对电路的影响： 0和1代表无和有； 以下图例，演示与门：左1右1输出1； 电平标准：使用不同的电压表示数字0和1； 高电平：1&#xff1…
阅读更多...
攻防世界-web-ctf-upload

攻防世界-web-ctf-upload

题目场景 查看源码 毫无有效的数据 官方WriteUp 本题需要利用文件上传漏洞点，通过绕过服务器的安全防护，达到getshell的目的 本题的主要考点为利用fastcgi的.user.ini特性进行任意命令执行 这里需要绕过的点如下 检查文件内容是否有php字符串 检查…
阅读更多...
haproxy七层代理

haproxy七层代理

目录 一、haproxy简介 二、haproxy实验 1.环境部署 2.haproxy的基本部署方法及负载均衡的实现 2.1安装软件 2.2haproxy的基本配置 3.haproxy的全局配置参数及日志分离 3.1多线程设定 3.2自定义日志 4.haproxy-proxies中的常用配置参数 4.1设置backup --- sorryserver…
阅读更多...
卷积神经网络（李宏毅老师系列）

卷积神经网络（李宏毅老师系列）

自学参考： 一文搞懂卷积神经网络（CNN）的原理 视频课 课件资料 笔记 一、引入 cnn设计灵感来自于生物学中的视觉系统，旨在模拟人类视觉处理的方式。常用场景：image classification 基本步骤： 把所有图片都先…
阅读更多...
数据结构--第六天

数据结构--第六天

--树 -树的基本概念 树结构通常用来储存逻辑关系为“一对多”的数据，例如： 上图的这些元素具有的就是 "一对多" 的逻辑关系，例如元素A同时和B、C、D有关系，元素D同时和A、H、I、J有关系等。 观察这些元素之间的逻辑关…
阅读更多...
模拟经营之神：《北境之地》安卓手机游戏，免费分享

模拟经营之神：《北境之地》安卓手机游戏，免费分享

《北境之地》（Northgard）是一款以北欧神话为背景的即时战略游戏，由Shiro Games开发。玩家在游戏中扮演维京部落的领袖，目标是探索新大陆、建立据点、管理资源，并在严酷的冬季和敌人的威胁下生存下来 。 游戏特色包括&a…
阅读更多...
gin路由

gin路由

1主文件 package main import ("github.com/gin-gonic/gin""godade/user""net/http" ) func main() {router : gin.Default()router.GET("/", func(c *gin.Context) {c.String(http.StatusOK, "Hello World")})v1 : router…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023

友情链接： 我的编程经验分享 一条长河网 尧图网站开发