安全的關(guān)鍵.png)
使用這些基本 REST API 最佳實(shí)踐構(gòu)建出色的 API
大型語(yǔ)言模型（LLM）系統(tǒng)雖覆蓋廣泛主題，但其知識(shí)范疇僅限于截至某一時(shí)間節(jié)點(diǎn)的公開(kāi)信息。若要開(kāi)發(fā)能夠理解機(jī)密數(shù)據(jù)或最新信息的AI工具，則需將相關(guān)數(shù)據(jù)融入模型中。我們通常將此過(guò)程命名為檢索增強(qiáng)生成（RAG），也可稱之為模型“接地”。
對(duì)于此項(xiàng)目，代碼和示例數(shù)據(jù)集可在我的 GitHub 上獲取。
在本教程中，我們將充分利用 Python，并需要在您的計(jì)算機(jī)上進(jìn)行相關(guān)設(shè)置。我們將借助 Python 和 LangChain，將向量數(shù)據(jù)導(dǎo)入 Azure Cosmos DB for MongoDB vCore，并執(zhí)行相似性搜索。本演練的開(kāi)發(fā)與測(cè)試均基于 Python 3.11.4 版本。
首先在 demo_loader 目錄中設(shè)置 python 虛擬環(huán)境。
python -m venv venv激活您的環(huán)境并在 demo_loader 目錄中安裝依賴項(xiàng)：
venv\Scripts\activate
python -m pip install -r requirements.txt在 demo_loader 目錄中創(chuàng)建名為 ‘.env’ 的文件，以存儲(chǔ)環(huán)境變量。
OPENAI_API_KEY="**Your Open AI Key**"
MONGO_CONNECTION_STRING="mongodb+srv:**your connection string from Azure Cosmos DB**"
AZURE_STORAGE_CONNECTION_STRING="**"| 環(huán)境變量 | 描述 |
|---|---|
| OPENAI_API_KEY | 連接到 OpenAI API 的關(guān)鍵。如果您沒(méi)有 Open AI 的 API 密鑰，您可以按照概述的指南繼續(xù)。 |
| MONGO_CONNECTION_STRING | Azure Cosmos DB for MongoDB vCore 的連接字符串（見(jiàn)下文） |
| AZURE_STORAGE_CONNECTION_STRING | Azure 存儲(chǔ)帳戶的連接字符串（見(jiàn)下文） |
從“.env”文件中的環(huán)境變量MONGO_CONNECTION_STRING將包含Azure Cosmos DB for MongoDB vCore的連接字符串。您可以通過(guò)在Azure門戶中選擇Cosmos DB實(shí)例的“連接字符串”來(lái)獲取此值，過(guò)程中可能需要在相應(yīng)字段中輸入您的用戶名和密碼。
“.env” 文件中的 AZURE_STORAGE_CONNECTION_STRING 環(huán)境變量將包含 Azure 存儲(chǔ)帳戶的連接字符串。您可以通過(guò)在 Azure 門戶中選擇存儲(chǔ)帳戶實(shí)例的“訪問(wèn)密鑰”來(lái)獲取此值。
Python 文檔作為將數(shù)據(jù)加載到 Cosmos DB 矢量存儲(chǔ)和 Azure 存儲(chǔ)帳戶的主要接入點(diǎn)。所提供的代碼片段有助于將文檔數(shù)據(jù)導(dǎo)入到 Cosmos DB 中，并將相關(guān)聯(lián)的圖像文件上傳到 Azure Blob 存儲(chǔ)容器中。此流程涵蓋了對(duì)給定文件名列表中每個(gè)文檔的有序處理。文件名為 vectorstoreloader.py。
file_names = ['documents/Rocket_Propulsion_Elements_with_images.json','documents/Introduction_To_Rocket_Science_And_Engineering_with_images.json']
file_names += file_names
for file_name in file_names:
CosmosDBLoader(f"{file_name}").load()
image_loader = BlobLoader()
with open(file_name) as file:
data = json.load(file)
resource_id = data['resource_id']
for page in data['pages']:
base64_string = page['image'].replace("b'","").replace("'","")
# Decode the Base64 string into bytes
decoded_bytes = base64.b64decode(base64_string)
image_loader.load_binay_data(decoded_bytes,f"{resource_id}/{page['page_id']}.png","images")file_names 包含兩個(gè)示例 JSON 文件，每個(gè)文件代表一個(gè)擁有相關(guān)圖像的文檔，總計(jì)大約包含 160 個(gè)文檔頁(yè)面。file_names 列表中的每個(gè)文件名。image_loader 和 BlobLoader。.data 的變量中。resource_id。base64.b64decode 將其解碼為字節(jié)。image_loader 對(duì)象中，同時(shí)指定圖像的目標(biāo)路徑和容器類型。BlobLoader 以一種簡(jiǎn)單的方式運(yùn)行，它會(huì)將 JSON 文檔中經(jīng)過(guò) base64 編碼轉(zhuǎn)換后的圖像，以字節(jié)形式存儲(chǔ)在 Azure 存儲(chǔ)帳戶的“images”容器中，這一過(guò)程是通過(guò)調(diào)用“l(fā)oad_binary_data”函數(shù)來(lái)實(shí)現(xiàn)的。以下代碼段定義了一個(gè)類，該類利用從環(huán)境變量中獲取到的 Azure 存儲(chǔ)連接字符串，將二進(jìn)制數(shù)據(jù)上傳到 Azure Blob 存儲(chǔ)的指定容器中。
GitHub 存儲(chǔ)庫(kù)中的代碼當(dāng)前默認(rèn)使用名為“映像”的 Azure 存儲(chǔ)帳戶容器，若您希望繼續(xù)操作，可以選擇創(chuàng)建一個(gè)名為“images”的容器，或者根據(jù)已有的容器名稱調(diào)整此段代碼。
例如，在調(diào)用 image_loader.load_binary_data 函數(shù)時(shí)，您可以這樣指定參數(shù)：
image_loader.load_binary_data(decoded_bytes, f"{resource_id}/{page['page_id']}.png", "images")其中，“decoded_bytes”代表解碼后的圖像字節(jié)數(shù)據(jù)，“f”{resource_id}/{page[‘page_id’]}.png””定義了 Blob 的名稱，而最后的“images”則是目標(biāo)容器的名稱。
from os import environ
from dotenv import load_dotenv
from azure.storage.blob import BlobServiceClient
load_dotenv(override=True)
class BlobLoader():
def __init__(self):
connection_string = environ.get("AZURE_STORAGE_CONNECTION_STRING")
# Create the BlobServiceClient object
self.blob_service_client = BlobServiceClient.from_connection_string(connection_string)
def load_binay_data(self,data, blob_name:str, container_name:str):
blob_client = self.blob_service_client.get_blob_client(container=container_name, blob=blob_name)
# Upload the blob data - default blob type is BlockBlob
blob_client.upload_blob(data,overwrite=True)load_dotenv 從 .env 文件中加載環(huán)境變量到腳本的運(yùn)行環(huán)境中。BlobLoader 類是一個(gè)用于將二進(jìn)制數(shù)據(jù)上傳到 Azure Blob 存儲(chǔ)容器的封裝器。BlobLoader 類的 __init__ 方法中：
environ.get("AZURE_STORAGE_CONNECTION_STRING") 從環(huán)境變量中檢索 Azure 存儲(chǔ)連接字符串。BlobServiceClient 對(duì)象。BlobLoader 類的 load_binary_data 方法中：
data（二進(jìn)制數(shù)據(jù)）、blob_name（Blob 名稱）和 container_name（容器名稱）。BlobServiceClient 的 get_blob_client 方法獲取一個(gè) blob_client 對(duì)象。blob_client 的 upload_blob 方法上傳 data，并通過(guò) overwrite=True 參數(shù)指示如果已存在同名的 Blob，則應(yīng)覆蓋它。現(xiàn)在我們已經(jīng)查看了代碼，讓我們只需從 demo_loader 目錄執(zhí)行以下命令即可加載文檔。
python vectorstoreloader.py使用 MongoDB Compass（或類似工具）驗(yàn)證文檔是否成功加載。
請(qǐng)驗(yàn)證是否已成功將“png”圖像上傳到 Azure 存儲(chǔ)帳戶的“image”容器。
訪問(wèn)該目錄以顯示圖像列表。
對(duì)一兩張圖像進(jìn)行采樣，以驗(yàn)證字節(jié)是否已正確解碼。
祝賀！在本指南的第二部分和第三部分中，您已成功利用 LangChain 將文檔從 JSON 文件導(dǎo)入至 Azure Cosmos DB for MongoDB vCore，并進(jìn)一步將二進(jìn)制文檔上傳至 Azure 存儲(chǔ)帳戶，以供應(yīng)用程序便捷使用。
FastAPI 是一款前沿的 Web 框架，專為 Python 3.8+ 量身打造，助力開(kāi)發(fā)者迅速構(gòu)建高效 API。它憑借卓越的性能脫穎而出，得益于與 Starlette 和 Pydantic 的深度融合，其速度可媲美 NodeJS 和 Go。FastAPI 不僅將開(kāi)發(fā)效率提升了高達(dá) 300%，更將人為錯(cuò)誤率降低了 40%。加之出色的編輯器支持、詳盡的文檔資源，F(xiàn)astAPI 確保了直觀易用的開(kāi)發(fā)體驗(yàn)，有效減少了代碼冗余，并最大限度地激發(fā)了生產(chǎn)力。
demo_api 項(xiàng)目經(jīng)過(guò)設(shè)置，能夠輕松應(yīng)對(duì) Web 請(qǐng)求，并高效處理與 Azure Cosmos DB for MongoDB 矢量數(shù)據(jù)庫(kù)及 Azure 存儲(chǔ)帳戶相關(guān)的數(shù)據(jù)事務(wù)。這一設(shè)計(jì)使得代碼結(jié)構(gòu)清晰明了，為開(kāi)發(fā)新功能及修復(fù)現(xiàn)有問(wèn)題提供了極大的便利。
在本教程中，Python 用于開(kāi)發(fā) FastAPI Web 界面，這需要在用戶的計(jì)算機(jī)上進(jìn)行設(shè)置。本教程涉及使用 Python 和 LangChain 對(duì) Azure Cosmos DB for MongoDB vCore 進(jìn)行矢量搜索，以及執(zhí)行 Q&A RAG 鏈。在本演練的整個(gè)開(kāi)發(fā)和測(cè)試過(guò)程中使用了 Python 版本 3.11.4。
首先在 demo_api 目錄中設(shè)置 python 虛擬環(huán)境。
python -m venv venv使用 demo_api 目錄中的需求文件激活您的環(huán)境并安裝依賴項(xiàng)：
venv\Scripts\activate
python -m pip install -r requirements.txt在 demo_api 目錄中創(chuàng)建一個(gè)名為“.env”的文件，以存儲(chǔ)環(huán)境變量。
VECTORDB_ENDPOINT='https://[your-web-app].azurewebsites.net/'
VECTORDB_API_KEY='**your_key**'
OPENAI_API_KEY="**Your Open AI Key**"
MONGO_CONNECTION_STRING="mongodb+srv:**your connection string from Azure Cosmos DB**"
AZURE_STORAGE_CONNECTION_STRING="**"
AZURE_STORAGE_CONTAINER="images"| 環(huán)境變量 | 描述 |
|---|---|
| VECTORDB_ENDPOINT | Weaviate 的 URL 端點(diǎn)。您可以保留默認(rèn)值：’https：//%5Byour-web-app%5D.azurewebsites.net/’因?yàn)楸揪毩?xí)中不使用 Weaviate。 |
| VECTORDB_API_KEY | Weaviate 身份驗(yàn)證 api 密鑰。您可以保留默認(rèn)值：‘**your_key**’，因?yàn)楸揪毩?xí)中不使用 Weaviate。 |
| OPENAI_API_KEY | 連接到 OpenAI API 的關(guān)鍵。如果您沒(méi)有 Open AI 的 API 密鑰，您可以按照概述的指南繼續(xù)。 |
| MONGO_CONNECTION_STRING | Azure Cosmos DB for MongoDB vCore 的連接字符串 |
| AZURE_STORAGE_CONNECTION_STRING | Azure 存儲(chǔ)帳戶的連接字符串 |
| AZURE_STORAGE_CONTAINER | 第 1 部分中使用的容器名稱默認(rèn)為 ‘images‘。 |
在 GitHub 存儲(chǔ)庫(kù)中，安排設(shè)置是為了促進(jìn)不同矢量存儲(chǔ)系統(tǒng)之間的轉(zhuǎn)換，特別是 Azure Cosmos DB for MongoDB 和 Weaviate。但是，對(duì)此功能的深入探索計(jì)劃在下一篇文章中介紹。
配置環(huán)境并設(shè)置變量后，我們就可以啟動(dòng) FastAPI 服務(wù)器了。從 demo_api 目錄運(yùn)行以下命令以啟動(dòng)服務(wù)器。
python main.py默認(rèn)情況下，F(xiàn)astAPI 服務(wù)器在 localhost 環(huán)回 127.0.0.1 端口 8000 上啟動(dòng)。您可以使用以下 localhost 地址訪問(wèn) Swagger 文檔：http://127.0.0.1:8000/docs
借助在本地運(yùn)行的 FastAPI 服務(wù)，我們可以通過(guò)其終端節(jié)點(diǎn)對(duì) vector 數(shù)據(jù)庫(kù)執(zhí)行向量搜索。在此場(chǎng)景下，我們輸入查詢?cè)~ “what is a turbofan”，隨后系統(tǒng)將返回一系列與之匹配的條目，并在 Web 界面上展示。訪問(wèn)路徑為 /search/{query}。
點(diǎn)擊 “Try It out” 按鈕，針對(duì) /search/{query} 路徑進(jìn)行測(cè)試。
為查詢輸入 “what is a turbofan” ，然后單擊 Execute（執(zhí)行）。
響應(yīng)包含一個(gè) JSON 數(shù)組，其中包含下面描述的資源。
{resource_id": "b801d68a8a0d4f1dac72dc6c170c395b",
"page_id": "cd3eea19611c423aaaf85e6da691f23d",
"title": "Rocket Propulsion Elements",
"source": "Chapter 1 - Classification (page-2)",
"content":"..text.."}檢索到的資源（或文件）可以作為大型語(yǔ)言模型（LLM）處理查詢的基礎(chǔ)。在這方面，我們的 RAG 端點(diǎn)特別適用于問(wèn)答（Q&A）場(chǎng)景。當(dāng)用戶提交查詢，例如“什么是渦輪風(fēng)扇”時(shí)，響應(yīng)中除了會(huì)包含相關(guān)的文檔，還會(huì)包含一個(gè)“text”字段。這個(gè)“text”字段構(gòu)成了 LLM 的答案，這些答案是從我們與查詢相匹配的向量搜索結(jié)果中得出的。具體的 Q&A 端點(diǎn)路徑為?/search/qa/{query}。
來(lái)自 RAG 終端節(jié)點(diǎn)的響應(yīng)包含兩部分內(nèi)容：一部分是存儲(chǔ)在 ‘text’ 值中的、針對(duì)查詢的大型語(yǔ)言模型（LLM）“答案”，另一部分則是位于“ResourceCollection”下、用于支撐該答案的資源列表。
{
"text": "A turbofan is a type of air-breathing engine that uses a fan to compress air and mix it with fuel for combustion. It is a type of ducted engine that is more fuel-efficient than a turbojet. Turbofans are commonly used in commercial aircraft for propulsion.",
"ResourceCollection": [
{
"resource_id": "b801d68a8a0d4f1dac72dc6c170c395b",
"page_id": "cd3eea19611c423aaaf85e6da691f23d",
"title": "Rocket Propulsion Elements",
"source": "Chapter 1 - Classification (page-2)",
"content": "....text..."
},...]
} 查詢首先會(huì)通過(guò)向量搜索來(lái)檢索相關(guān)文檔，然后會(huì)提示大型語(yǔ)言模型（LLM）利用這些文檔和查詢來(lái)生成答案。
在 API 中集成 RAG 時(shí)，Web 搜索組件會(huì)啟動(dòng)所有請(qǐng)求，然后是搜索服務(wù)，最后是數(shù)據(jù)組件。在本示例中，我們采用的是 MongoDB 數(shù)據(jù)搜索，它具體連接到的是 Azure Cosmos DB for MongoDB vCore。在這幾層架構(gòu)中，Model 組件會(huì)在各層之間傳遞，而大部分的 LangChain 代碼則主要駐留在服務(wù)層。我采用這種方法的目的，是為了在使用相同的鏈時(shí)能夠方便地切換不同的數(shù)據(jù)源。
Web 層負(fù)責(zé)路由請(qǐng)求并管理與調(diào)用方的通信。在 Q&A RAG 的上下文中，我們只有兩個(gè)代碼文件：主要的 FastAPI 應(yīng)用程序 （main.py） 和用于搜索的 API 路由器 （search.py）。
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from web import search, content
app = FastAPI()
origins = ["*"]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(search.router)
app.include_router(content.router)
@app.get("/")
def get() -> str:
return "running"
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", reload=True)main.py 這段代碼啟動(dòng)并配置了一個(gè)集成了 CORS（跨源資源共享）中間件的 FastAPI Web 應(yīng)用程序，使其能夠順暢地處理來(lái)自不同源的請(qǐng)求。該應(yīng)用程序通過(guò)整合搜索和內(nèi)容相關(guān)的終端節(jié)點(diǎn)路由器，能夠高效地應(yīng)對(duì)多種類型的請(qǐng)求。同時(shí)，它還定義了一個(gè)默認(rèn)端點(diǎn)（“/”），該端點(diǎn)會(huì)返回一個(gè)簡(jiǎn)單的 “running” 消息，這對(duì)于測(cè)試應(yīng)用程序的功能來(lái)說(shuō)是一個(gè)實(shí)用的工具。在開(kāi)發(fā)和測(cè)試階段，我充分利用了這個(gè)簡(jiǎn)潔的消息來(lái)驗(yàn)證應(yīng)用狀態(tài)。利用 uvicorn 運(yùn)行 FastAPI 應(yīng)用程序，并啟用自動(dòng)重載功能，極大地提升了開(kāi)發(fā)效率，減少了在測(cè)試和優(yōu)化應(yīng)用程序過(guò)程中所需的時(shí)間和精力。
from fastapi import APIRouter
from service import search as search
from model.resource import Resource
from model.airesults import AIResults
router = APIRouter(prefix="/search")
@router.get("/{query}")
def get_search(query) -> list[Resource]:
return search.get_query(query)
@router.get("/summary/{query}")
def get_query_summary(query) -> AIResults:
return search.get_query_summary(query)
@router.get("/qa/{query}")
def get_query_qa(query) -> AIResults:
return search.get_qa_from_query(query)Web 搜索代碼負(fù)責(zé)管理各種與搜索相關(guān)的終端節(jié)點(diǎn)。它有助于集成 service 模塊中的功能，并且包含處理資源和 AI 結(jié)果的模型。路由器的前綴明顯為 “/search”，并包含三個(gè)基本的 GET 路由，每個(gè)路由都有其特定的用途：searchResource 和 searchAIResults。
第一個(gè)端點(diǎn)?/search/{query}?在根據(jù)提供的查詢檢索搜索結(jié)果方面起著關(guān)鍵作用。此過(guò)程涉及調(diào)用 service 模塊中的函數(shù)，然后該函數(shù)會(huì)編排并返回一個(gè)對(duì)象列表，這些對(duì)象封裝了所獲得的搜索結(jié)果的本質(zhì)。
繼續(xù)介紹該路線，/search/summary/{query} 途徑旨在獲取與提供的查詢相關(guān)聯(lián)的搜索結(jié)果的摘要。這是通過(guò)調(diào)用 service 模塊中相應(yīng)的函數(shù)來(lái)實(shí)現(xiàn)的，該函數(shù)最終返回一個(gè)封裝了搜索過(guò)程匯總結(jié)果的對(duì)象。
最后，/search/qa/{query}?路由側(cè)重于使用 LangChain 的 RAG 模式來(lái)檢索問(wèn)答結(jié)果。這是通過(guò)調(diào)用 service 模塊中的某個(gè)函數(shù)來(lái)實(shí)現(xiàn)的。
服務(wù)層構(gòu)成了我們主要業(yè)務(wù)邏輯的基礎(chǔ)。在這個(gè)特定用例的上下文中，服務(wù)層扮演著 LangChain 代碼存儲(chǔ)庫(kù)的重要角色，將它們定位為我們技術(shù)框架中的核心組件。
from data.mongodb import search as search
from langchain.docstore.document import Document
from langchain.chains.combine_documents.stuff import StuffDocumentsChain
from langchain.chains.llm import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from model.airesults import AIResults
from model.resource import Resource
template:str = """Use the following pieces of context to answer the question at the end.
If none of the pieces of context answer the question, just say you don't know.
If you don't know the answer, just say that you don't know, don't try to make up an answer.
Use three sentences maximum and keep the answer as concise as possible.
{context}
Question: {question}
Answer:"""
def get_query(query:str)-> list[Resource]:
resources, docs = search.similarity_search(query)
return resources
def get_query_summary(query:str) -> str:
prompt_template = """Write a summary of the following:
"{text}"
CONCISE SUMMARY:"""
prompt = PromptTemplate.from_template(prompt_template)
resources, docs = search.similarity_search(query)
if len(resources)==0:return AIResults(text="No Documents Found",ResourceCollection=resources)
llm = ChatOpenAI(temperature=0, model_name="gpt-3.5-turbo-16k")
llm_chain = LLMChain(llm=llm, prompt=prompt)
stuff_chain = StuffDocumentsChain(llm_chain=llm_chain, document_variable_name="text")
return AIResults(stuff_chain.run(docs),resources)
def get_qa_from_query(query:str) -> str:
resources, docs = search.similarity_search(query)
if len(resources) ==0 :return AIResults(text="No Documents Found",ResourceCollection=resources)
custom_rag_prompt = PromptTemplate.from_template(template)
llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0)
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
content = format_docs(docs)
rag_chain = (
{"context": lambda x: content , "question": RunnablePassthrough()}
| custom_rag_prompt
| llm
| StrOutputParser()
)
return AIResults(text=rag_chain.invoke(query),ResourceCollection=resources)該 search.py 服務(wù)展示了數(shù)據(jù)庫(kù)交互、語(yǔ)言模型以及提示模板的集成，從而能夠根據(jù)用戶查詢提供摘要和問(wèn)答等功能。起初，它定義了一個(gè)模板字符串，該模板用于創(chuàng)建與 RAG（檢索增強(qiáng)型生成模型）相關(guān)的問(wèn)題和答案。此模板內(nèi)嵌了上下文和問(wèn)題提示的占位符。搜索服務(wù)包含以下三個(gè)功能：
第一個(gè)搜索服務(wù)函數(shù)：該函數(shù)接收查詢字符串作為輸入，隨后在數(shù)據(jù)庫(kù)中執(zhí)行與查詢相似的資源向量搜索，并返回匹配資源的列表。函數(shù)簽名為?get_query(query: str) -> list[Resource]。
第二個(gè)函數(shù) ：該函數(shù)生成與給定查詢相關(guān)聯(lián)的文檔的簡(jiǎn)潔摘要。它首先通過(guò)向量搜索找到相關(guān)文檔，然后利用某個(gè)文檔鏈（在描述中被誤寫(xiě)為 StuffDocumentChain，可能是指具體的文檔處理鏈或模塊）來(lái)總結(jié)這些文檔。函數(shù)簽名為 get_query_summary(query: str) -> str。
最后一個(gè)函數(shù) ：該函數(shù)檢索與查詢相關(guān)的文檔，并基于這些文檔構(gòu)建問(wèn)題的答案。它運(yùn)用全局提示來(lái)生成問(wèn)題和答案，同時(shí)利用從向量搜索返回的文檔作為輸入。函數(shù)簽名為 get_qa_from_query(query: str) -> str。
最后，我們到達(dá)了數(shù)據(jù)層。盡管 GitHub 存儲(chǔ)庫(kù)包含 Weaviate 和 Cosmos DB for MongoDB 的代碼（但在此討論中僅涉及 MongoDB 的部分），本系列將專注于與 Cosmos DB 的連接。在數(shù)據(jù)層中，我們利用?init.py?文件中的模塊作為單例來(lái)處理與數(shù)據(jù)庫(kù)相關(guān)的連接，這是一種清晰的方法，盡管有時(shí)我也會(huì)考慮使用全局變量作為替代方案。除了?init.py?文件，search.py?文件還負(fù)責(zé)執(zhí)行對(duì) Cosmos DB 的向量搜索。
from os import environ
from dotenv import load_dotenv
from pymongo import MongoClient
from pymongo.collection import Collection
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores.azure_cosmos_db import AzureCosmosDBVectorSearch
load_dotenv(override=True)
collection: Collection | None = None
vector_store: AzureCosmosDBVectorSearch | None=None
def mongodb_init():
MONGO_CONNECTION_STRING = environ.get("MONGO_CONNECTION_STRING")
DB_NAME = "research"
COLLECTION_NAME = "resources"
INDEX_NAME = "vectorSearchIndex"
global collection, vector_store
client = MongoClient(MONGO_CONNECTION_STRING)
db = client[DB_NAME]
collection = db[COLLECTION_NAME]
vector_store = AzureCosmosDBVectorSearch.from_connection_string(MONGO_CONNECTION_STRING,
DB_NAME + "." + COLLECTION_NAME,
OpenAIEmbeddings(disallowed_special=()),
index_name=INDEX_NAME
)
mongodb_init()該文件首先使用 load_dotenv(override=True) 方法從 .env 文件中加載環(huán)境變量。隨后，它聲明了兩個(gè)全局變量：一個(gè)表示 CosmosDB MongoDB 集合的 collection，另一個(gè)表示 CosmosDB 向量搜索的 vector_store。接著，使用 MongoDB 的連接字符串初始化 MongoClient，并通過(guò)該客戶端檢索數(shù)據(jù)庫(kù)和集合對(duì)象。之后，這些對(duì)象被傳遞到 AzureCosmosDBVectorSearch.from_connection_string() 方法中。
from model.resource import Resource
from .init import collection, vector_store
from langchain.docstore.document import Document
from typing import List, Optional, Union
def results_to_model(result:Document) -> Resource:
return Resource(resource_id = result.metadata["resource_id"],
page_id = result.metadata["page_id"],
title=result.metadata["title"],
source=f"{result.metadata['chapter']} (page-{result.metadata['pagenumber']})",
content=result.page_content)
def similarity_search(query:str)-> tuple[list[Resource], list[Document]]:
docs = vector_store.similarity_search_with_score(query,4)
# Cosine Similarity:
#It measures the cosine of the angle between two vectors in an n-dimensional space.
#The values of similarity metrics typically range between 0 and 1, with higher values indicating greater similarity between the vectors.
docs_filters = [doc for doc, score in docs if score >=.75]
# List the scores for documents
for doc, score in docs:
print(score)
# Print number of documents passing score threshold
print(len(docs_filters))
return [results_to_model(document) for document in docs_filters],docs_filters數(shù)據(jù)檢索代碼通過(guò)我們的單例訪問(wèn)全局變量，當(dāng)前包含兩個(gè)關(guān)鍵函數(shù)，分別涉及相似性搜索和數(shù)據(jù)轉(zhuǎn)換。
第一個(gè)函數(shù)接收一個(gè)?LangChain?對(duì)象作為輸入，并輸出一個(gè)?ResourceDocument?對(duì)象。該函數(shù)從提供的?Document?中抽取元數(shù)據(jù)屬性，例如?resource_id、page_id、標(biāo)題、章節(jié)以及頁(yè)碼。此函數(shù)的目的是確保向量搜索返回的資源能夠在?LangChain?外部得到重用。函數(shù)簽名為?results_to_model(result: Document) -> Resource。
第二個(gè)函數(shù)利用指定的查詢來(lái)執(zhí)行相似性搜索。它調(diào)用 vector_store 的某個(gè)方法來(lái)檢索與給定查詢相似的文檔，以及這些文檔各自的相似性分?jǐn)?shù)。該函數(shù)根據(jù)閾值分?jǐn)?shù)（0.75）對(duì)檢索到的文檔進(jìn)行篩選，并利用 results_to_model 函數(shù)將篩選后的文檔轉(zhuǎn)換為 Resource 對(duì)象。最終，該函數(shù)返回一個(gè)元組，包含轉(zhuǎn)換后的 Resource 對(duì)象列表和原始的 Document 對(duì)象列表。函數(shù)簽名為 similarity_search(query: str) -> Tuple[List[Resource], List[Document]]。
恭喜您順利完成了 Web 框架的設(shè)置！這確實(shí)是一項(xiàng)艱巨的任務(wù)，但為第3部分中即將展開(kāi)的工作奠定了堅(jiān)實(shí)的基礎(chǔ)。
原文鏈接：https://stochasticcoder.com/2024/02/27/langchain-rag-with-react-fastapi-cosmos-db-vector-part-1/
https://stochasticcoder.com/2024/02/29/langchain-rag-with-react-fastapi-cosmos-db-vector-part-2/
對(duì)比大模型API的內(nèi)容創(chuàng)意新穎性、情感共鳴力、商業(yè)轉(zhuǎn)化潛力
一鍵對(duì)比試用API 限時(shí)免費(fèi)