使用這些基本 REST API 最佳實踐構建出色的 API
大型語言模型（LLM）系統雖覆蓋廣泛主題，但其知識范疇僅限于截至某一時間節點的公開信息。若要開發能夠理解機密數據或最新信息的AI工具，則需將相關數據融入模型中。我們通常將此過程命名為檢索增強生成（RAG），也可稱之為模型“接地”。
對于此項目，代碼和示例數據集可在我的 GitHub 上獲取。
在本教程中，我們將充分利用 Python，并需要在您的計算機上進行相關設置。我們將借助 Python 和 LangChain，將向量數據導入 Azure Cosmos DB for MongoDB vCore，并執行相似性搜索。本演練的開發與測試均基于 Python 3.11.4 版本。
首先在 demo_loader 目錄中設置 python 虛擬環境。
python -m venv venv激活您的環境并在 demo_loader 目錄中安裝依賴項：
venv\Scripts\activate
python -m pip install -r requirements.txt在 demo_loader 目錄中創建名為 ‘.env’ 的文件，以存儲環境變量。
OPENAI_API_KEY="**Your Open AI Key**"
MONGO_CONNECTION_STRING="mongodb+srv:**your connection string from Azure Cosmos DB**"
AZURE_STORAGE_CONNECTION_STRING="**"| 環境變量 | 描述 |
|---|---|
| OPENAI_API_KEY | 連接到 OpenAI API 的關鍵。如果您沒有 Open AI 的 API 密鑰，您可以按照概述的指南繼續。 |
| MONGO_CONNECTION_STRING | Azure Cosmos DB for MongoDB vCore 的連接字符串（見下文） |
| AZURE_STORAGE_CONNECTION_STRING | Azure 存儲帳戶的連接字符串（見下文） |
從“.env”文件中的環境變量MONGO_CONNECTION_STRING將包含Azure Cosmos DB for MongoDB vCore的連接字符串。您可以通過在Azure門戶中選擇Cosmos DB實例的“連接字符串”來獲取此值，過程中可能需要在相應字段中輸入您的用戶名和密碼。
“.env” 文件中的 AZURE_STORAGE_CONNECTION_STRING 環境變量將包含 Azure 存儲帳戶的連接字符串。您可以通過在 Azure 門戶中選擇存儲帳戶實例的“訪問密鑰”來獲取此值。
Python 文檔作為將數據加載到 Cosmos DB 矢量存儲和 Azure 存儲帳戶的主要接入點。所提供的代碼片段有助于將文檔數據導入到 Cosmos DB 中，并將相關聯的圖像文件上傳到 Azure Blob 存儲容器中。此流程涵蓋了對給定文件名列表中每個文檔的有序處理。文件名為 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 包含兩個示例 JSON 文件，每個文件代表一個擁有相關圖像的文檔，總計大約包含 160 個文檔頁面。file_names 列表中的每個文件名。image_loader 和 BlobLoader。.data 的變量中。resource_id。base64.b64decode 將其解碼為字節。image_loader 對象中，同時指定圖像的目標路徑和容器類型。BlobLoader 以一種簡單的方式運行，它會將 JSON 文檔中經過 base64 編碼轉換后的圖像，以字節形式存儲在 Azure 存儲帳戶的“images”容器中，這一過程是通過調用“load_binary_data”函數來實現的。以下代碼段定義了一個類，該類利用從環境變量中獲取到的 Azure 存儲連接字符串，將二進制數據上傳到 Azure Blob 存儲的指定容器中。
GitHub 存儲庫中的代碼當前默認使用名為“映像”的 Azure 存儲帳戶容器，若您希望繼續操作，可以選擇創建一個名為“images”的容器，或者根據已有的容器名稱調整此段代碼。
例如，在調用 image_loader.load_binary_data 函數時，您可以這樣指定參數：
image_loader.load_binary_data(decoded_bytes, f"{resource_id}/{page['page_id']}.png", "images")其中，“decoded_bytes”代表解碼后的圖像字節數據，“f”{resource_id}/{page[‘page_id’]}.png””定義了 Blob 的名稱，而最后的“images”則是目標容器的名稱。
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 文件中加載環境變量到腳本的運行環境中。BlobLoader 類是一個用于將二進制數據上傳到 Azure Blob 存儲容器的封裝器。BlobLoader 類的 __init__ 方法中：
environ.get("AZURE_STORAGE_CONNECTION_STRING") 從環境變量中檢索 Azure 存儲連接字符串。BlobServiceClient 對象。BlobLoader 類的 load_binary_data 方法中：
data（二進制數據）、blob_name（Blob 名稱）和 container_name（容器名稱）。BlobServiceClient 的 get_blob_client 方法獲取一個 blob_client 對象。blob_client 的 upload_blob 方法上傳 data，并通過 overwrite=True 參數指示如果已存在同名的 Blob，則應覆蓋它。現在我們已經查看了代碼，讓我們只需從 demo_loader 目錄執行以下命令即可加載文檔。
python vectorstoreloader.py使用 MongoDB Compass（或類似工具）驗證文檔是否成功加載。
請驗證是否已成功將“png”圖像上傳到 Azure 存儲帳戶的“image”容器。
訪問該目錄以顯示圖像列表。
對一兩張圖像進行采樣，以驗證字節是否已正確解碼。
祝賀！在本指南的第二部分和第三部分中，您已成功利用 LangChain 將文檔從 JSON 文件導入至 Azure Cosmos DB for MongoDB vCore，并進一步將二進制文檔上傳至 Azure 存儲帳戶，以供應用程序便捷使用。
FastAPI 是一款前沿的 Web 框架，專為 Python 3.8+ 量身打造，助力開發者迅速構建高效 API。它憑借卓越的性能脫穎而出，得益于與 Starlette 和 Pydantic 的深度融合，其速度可媲美 NodeJS 和 Go。FastAPI 不僅將開發效率提升了高達 300%，更將人為錯誤率降低了 40%。加之出色的編輯器支持、詳盡的文檔資源，FastAPI 確保了直觀易用的開發體驗，有效減少了代碼冗余，并最大限度地激發了生產力。
demo_api 項目經過設置，能夠輕松應對 Web 請求，并高效處理與 Azure Cosmos DB for MongoDB 矢量數據庫及 Azure 存儲帳戶相關的數據事務。這一設計使得代碼結構清晰明了，為開發新功能及修復現有問題提供了極大的便利。
在本教程中，Python 用于開發 FastAPI Web 界面，這需要在用戶的計算機上進行設置。本教程涉及使用 Python 和 LangChain 對 Azure Cosmos DB for MongoDB vCore 進行矢量搜索，以及執行 Q&A RAG 鏈。在本演練的整個開發和測試過程中使用了 Python 版本 3.11.4。
首先在 demo_api 目錄中設置 python 虛擬環境。
python -m venv venv使用 demo_api 目錄中的需求文件激活您的環境并安裝依賴項：
venv\Scripts\activate
python -m pip install -r requirements.txt在 demo_api 目錄中創建一個名為“.env”的文件，以存儲環境變量。
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"| 環境變量 | 描述 |
|---|---|
| VECTORDB_ENDPOINT | Weaviate 的 URL 端點。您可以保留默認值：’https：//%5Byour-web-app%5D.azurewebsites.net/’因為本練習中不使用 Weaviate。 |
| VECTORDB_API_KEY | Weaviate 身份驗證 api 密鑰。您可以保留默認值：‘**your_key**’，因為本練習中不使用 Weaviate。 |
| OPENAI_API_KEY | 連接到 OpenAI API 的關鍵。如果您沒有 Open AI 的 API 密鑰，您可以按照概述的指南繼續。 |
| MONGO_CONNECTION_STRING | Azure Cosmos DB for MongoDB vCore 的連接字符串 |
| AZURE_STORAGE_CONNECTION_STRING | Azure 存儲帳戶的連接字符串 |
| AZURE_STORAGE_CONTAINER | 第 1 部分中使用的容器名稱默認為 ‘images‘。 |
在 GitHub 存儲庫中，安排設置是為了促進不同矢量存儲系統之間的轉換，特別是 Azure Cosmos DB for MongoDB 和 Weaviate。但是，對此功能的深入探索計劃在下一篇文章中介紹。
配置環境并設置變量后，我們就可以啟動 FastAPI 服務器了。從 demo_api 目錄運行以下命令以啟動服務器。
python main.py默認情況下，FastAPI 服務器在 localhost 環回 127.0.0.1 端口 8000 上啟動。您可以使用以下 localhost 地址訪問 Swagger 文檔：http://127.0.0.1:8000/docs
借助在本地運行的 FastAPI 服務，我們可以通過其終端節點對 vector 數據庫執行向量搜索。在此場景下，我們輸入查詢詞 “what is a turbofan”，隨后系統將返回一系列與之匹配的條目，并在 Web 界面上展示。訪問路徑為 /search/{query}。
點擊 “Try It out” 按鈕，針對 /search/{query} 路徑進行測試。
為查詢輸入 “what is a turbofan” ，然后單擊 Execute（執行）。
響應包含一個 JSON 數組，其中包含下面描述的資源。
{resource_id": "b801d68a8a0d4f1dac72dc6c170c395b",
"page_id": "cd3eea19611c423aaaf85e6da691f23d",
"title": "Rocket Propulsion Elements",
"source": "Chapter 1 - Classification (page-2)",
"content":"..text.."}檢索到的資源（或文件）可以作為大型語言模型（LLM）處理查詢的基礎。在這方面，我們的 RAG 端點特別適用于問答（Q&A）場景。當用戶提交查詢，例如“什么是渦輪風扇”時，響應中除了會包含相關的文檔，還會包含一個“text”字段。這個“text”字段構成了 LLM 的答案，這些答案是從我們與查詢相匹配的向量搜索結果中得出的。具體的 Q&A 端點路徑為?/search/qa/{query}。
來自 RAG 終端節點的響應包含兩部分內容：一部分是存儲在 ‘text’ 值中的、針對查詢的大型語言模型（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..."
},...]
} 查詢首先會通過向量搜索來檢索相關文檔，然后會提示大型語言模型（LLM）利用這些文檔和查詢來生成答案。
在 API 中集成 RAG 時，Web 搜索組件會啟動所有請求，然后是搜索服務，最后是數據組件。在本示例中，我們采用的是 MongoDB 數據搜索，它具體連接到的是 Azure Cosmos DB for MongoDB vCore。在這幾層架構中，Model 組件會在各層之間傳遞，而大部分的 LangChain 代碼則主要駐留在服務層。我采用這種方法的目的，是為了在使用相同的鏈時能夠方便地切換不同的數據源。
Web 層負責路由請求并管理與調用方的通信。在 Q&A RAG 的上下文中，我們只有兩個代碼文件：主要的 FastAPI 應用程序 （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 這段代碼啟動并配置了一個集成了 CORS（跨源資源共享）中間件的 FastAPI Web 應用程序，使其能夠順暢地處理來自不同源的請求。該應用程序通過整合搜索和內容相關的終端節點路由器，能夠高效地應對多種類型的請求。同時，它還定義了一個默認端點（“/”），該端點會返回一個簡單的 “running” 消息，這對于測試應用程序的功能來說是一個實用的工具。在開發和測試階段，我充分利用了這個簡潔的消息來驗證應用狀態。利用 uvicorn 運行 FastAPI 應用程序，并啟用自動重載功能，極大地提升了開發效率，減少了在測試和優化應用程序過程中所需的時間和精力。
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 搜索代碼負責管理各種與搜索相關的終端節點。它有助于集成 service 模塊中的功能，并且包含處理資源和 AI 結果的模型。路由器的前綴明顯為 “/search”，并包含三個基本的 GET 路由，每個路由都有其特定的用途：searchResource 和 searchAIResults。
第一個端點?/search/{query}?在根據提供的查詢檢索搜索結果方面起著關鍵作用。此過程涉及調用 service 模塊中的函數，然后該函數會編排并返回一個對象列表，這些對象封裝了所獲得的搜索結果的本質。
繼續介紹該路線，/search/summary/{query} 途徑旨在獲取與提供的查詢相關聯的搜索結果的摘要。這是通過調用 service 模塊中相應的函數來實現的，該函數最終返回一個封裝了搜索過程匯總結果的對象。
最后，/search/qa/{query}?路由側重于使用 LangChain 的 RAG 模式來檢索問答結果。這是通過調用 service 模塊中的某個函數來實現的。
服務層構成了我們主要業務邏輯的基礎。在這個特定用例的上下文中，服務層扮演著 LangChain 代碼存儲庫的重要角色，將它們定位為我們技術框架中的核心組件。
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 服務展示了數據庫交互、語言模型以及提示模板的集成，從而能夠根據用戶查詢提供摘要和問答等功能。起初，它定義了一個模板字符串，該模板用于創建與 RAG（檢索增強型生成模型）相關的問題和答案。此模板內嵌了上下文和問題提示的占位符。搜索服務包含以下三個功能：
第一個搜索服務函數：該函數接收查詢字符串作為輸入，隨后在數據庫中執行與查詢相似的資源向量搜索，并返回匹配資源的列表。函數簽名為?get_query(query: str) -> list[Resource]。
第二個函數 ：該函數生成與給定查詢相關聯的文檔的簡潔摘要。它首先通過向量搜索找到相關文檔，然后利用某個文檔鏈（在描述中被誤寫為 StuffDocumentChain，可能是指具體的文檔處理鏈或模塊）來總結這些文檔。函數簽名為 get_query_summary(query: str) -> str。
最后一個函數 ：該函數檢索與查詢相關的文檔，并基于這些文檔構建問題的答案。它運用全局提示來生成問題和答案，同時利用從向量搜索返回的文檔作為輸入。函數簽名為 get_qa_from_query(query: str) -> str。
最后，我們到達了數據層。盡管 GitHub 存儲庫包含 Weaviate 和 Cosmos DB for MongoDB 的代碼（但在此討論中僅涉及 MongoDB 的部分），本系列將專注于與 Cosmos DB 的連接。在數據層中，我們利用?init.py?文件中的模塊作為單例來處理與數據庫相關的連接，這是一種清晰的方法，盡管有時我也會考慮使用全局變量作為替代方案。除了?init.py?文件，search.py?文件還負責執行對 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 文件中加載環境變量。隨后，它聲明了兩個全局變量：一個表示 CosmosDB MongoDB 集合的 collection，另一個表示 CosmosDB 向量搜索的 vector_store。接著，使用 MongoDB 的連接字符串初始化 MongoClient，并通過該客戶端檢索數據庫和集合對象。之后，這些對象被傳遞到 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數據檢索代碼通過我們的單例訪問全局變量，當前包含兩個關鍵函數，分別涉及相似性搜索和數據轉換。
第一個函數接收一個?LangChain?對象作為輸入，并輸出一個?ResourceDocument?對象。該函數從提供的?Document?中抽取元數據屬性，例如?resource_id、page_id、標題、章節以及頁碼。此函數的目的是確保向量搜索返回的資源能夠在?LangChain?外部得到重用。函數簽名為?results_to_model(result: Document) -> Resource。
第二個函數利用指定的查詢來執行相似性搜索。它調用 vector_store 的某個方法來檢索與給定查詢相似的文檔，以及這些文檔各自的相似性分數。該函數根據閾值分數（0.75）對檢索到的文檔進行篩選，并利用 results_to_model 函數將篩選后的文檔轉換為 Resource 對象。最終，該函數返回一個元組，包含轉換后的 Resource 對象列表和原始的 Document 對象列表。函數簽名為 similarity_search(query: str) -> Tuple[List[Resource], List[Document]]。
恭喜您順利完成了 Web 框架的設置！這確實是一項艱巨的任務，但為第3部分中即將展開的工作奠定了堅實的基礎。
原文鏈接：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/