LLMChat

llmchat？

歡迎來到LLMCHAT存儲庫，這是由Python Fastapi構建的API服務器的全棧實現，以及由Flutter提供動力的美麗前端。該項目旨在與高級ChatGPT和其他LLM型號提供無縫的聊天體驗。 ？提供現代基礎架構，當GPT-4的多模式和插件功能可用時，可以輕鬆擴展。享受您的住宿！

演示

---

享受Flutter提供的美麗的UI和豐富的可定制小部件。

• 它支持mobile環境和PC環境。
• 也支持Markdown ，因此您可以使用它來格式化消息。

---

網絡瀏覽

• 達克達克戈

  您可以使用DuckDuckgo搜索引擎在網絡上找到相關信息。只需激活“瀏覽”切換按鈕！

  觀看完整瀏覽的演示視頻：https：//www.youtube.com/watch?v=mj_cvrwrs08

---

向量嵌入

• 嵌入任何文本

  使用/embed命令，您可以無限期地將文本存儲在您自己的私人矢量數據庫中，並隨時以後查詢。如果使用/share命令，則文本存儲在每個人都可以共享的公共矢量數據庫中。啟用Query切換按鈕或/query命令通過在公共和私人數據庫中搜索文本相似性來幫助AI生成上下文化的答案。這解決了語言模型的最大局限性之一：內存。

• 上傳您的PDF文件

  您可以通過單擊左下方的Embed Document來嵌入PDF文件。在幾秒鐘內，PDF的文本內容將轉換為向量並嵌入到Redis Cache中。

---

• 更改您的聊天模型

  您可以通過下拉菜單更改聊天模型。您可以定義要在app/models/llms.py中使用的LLMModels中使用的任何型號。

  

---

• 更改您的聊天標題

  您可以通過單擊聊天標題來更改聊天標題。這將存儲在更改或刪除之前！

  

---

？本地LLM

對於本地的Llalam LLM，假定它僅在本地環境中起作用，並使用http://localhost:8002/v1/completions端點。它通過連接到http://localhost:8002/health秒鐘來查看是否返回200 OK響應，並且它會自動運行單獨的進程以創建API服務器，從而不斷地檢查Llama API服務器的狀態。

Llama.cpp

Llama.cpp的主要目標是使用GGML 4位量化使用普通的C/C ++實現，而無需依賴性。您必須從HuggingFace下載GGML bin文件，然後將其放入llama_models/ggml文件夾中，並在app/models/llms.py中定義LLMMODEL。很少有示例，因此您可以輕鬆地定義自己的模型。有關更多信息，請參閱llama.cpp存儲庫：https：//github.com/ggerganov/llama.cpp

exllama

獨立的Python/C ++/CUDA實現Llama，可與4位GPTQ權重一起使用，該重量是在現代GPU上快速且記憶效率的。它使用pytorch和sentencepiece來運行模型。假定它僅在當地環境中起作用，至少需要一個NVIDIA CUDA GPU 。您必須從HuggingFace下載Tokenizer，config和gptq文件，然後將其放入llama_models/gptq/YOUR_MODEL_FOLDER文件夾中，並在app/models/llms.py中定義llmmodel。很少有示例，因此您可以輕鬆地定義自己的模型。有關更多詳細信息，請參閱exllama存儲庫：https：//github.com/turboderp/exllama

---

關鍵功能

• FastApi-使用Python構建API的高性能web framework 。
• Flutter Webapp前端具有美麗的UI和豐富的可自定義小部件。
• CHATGPT-與OpenAI API無縫集成，用於文本生成和消息管理。
• Llama-支持Localllm， LlamaCpp和Exllama模型。
• Websocket連接- 與Flutter Frontend WebApp的Real-time ，與Chatgpt的雙向通信以及其他LLM型號。
• 向量店- 使用Redis和Langchain ，存儲和檢索矢量嵌入以進行相似性搜索。它將幫助AI產生更相關的響應。
• 自動摘要- 使用Langchain的摘要鏈，匯總對話並將其存儲在數據庫中。這將有助於節省很多令牌。
• Web瀏覽- 使用Duckduckgo搜索引擎，瀏覽網絡並找到相關信息。
• 並發- 異步編程與async / await並發性和並行性的語法。
• 安全性- 令牌驗證和身份驗證以確保API安全。
• 數據庫- 管理數據庫連接並執行MySQL查詢。使用sqlalchemy.asyncio輕鬆執行創建，讀取，更新和刪除操作
• 緩存- 管理緩存連接並使用Aioredis執行Redis查詢。使用aioredis輕鬆地執行創建，讀取，更新和刪除操作。

入門 /安裝

要在本地計算機上設置該設置，請按照以下簡單步驟操作。在開始之前，請確保您的機器上安裝了docker和docker-compose 。如果您想在沒有Docker的情況下運行服務器，則必須安裝Python 3.11 。即使您需要Docker運行DB服務器。

1。克隆存儲庫

要遞歸克隆用於使用Exllama或llama.cpp型號的子模型，請使用以下命令：

git clone --recurse-submodules https://github.com/c0sogi/llmchat.git

您只想使用核心功能（OpenAI），使用以下命令：

git clone https://github.com/c0sogi/llmchat.git

2。更改為項目目錄

 cd LLMChat

3。創建.env文件

設置一個Env文件，參考.env-sample文件。輸入數據庫信息以創建，OpenAI API密鑰以及其他必要的配置。不需要選擇，只需將它們視為原樣。

4。運行服務器

執行這些。首次啟動服務器可能需要幾分鐘：

docker-compose -f docker-compose-local.yaml up

5。停止服務器

docker-compose -f docker-compose-local.yaml down

6。享受它

現在，您可以通過http://localhost:8000/docs和數據庫訪問服務器db:3306或cache:6379 。您也可以通過http://localhost:8000/chat訪問該應用程序。

• 要在沒有Docker的情況下運行服務器，如果您想在沒有Docker的情況下運行服務器，則必須安裝Python 3.11 。即使您需要Docker運行DB服務器。關閉已經使用docker-compose -f docker-compose-local.yaml down api運行的API服務器。不要忘記在Docker上運行其他DB服務器！然後，運行以下命令：

  python -m main

  現在，您的服務器應在http://localhost:8001上啟動並運行。

執照

該項目是根據MIT許可證獲得許可的，該項目允許免費使用，修改和分發，只要原始版權和許可聲明包含在軟件的任何副本或大部分中。

為什麼要Fastapi？

FastAPI是一個現代的網絡框架，用於使用Python構建API。 ？它具有高性能，易於學習，快速進行編碼和準備生產。 ？ FastAPI的主要特徵之一是它支持並發和async / await語法。 ？這意味著您可以編寫可以同時處理多個任務的代碼，而不會互相阻止，尤其是在處理I/O綁定操作時，例如網絡請求，數據庫查詢，文件操作等。

為什麼要顫抖？

Flutter是由Google開發的開源UI工具包，用於從單個代碼庫中構建用於移動，Web和桌面平台的本機用戶界面。 ？‍使用Dart ，一種現代的面向對象的編程語言，並提供了一組可自定義的小部件，可以適應任何設計。

Websocket連接

您可以使用兩個模塊通過WebSocket連接訪問ChatGPT或LlamaCpp ： app/routers/websocket和app/utils/chat/chat_stream_manager 。這些模塊通過Websocket促進了Flutter客戶端和聊天模型之間的通信。使用Websocket，您可以建立一個實時的雙向通信渠道與LLM互動。

用法

要啟動對話，請使用在數據庫中註冊的有效API密鑰連接到Websocket /ws/chat/{api_key} 。請注意，此API密鑰與OpenAI API密鑰不同，但僅適用於服務器驗證用戶。連接後，您可以發送消息和命令與LLM模型進行交互。 Websocket將實時發送聊天響應。該Websocket連接是通過Flutter應用程序建立的，該應用程序可以使用端點/chat訪問。

websocket.py

websocket.py負責設置Websocket連接和處理用戶身份驗證。它定義了接受Websocket和API密鑰作為參數的Websocket路由/chat/{api_key} 。

當客戶端連接到Websocket時，它首先檢查API鍵以對用戶進行身份驗證。如果API鍵有效，則從stream_manager.py模塊調用begin_chat()函數以啟動對話。

如果有未註冊的API密鑰或出乎意料的錯誤，則將適當的消息發送給客戶端並關閉連接。

 @ router . websocket ( "/chat/{api_key}" )
async def ws_chat ( websocket : WebSocket , api_key : str ):
    ...

stream_manager.py

stream_manager.py負責管理對話和處理用戶消息。它定義了begin_chat()函數，該函數將websocket，一個用戶ID作為參數。

該功能首先從CACHE MARAFER初始化用戶的聊天上下文。然後，它通過Websocket將初始消息歷史記錄發送給客戶端。

對話繼續循環，直到連接關閉。在對話期間，處理用戶的消息，並相應地生成GPT的響應。

 class ChatStreamManager :
    @ classmethod
    async def begin_chat ( cls , websocket : WebSocket , user : Users ) -> None :
    ...

將消息發送到WebSocket

SendToWebsocket類用於將消息和流發送到Websocket。它有兩種方法： message()和stream() 。 message()方法將完整的消息發送到Websocket，而stream()方法將流發送到Websocket。

 class SendToWebsocket :
    @ staticmethod
    async def message (...):
        ...

    @ staticmethod
    async def stream (...):
        ...

處理人工智能響應

MessageHandler類還處理AI響應。 ai()方法將AI響應發送到Websocket。如果啟用了翻譯，則在將其發送給客戶端之前，請使用Google Translate API進行翻譯。

 class MessageHandler :
    ...
    @ staticmethod
    async def ai (...):
        ...

處理自定義命令

用戶消息是使用HandleMessage類處理的。如果消息以/為開頭，例如/YOUR_CALLBACK_NAME 。它被視為命令，並生成適當的命令響應。否則，將處理用戶的消息並將其發送到LLM模型以生成響應。

使用ChatCommands類處理命令。它根據命令執行相應的回調函數。您可以通過在app.utils.chat.chat_commands中的ChatCommands類中添加回調來添加新命令。

矢量嵌入

使用redis存儲對話的矢量嵌入嗎？…可以幫助chatgpt模型嗎？通過幾種方式，例如對對話上下文的高效且快速檢索‍♀️，處理大量數據，並通過向量相似性搜索提供更相關的響應？

一些有趣的例子在實踐中如何起作用：

• 想像一下，用戶正在與Chatgpt聊天有關他們最喜歡的電視節目，並提到一個特定的角色？使用Redis，Chatgpt可以檢索提到該角色的先前對話，並使用該信息來提供有關該角色的更詳細的見解或瑣事？
• 另一種情況可能是討論他們的旅行計劃的用戶✈️與chatgpt。如果他們提到一個特定的城市？或具有里程碑意義的？，chatgpt可以使用矢量相似性搜索檢索以前討論相同位置並基於該上下文的建議或提示的對話。
• 如果用戶提到特定美食？或Dish？，Chatgpt可以檢索以前討論這些主題並基於這種情況的建議或建議的對話？

1。使用/embed命令嵌入文本

當用戶在聊天窗口中輸入命令（例如/embed <text_to_embed>時，調用了VectorStoreManager.create_documents方法。此方法使用OpenAI的text-embedding-ada-002型號將輸入文本轉換為向量，並將其存儲在Redis VectorStore中。

 @ staticmethod
@ command_response . send_message_and_stop
async def embed ( text_to_embed : str , / , buffer : BufferedUserContext ) -> str :
    """Embed the text and save its vectors in the redis vectorstore. n
    /embed <text_to_embed>"""
    ...

2。使用/query命令查詢嵌入式數據

當用戶輸入/query <query>命令時， asimilarity_search函數將用於查找與Redis VectorStore中嵌入式數據相似性最高的三個結果。這些結果暫時存儲在聊天的上下文中，這可以通過參考這些數據來回答查詢。

 @ staticmethod
async def query ( query : str , / , buffer : BufferedUserContext , ** kwargs ) -> Tuple [ str | None , ResponseType ]:
    """Query from redis vectorstore n
    /query <query>"""
    ...

3。自動嵌入上傳的文本文件

在運行begin_chat函數時，如果用戶上傳包含文本的文件（例如，pdf或txt文件），則將自動從文件中提取文本，並將其向量嵌入將其保存到redis中。

 @ classmethod
async def embed_file_to_vectorstore ( cls , file : bytes , filename : str , collection_name : str ) -> str :
    # if user uploads file, embed it
    ...

4。 commands.py功能

在commands.py文件中，有幾個重要組件：

• command_response ：此類用於在命令方法上設置裝飾器，以指定下一個操作。它有助於定義各種響應類型，例如發送消息並停止，發送消息以及繼續，處理用戶輸入，處理AI響應等等。
• command_handler ：此功能負責基於用戶輸入的文本執行命令回調方法。
• arguments_provider ：此功能會根據命令方法的註釋類型自動提供命令方法所需的參數。

自動摘要

通過將任務添加到LLM中，可以節省令牌，以匯總消息。自動摘要任務是提高聊天機器人效率的關鍵功能。讓我們分解此功能的功能：

1. 任務觸發：每當用戶鍵入消息或AI響應消息時，就會激活此功能。此時，生成了自動摘要任務以凝結文本內容。

2. 任務存儲：然後將自動 - 夏線任務存儲在BufferUserChatContext的task_list屬性中。這是管理鏈接到用戶聊天上下文的任務的隊列。

3. 任務收集：在MessageHandler完成用戶問題和答案週期之後，調用了harvest_done_tasks功能。此功能收集了摘要任務的結果，確保沒有遺漏。

4. 摘要申請：收穫過程後，匯總結果替換了我們的聊天機器人從語言學習模型（LLMS）（例如OpenAI和LLAMA_CPP）的答案時，替換了實際消息。通過這樣做，我們可以發送比最初的冗長消息更多的簡潔提示。

5. 用戶體驗：重要的是，從用戶的角度來看，他們只看到原始消息。該消息的摘要版本未顯示給他們，保持透明度並避免潛在的混亂。

6. 同時任務：此自動簽到任務的另一個關鍵功能是它不會阻礙其他任務。換句話說，雖然聊天機器人忙於總結文本，但仍可以執行其他任務，從而提高我們的聊天機器人的整體效率。

默認情況下，總結鏈僅適用於512個令牌或更多的消息。這可以打開/關閉，並且在ChatConfig中設置了閾值。

LLM型號

該存儲庫包含不同的LLM模型，該模型在llms.py中定義。每個LLM模型類都從基類LLMModel繼承。 LLMModels枚舉是這些LLM的集合。

所有操作均異步處理，而不會使主螺紋交織。但是，本地LLM無法同時處理多個請求，因為它們在計算上太昂貴。因此，使用Semaphore將請求數限制為1。

？用法

用戶通過UserChatContext.construct_default使用的默認LLM模型為gpt-3.5-turbo 。您可以更改該功能的默認值。

模型描述

1️⃣Openiaimodel

OpenAIModel通過請求OpenAI服務器的聊天完成來使文本異步生成。它需要一個OpenAI API鍵。

2️⃣llamacppmodel

LlamaCppModel讀取本地存儲的GGML模型。 Llama.cpp GGML模型必須以.bin文件為單位的llama_models/ggml文件夾中。例如，如果您從“ https://huggingface.co/thebloke/robin-7b-v2-ggml”下載了Q4_0量化的模型，則該模型的路徑必須為“ Robin-7b.ggmlv3.q4_0.bin”。

3️⃣exllamamydel

ExllamaModel讀取本地存儲的GPTQ模型。 Exllama GPTQ模型必須以文件夾的形式放置在llama_models/gptq文件夾中。例如，如果您從“ https://huggingface.co/thebloke/orca_mini_7b-gptq/tree/main/main”下載了3個文件：

• ORCA-MINI-7B-GPTQ-4BIT-128G.NO-ACT.ORDER.SAFETENSORS
• tokenizer.model
• config.json

然後，您需要將它們放在文件夾中。模型的路徑必須是文件夾名稱。假設包含3個文件的“ orca_mini_7b”。

處理例外

處理文本生成期間可能發生的例外。如果拋出了ChatLengthException ，它將自動執行一個例程，以將消息重新限制為cutoff_message_histories函數限制令牌內的消息，並將其重新發送。這可以確保用戶具有平穩的聊天體驗，而不管令牌限制如何。

Websocket連接後面...

該項目旨在創建API後端，以啟用大型語言模型聊天機器人服務。它利用緩存管理器將消息和用戶配置文件存儲在REDIS中，以及消息管理器安全緩存消息，以使令牌數不超過可接受的限制。

緩存管理器

緩存管理器（ CacheManager ）負責處理用戶上下文信息和消息歷史記錄。它將這些數據存儲在REDIS中，從而可以輕鬆檢索和修改。經理提供了幾種與緩存交互的方法，例如：

• read_context_from_profile ：根據用戶的配置文件，從Redis讀取用戶的聊天上下文。
• create_context ：在Redis中創建一個新的用戶聊天上下文。
• reset_context ：將用戶的聊天上下文重置為默認值。
• update_message_histories ：更新特定角色（用戶，AI或系統）的消息歷史記錄。
• lpop_message_history / rpop_message_history ：從列表的左側或右端刪除並返回消息歷史記錄。
• append_message_history ：將消息歷史記錄附加到列表的末尾。
• get_message_history ：檢索特定角色的消息歷史記錄。
• delete_message_history ：刪除特定角色的消息歷史記錄。
• set_message_history ：為角色和索引設置特定的消息歷史記錄。

消息管理器

消息管理器（ MessageManager ）確保消息歷史記錄中的令牌數不超過指定的限制。它可以安全地處理在用戶的聊天上下文中添加，刪除和設置消息歷史，同時保持令牌限制。經理提供了幾種與消息歷史互動的方法，例如：

• add_message_history_safely ：將消息歷史記錄添加到用戶的聊天上下文中，以確保不超過令牌限制。
• pop_message_history_safely ：在更新令牌計數時從列表的右端刪除並返回消息歷史記錄。
• set_message_history_safely ：在用戶聊天上下文中設置特定消息歷史記錄，更新令牌計數並確保不超過令牌限制。

用法

要在您的項目中使用緩存管理器和消息管理器，請按以下方式導入它們：

 from app . utils . chat . managers . cache import CacheManager
from app . utils . chat . message_manager import MessageManager

然後，您可以使用他們的方法與REDIS緩存進行交互，並根據您的要求管理消息歷史記錄。

例如，創建一個新的用戶聊天上下文：

 user_id = "[email protected]"  # email format
chat_room_id = "example_chat_room_id"  # usually the 32 characters from `uuid.uuid4().hex`
default_context = UserChatContext . construct_default ( user_id = user_id , chat_room_id = chat_room_id )
await CacheManager . create_context ( user_chat_context = default_context )

安全地將消息歷史記錄添加到用戶的聊天上下文中：

 user_chat_context = await CacheManager . read_context_from_profile ( user_chat_profile = UserChatProfile ( user_id = user_id , chat_room_id = chat_room_id ))
content = "This is a sample message."
role = ChatRoles . USER  # can be enum such as ChatRoles.USER, ChatRoles.AI, ChatRoles.SYSTEM
await MessageManager . add_message_history_safely ( user_chat_context , content , role )

中間

該項目使用token_validator中間件和FastAPI應用程序中使用的其他中間件。這些中間Wares負責控制對API的訪問，確保僅處理授權和身份驗證的請求。

例子

以下中間件添加到FastAPI應用程序中：

1. 訪問控制中間軟件：確保僅處理授權請求。
2. CORS中間件：允許在應用程序配置中定義的特定起源的請求。
3. 受信任的主機中間件：確保請求來自App Configuration中定義的受信任主機。

訪問控制中間軟件

訪問控制中間軟件是在token_validator.py文件中定義的。它負責驗證API鍵和JWT令牌。

州經理

StateManager類用於初始化請求狀態變量。它設置請求時間，開始時間，IP地址和用戶令牌。

訪問控制

AccessControl類包含兩個用於驗證API鍵和JWT令牌的靜態方法：

1. api_service ：通過檢查請求中所需的查詢參數和標頭的存在來驗證API鍵。它調用Validator.api_key方法來驗證API密鑰，秘密和時間戳。
2. non_api_service ：通過在請求中檢查“授權”標頭或“授權” cookie的存在來驗證JWT令牌。它調用Validator.jwt方法來解碼和驗證JWT令牌。

驗證器

Validator類包含兩個用於驗證API密鑰和JWT令牌的靜態方法：

1. api_key ：驗證API訪問密鑰，Hashed Secret和時間戳。如果驗證成功，則返回UserToken對象。
2. jwt ：解碼和驗證JWT令牌。如果驗證成功，則返回UserToken對象。

訪問控制功能

access_control函數是一個異步函數，可處理中間件的請求和響應流。它使用StateManager類初始化了請求狀態，確定請求URL（API鍵或JWT令牌）所需的身份驗證類型，並使用AccessControl類驗證身份驗證。如果在驗證過程中發生錯誤，則會提出適當的HTTP異常。

令牌

令牌實用程序在token.py文件中定義。它包含兩個功能：

1. create_access_token ：使用給定的數據和到期時間創建JWT令牌。
2. token_decode ：解碼和驗證JWT令牌。如果令牌已過期或無法解碼，則會引起例外。

參數實用程序

params_utils.py文件包含使用HMAC和SHA256的哈希查詢參數和秘密密鑰的實用程序函數：

1. hash_params ：以查詢參數和秘密鍵為輸入，並返回基本64編碼的哈希德字符串。

日期公用事業

date_utils.py文件包含帶有實用程序功能的UTC類，用於使用日期和時間戳：

1. now ：以可選的小時差返回當前的UTC DateTime。
2. timestamp ：以可選的小時差返回當前的UTC時間戳。
3. timestamp_to_datetime ：將時間戳轉換為具有可選小時差的DateTime對象。

記錄器

logger.py文件包含ApiLogger類，該類可記錄API請求和響應信息，包括請求URL，方法，狀態代碼，客戶端信息，處理時間和錯誤詳細信息（如果適用）。在access_control函數的末尾調用記錄器函數以記錄已處理的請求和響應。

用法

要在您的FastAPI應用程序中使用token_validator中間件，只需導入access_control函數，然後將其添加為中間件：

 from app . middlewares . token_validator import access_control

app = FastAPI ()

app . add_middleware ( dispatch = access_control , middleware_class = BaseHTTPMiddleware )

確保還添加CORS和受信任的主機中間Wares以進行完整的訪問控制：

 app . add_middleware (
    CORSMiddleware ,
    allow_origins = config . allowed_sites ,
    allow_credentials = True ,
    allow_methods = [ "*" ],
    allow_headers = [ "*" ],
)

app . add_middleware (
    TrustedHostMiddleware ,
    allowed_hosts = config . trusted_hosts ,
    except_path = [ "/health" ],
)

現在， token_validator中間件和其他中間件將處理對您的FastAPI應用程序的任何傳入請求，以確保僅處理授權和認證的請求。

數據庫連接

此模塊app.database.connection提供了一個易於使用的接口，用於使用SQLalchemy和Redis來管理數據庫連接和執行SQL查詢。它支持MySQL，並且可以輕鬆地與該項目集成。

特徵

• 創建和刪除數據庫
• 創建和管理用戶
• 向用戶授予特權
• 執行RAW SQL查詢
• 通過異步支持管理數據庫會話
• 重新加緩存支持更快的數據訪問

用法

首先，從模塊導入所需類：

 from app . database . connection import MySQL , SQLAlchemy , CacheFactory

接下來，創建一個SQLAlchemy類的實例，並使用您的數據庫設置進行配置：

 from app . common . config import Config

config : Config = Config . get ()
db = SQLAlchemy ()
db . start ( config )

現在，您可以使用db實例執行SQL查詢並管理會話：

 # Execute a raw SQL query
result = await db . execute ( "SELECT * FROM users" )

# Use the run_in_session decorator to manage sessions
@ db . run_in_session
async def create_user ( session , username , password ):
    await session . execute ( "INSERT INTO users (username, password) VALUES (:username, :password)" , { "username" : username , "password" : password })

await create_user ( "JohnDoe" , "password123" )

要使用redis緩存，請創建一個CacheFactory類的實例，然後使用您的redis設置進行配置：

 cache = CacheFactory ()
cache . start ( config )

您現在可以使用cache實例與Redis交互：

 # Set a key in Redis
await cache . redis . set ( "my_key" , "my_value" )

# Get a key from Redis
value = await cache . redis . get ( "my_key" )

實際上，在此項目中， MySQL類在應用程序啟動時進行初始設置，並且所有數據庫連接僅使用模塊末尾的db和cache變量進行。 ？

所有DB設置將在app.common.app_settings中的create_app()中完成。例如， app.common.app_settings中的create_app()函數看起來像這樣：

 def create_app ( config : Config ) -> FastAPI :
    # Initialize app & db & js
    new_app = FastAPI (
        title = config . app_title ,
        description = config . app_description ,
        version = config . app_version ,
    )
    db . start ( config = config )
    cache . start ( config = config )
    js_url_initializer ( js_location = "app/web/main.dart.js" )
    # Register routers
    # ...
    return new_app

數據庫CRUD操作

該項目使用簡單有效的方法來處理使用SQLalchemy和兩個模塊和路徑的數據庫CRUD（創建，讀取，更新，刪除）操作： app.database.models.schema和app.database.crud 。

概述

app.database.models.schema

schema.py模塊負責使用SQLalchemy定義數據庫模型及其關係。它包括一組從Base繼承的類，該類別是declarative_base()的實例。每個類代表數據庫中的表，其屬性表示表中的列。這些類還從Mixin類中繼承，該類提供了所有模型的一些常見方法和屬性。

Mixin類

Mixin類為所有從中繼承的類提供了一些常見的屬性和方法。一些屬性包括：

• id ：表的整數主鍵。
• created_at ：dateTime創建記錄的時間。
• updated_at ：記錄最後更新何時的日期時間。
• ip_address ：創建或更新記錄的客戶端的IP地址。

它還提供了多種使用SQLalchemy進行CRUD操作的類方法，例如：

• add_all() ：將多個記錄添加到數據庫中。
• add_one() ：將單個記錄添加到數據庫中。
• update_where() ：基於過濾器中數據庫中的記錄。
• fetchall_filtered_by() ：從數據庫中獲取與提供過濾器相匹配的所有記錄。
• one_filtered_by() ：從數據庫中獲取與所提供過濾器匹配的單個記錄。
• first_filtered_by() ：從與提供過濾器匹配的數據庫中獲取第一個記錄。
• one_or_none_filtered_by() ：獲取單個記錄或如果沒有記錄匹配所提供的過濾器，則返回None 。

app.database.crud

users.py和api_keys.py模塊包含一組功能，使用schema.py中定義的類執行CRUD操作。這些功能使用Mixin類提供的類方法與數據庫進行交互。

該模塊中的某些功能包括：

• create_api_key() ：為用戶創建一個新的API密鑰。
• get_api_keys() ：檢索用戶的所有API鍵。
• get_api_key_owner() ：檢索API密鑰的所有者。
• get_api_key_and_owner() ：檢索API鍵及其所有者。
• update_api_key() ：更新API密鑰。
• delete_api_key() ：刪除API鍵。
• is_email_exist() ：檢查數據庫中是否存在電子郵件。
• get_me() ：根據用戶ID檢索用戶信息。
• is_valid_api_key() ：檢查API鍵是否有效。
• register_new_user() ：在數據庫中註冊新用戶。
• find_matched_user() ：在數據庫中找到帶有匹配電子郵件的用戶。

用法

要使用提供的CRUD操作，請從crud.py模塊中導入相關功能，並使用所需的參數調用它們。例如：

 import asyncio
from app . database . crud . users import register_new_user , get_me , is_email_exist
from app . database . crud . api_keys import create_api_key , get_api_keys , update_api_key , delete_api_key

async def main ():
    # `user_id` is an integer index in the MySQL database, and `email` is user's actual name
    # the email will be used as `user_id` in chat. Don't confuse with `user_id` in MySQL

    # Register a new user
    new_user = await register_new_user ( email = "[email protected]" , hashed_password = "..." )

    # Get user information
    user = await get_me ( user_id = 1 )

    # Check if an email exists in the database
    email_exists = await is_email_exist ( email = "[email protected]" )

    # Create a new API key for user with ID 1
    new_api_key = await create_api_key ( user_id = 1 , additional_key_info = { "user_memo" : "Test API Key" })

    # Get all API keys for user with ID 1
    api_keys = await get_api_keys ( user_id = 1 )

    # Update the first API key in the list
    updated_api_key = await update_api_key ( updated_key_info = { "user_memo" : "Updated Test API Key" }, access_key_id = api_keys [ 0 ]. id , user_id = 1 )

    # Delete the first API key in the list
    await delete_api_key ( access_key_id = api_keys [ 0 ]. id , access_key = api_keys [ 0 ]. access_key , user_id = 1 )

if __name__ == "__main__" :
    asyncio . run ( main ())