全網(wǎng)最詳細的Spring入門教程
你可能也喜歡這些API文章!
現(xiàn)在，讓我們安裝必要的軟件包。創(chuàng)建包含以下內(nèi)容的文件：requirements.txt
langchain
langchain-openai
langchain-core
langchain_community
docx2txt
pypdf
langchain_chroma
python-multipart
fastapi
uvicorn使用 pip 安裝這些軟件包：
pip install -r requirements.txt設(shè)置好環(huán)境后，我們就可以開始構(gòu)建生產(chǎn)就緒的 RAG 聊天機器人 API。在下一節(jié)中，我們將深入研究項目結(jié)構(gòu)并開始實現(xiàn)我們的 FastAPI 應(yīng)用程序。
當然！讓我們繼續(xù)下一部分，在那里我們將討論項目結(jié)構(gòu)概述。本節(jié)將幫助讀者了解我們?nèi)绾谓M織代碼以提高可維護性和可伸縮性。（當然！讓我們進入下一部分，我們將討論項目結(jié)構(gòu)概述。這一節(jié)將幫助讀者了解我們?nèi)绾谓M織代碼以實現(xiàn)更好的可維護性和可擴展性。）
在從原型向生產(chǎn)就緒型應(yīng)用過渡的過程中，合理的代碼組織變得至關(guān)重要。結(jié)構(gòu)良好的項目更易于維護、測試和擴展。對于我們的RAG聊天機器人API，我們將采用模塊化結(jié)構(gòu)，以分離關(guān)注點并促進代碼重用。
以下是我們項目結(jié)構(gòu)的概述：
rag-fastapi-project/
│
├── main.py
├── chroma_utils.py
├── db_utils.py
├── langchain_utils.py
├── pydantic_models.py
├── requirements.txt
└── chroma_db/ (directory for Chroma persistence)讓我們分解每個文件的用途：
此結(jié)構(gòu)遵循 FastAPI 應(yīng)用程序的最佳實踐，并為構(gòu)建我們的 RAG 聊天機器人 API 提供了堅實的基礎(chǔ)。在本教程中，我們將深入研究這些文件，解釋它們的內(nèi)容以及它們?nèi)绾螀f(xié)同工作以創(chuàng)建我們的生產(chǎn)就緒系統(tǒng)。
main.py?文件是我們 FastAPI 應(yīng)用程序的核心。它定義了我們的 API 端點并編排了我們系統(tǒng)的不同組件之間的交互。讓我們分解此文件的關(guān)鍵元素：（main.py文件是我們FastAPI應(yīng)用的核心。它定義了我們的API端點，并協(xié)調(diào)我們系統(tǒng)不同組件之間的交互。讓我們分解此文件的關(guān)鍵元素：）
from fastapi import FastAPI, File, UploadFile, HTTPException
from pydantic_models import QueryInput, QueryResponse, DocumentInfo, DeleteFileRequest
from langchain_utils import get_rag_chain
from db_utils import insert_application_logs, get_chat_history, get_all_documents, insert_document_record, delete_document_record
from chroma_utils import index_document_to_chroma, delete_doc_from_chroma
import os
import uuid
import logging
import shutil
# Set up logging
logging.basicConfig(filename='app.log', level=logging.INFO)
# Initialize FastAPI app
app = FastAPI()在這里，我們導(dǎo)入必要的模塊并初始化我們的 FastAPI 應(yīng)用程序。我們還設(shè)置了基本日志記錄來跟蹤應(yīng)用程序中的重要事件。
現(xiàn)在，讓我們看看我們的主要 API 端點：
@app.post("/chat", response_model=QueryResponse)
def chat(query_input: QueryInput):
session_id = query_input.session_id or str(uuid.uuid4())
logging.info(f"Session ID: {session_id}, User Query: {query_input.question}, Model: {query_input.model.value}")
chat_history = get_chat_history(session_id)
rag_chain = get_rag_chain(query_input.model.value)
answer = rag_chain.invoke({
"input": query_input.question,
"chat_history": chat_history
})['answer']
insert_application_logs(session_id, query_input.question, answer, query_input.model.value)
logging.info(f"Session ID: {session_id}, AI Response: {answer}")
return QueryResponse(answer=answer, session_id=session_id, model=query_input.model)此終端節(jié)點處理聊天交互。如果未提供，它會生成會話 ID，檢索聊天記錄，調(diào)用 RAG 鏈以生成響應(yīng)，記錄交互并返回響應(yīng)。（此端點處理聊天交互。如果未提供會話ID，則生成一個會話ID，檢索聊天歷史，調(diào)用RAG鏈生成響應(yīng)，記錄交互，并返回響應(yīng)。）
@app.post("/upload-doc")
def upload_and_index_document(file: UploadFile = File(...)):
allowed_extensions = ['.pdf', '.docx', '.html']
file_extension = os.path.splitext(file.filename)[1].lower()
if file_extension not in allowed_extensions:
raise HTTPException(status_code=400, detail=f"Unsupported file type. Allowed types are: {', '.join(allowed_extensions)}")
temp_file_path = f"temp_{file.filename}"
try:
# Save the uploaded file to a temporary file
with open(temp_file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
file_id = insert_document_record(file.filename)
success = index_document_to_chroma(temp_file_path, file_id)
if success:
return {"message": f"File {file.filename} has been successfully uploaded and indexed.", "file_id": file_id}
else:
delete_document_record(file_id)
raise HTTPException(status_code=500, detail=f"Failed to index {file.filename}.")
finally:
if os.path.exists(temp_file_path):
os.remove(temp_file_path)此終端節(jié)點處理文檔上傳。它會檢查允許的文件類型，臨時保存文件，在 Chroma 中為其編制索引，并更新數(shù)據(jù)庫中的文檔記錄。
@app.get("/list-docs", response_model=list[DocumentInfo])
def list_documents():
return get_all_documents()這個簡單的端點返回所有索引文檔的列表。
@app.post("/delete-doc")
def delete_document(request: DeleteFileRequest):
chroma_delete_success = delete_doc_from_chroma(request.file_id)
if chroma_delete_success:
db_delete_success = delete_document_record(request.file_id)
if db_delete_success:
return {"message": f"Successfully deleted document with file_id {request.file_id} from the system."}
else:
return {"error": f"Deleted from Chroma but failed to delete document with file_id {request.file_id} from the database."}
else:
return {"error": f"Failed to delete document with file_id {request.file_id} from Chroma."}此端點處理文檔刪除，從 Chroma 和數(shù)據(jù)庫中刪除文檔。
Pydantic是一個數(shù)據(jù)驗證庫，它使用Python類型注解來定義數(shù)據(jù)模式。在我們的FastAPI應(yīng)用中，我們使用Pydantic模型來定義請求和響應(yīng)數(shù)據(jù)的結(jié)構(gòu)。讓我們分解在models.py中定義的模型：
from pydantic import BaseModel, Field
from enum import Enum
from datetime import datetime
class ModelName(str, Enum):
GPT4_O = "gpt-4o"
GPT4_O_MINI = "gpt-4o-mini"
class QueryInput(BaseModel):
question: str
session_id: str = Field(default=None)
model: ModelName = Field(default=ModelName.GPT4_O_MINI)
class QueryResponse(BaseModel):
answer: str
session_id: str
model: ModelName
class DocumentInfo(BaseModel):
id: int
filename: str
upload_timestamp: datetime
class DeleteFileRequest(BaseModel):
file_id: int讓我們看看每個模型及其用途：
ModelName（枚舉）：
QueryInput:
question：用戶的問題 （必填）。session_id：可選會話 ID。如果未提供，將生成一個。model：要使用的語言模型，默認為 GPT4_O_MINI。QueryResponse:
answer：生成的答案。session_id：會話 ID（用于繼續(xù)對話）。model：用于生成響應(yīng)的模型。DocumentInfo:
id：文檔的唯一標識符。filename：上傳文件的名稱。upload_timestamp：上傳文檔并為其編制索引的時間。DeleteFileRequest:
file_id：要刪除的文檔的 ID。在我們的?main.py?中，我們使用這些模型來定義請求和響應(yīng)數(shù)據(jù)的形狀。例如：
@app.post("/chat", response_model=QueryResponse)
def chat(query_input: QueryInput):
# Function implementation在這里，F(xiàn)astAPI 用于驗證傳入的請求數(shù)據(jù)，以及驗證和序列化響應(yīng)。這可確保我們的 API 行為一致，并在提供無效數(shù)據(jù)時提供清晰的錯誤消息。（在這里，F(xiàn)astAPI使用Pydantic模型來驗證傳入的請求數(shù)據(jù)，并驗證和序列化響應(yīng)。這確保了我們的API行為一致，并在提供無效數(shù)據(jù)時提供清晰的錯誤消息，QueryInput和QueryResponse）QueryInputQueryResponse
隨著我們的API不斷發(fā)展，我們可以輕松地擴展這些模型。例如，如果我們想為我們的文檔信息添加更多元數(shù)據(jù)，只需向模型中添加字段即可：DocumentInfo
class DocumentInfo(BaseModel):
id: int
filename: str
upload_timestamp: datetime
file_size: int # New field
content_type: str # New fieldFastAPI 和 Pydantic 將自動處理新字段，提供驗證和文檔，而無需對我們的端點邏輯進行任何更改。
通過使用 Pydantic 模型，我們?yōu)?API 創(chuàng)建了強大的基礎(chǔ)，確保了數(shù)據(jù)的完整性，并為我們的端點提供了清晰的契約。這種方法顯著減少了我們需要編寫的手動驗證代碼的數(shù)量，并有助于防止與錯誤數(shù)據(jù)處理相關(guān)的漏洞。
utils.py?文件包含用于與我們的 SQLite 數(shù)據(jù)庫交互的函數(shù)。我們使用 SQLite 是因為它的簡單性和易于設(shè)置，使其非常適合原型設(shè)計和中小型應(yīng)用程序。讓我們分解此文件的關(guān)鍵組件：db_utils.py文件包含與SQLite數(shù)據(jù)庫交互的函數(shù)。我們選擇SQLite是因為其簡單性和易于設(shè)置的特性，使其非常適合原型設(shè)計和中小型應(yīng)用程序。讓我們分解該文件的關(guān)鍵組件：）
import sqlite3
from datetime import datetime
DB_NAME = "rag_app.db"
def get_db_connection():
conn = sqlite3.connect(DB_NAME)
conn.row_factory = sqlite3.Row
return conn我們首先導(dǎo)入必要的模塊并定義數(shù)據(jù)庫名稱。get_db_connection()函數(shù)用于創(chuàng)建與SQLite數(shù)據(jù)庫的連接，并將行工廠設(shè)置為sqlite3.Row，以便更容易地訪問數(shù)據(jù)。
def create_application_logs():
conn = get_db_connection()
conn.execute('''CREATE TABLE IF NOT EXISTS application_logs
(id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT,
user_query TEXT,
gpt_response TEXT,
model TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)''')
conn.close()
def create_document_store():
conn = get_db_connection()
conn.execute('''CREATE TABLE IF NOT EXISTS document_store
(id INTEGER PRIMARY KEY AUTOINCREMENT,
filename TEXT,
upload_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP)''')
conn.close()這些函數(shù)創(chuàng)建了我們的兩個主要表：
application_logs：存儲聊天歷史記錄和模型響應(yīng)。document_store：跟蹤上傳的文檔。def insert_application_logs(session_id, user_query, gpt_response, model):
conn = get_db_connection()
conn.execute('INSERT INTO application_logs (session_id, user_query, gpt_response, model) VALUES (?, ?, ?, ?)',
(session_id, user_query, gpt_response, model))
conn.commit()
conn.close()
def get_chat_history(session_id):
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute('SELECT user_query, gpt_response FROM application_logs WHERE session_id = ? ORDER BY created_at', (session_id,))
messages = []
for row in cursor.fetchall():
messages.extend([
{"role": "human", "content": row['user_query']},
{"role": "ai", "content": row['gpt_response']}
])
conn.close()
return messages這些函數(shù)處理插入新的聊天日志和檢索給定會話的聊天歷史記錄。聊天記錄的格式設(shè)置為我們的 RAG 系統(tǒng)易于使用。
def insert_document_record(filename):
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute('INSERT INTO document_store (filename) VALUES (?)', (filename,))
file_id = cursor.lastrowid
conn.commit()
conn.close()
return file_id
def delete_document_record(file_id):
conn = get_db_connection()
conn.execute('DELETE FROM document_store WHERE id = ?', (file_id,))
conn.commit()
conn.close()
return True
def get_all_documents():
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute('SELECT id, filename, upload_timestamp FROM document_store ORDER BY upload_timestamp DESC')
documents = cursor.fetchall()
conn.close()
return [dict(doc) for doc in documents]這些函數(shù)處理文檔記錄的 CRUD 操作：
在文件末尾，我們初始化我們的數(shù)據(jù)庫表：
# Initialize the database tables
create_application_logs()
create_document_store()這可確保在應(yīng)用程序啟動時創(chuàng)建我們的表（如果它們尚不存在）。（這確保了當應(yīng)用程序啟動時（如果表尚不存在），我們的表會被創(chuàng)建。）
通過將數(shù)據(jù)庫操作集中在utils.py中，我們保持了關(guān)注點的清晰分離。我們的主要應(yīng)用程序邏輯無需擔(dān)心數(shù)據(jù)庫交互的細節(jié)，從而使代碼更加模塊化且易于維護。
在生產(chǎn)環(huán)境中，您可以考慮使用像 SQLAlchemy 這樣的 ORM（對象關(guān)系映射）庫來實現(xiàn)更復(fù)雜的數(shù)據(jù)庫操作和更好的可擴展性。但是，對于我們當前的需求，這種簡單的 SQLite 實現(xiàn)效果很好。（在生產(chǎn)環(huán)境中，對于更復(fù)雜的數(shù)據(jù)庫操作和更好的可擴展性，您可能會考慮使用像SQLAlchemy這樣的ORM（對象關(guān)系映射）庫。然而，對于我們當前的需求，這種直接的SQLite實現(xiàn)已經(jīng)足夠好了。）
utils.py文件包含與Chroma向量存儲交互的函數(shù)，這對于我們的檢索增強型生成（RAG）系統(tǒng)的檢索功能至關(guān)重要。讓我們分解該文件的關(guān)鍵組件：
from langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader, UnstructuredHTMLLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from typing import List
from langchain_core.documents import Document
import os
# Initialize text splitter and embedding function
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200, length_function=len)
embedding_function = OpenAIEmbeddings()
# Initialize Chroma vector store
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embedding_function)在這里，我們導(dǎo)入了必要的模塊，并初始化了文本分割器、嵌入函數(shù)和Chroma向量存儲。RecursiveCharacterTextSplitter用于將文檔分割成可管理的塊，而OpenAIEmbeddings則提供了我們的文檔嵌入函數(shù)。
def load_and_split_document(file_path: str) -> List[Document]:
if file_path.endswith('.pdf'):
loader = PyPDFLoader(file_path)
elif file_path.endswith('.docx'):
loader = Docx2txtLoader(file_path)
elif file_path.endswith('.html'):
loader = UnstructuredHTMLLoader(file_path)
else:
raise ValueError(f"Unsupported file type: {file_path}")
documents = loader.load()
return text_splitter.split_documents(documents)此函數(shù)處理加載不同的文檔類型（PDF、DOCX、HTML）并將它們拆分為塊。它根據(jù)文件擴展名使用適當?shù)募虞d器，然后應(yīng)用我們的文本拆分器來創(chuàng)建可管理的文檔塊。
def index_document_to_chroma(file_path: str, file_id: int) -> bool:
try:
splits = load_and_split_document(file_path)
# Add metadata to each split
for split in splits:
split.metadata['file_id'] = file_id
vectorstore.add_documents(splits)
return True
except Exception as e:
print(f"Error indexing document: {e}")
return False此函數(shù)采用文件路徑和文件 ID，加載并拆分文檔，將元數(shù)據(jù)（文件 ID）添加到每個拆分中，然后將這些文檔塊添加到我們的 Chroma 矢量存儲中。元數(shù)據(jù)允許我們將 vector store 條目鏈接回我們的數(shù)據(jù)庫記錄。(此函數(shù)接受文件路徑和文件ID，加載并分割文檔，為每個分割添加元數(shù)據(jù)（文件ID），然后將這些文檔塊添加到我們的Chroma向量存儲中。元數(shù)據(jù)使我們能夠?qū)⑾蛄看鎯l目鏈接回我們的數(shù)據(jù)庫記錄。)
def delete_doc_from_chroma(file_id: int):
try:
docs = vectorstore.get(where={"file_id": file_id})
print(f"Found {len(docs['ids'])} document chunks for file_id {file_id}")
vectorstore._collection.delete(where={"file_id": file_id})
print(f"Deleted all documents with file_id {file_id}")
return True
except Exception as e:
print(f"Error deleting document with file_id {file_id} from Chroma: {str(e)}")
return False此函數(shù)從 Chroma 矢量存儲中刪除與給定文件 ID 關(guān)聯(lián)的所有文檔塊。它首先檢索文檔以確認其存在，然后執(zhí)行刪除。
盡管在utils.py文件中沒有明確顯示，但Chroma向量存儲對于我們的RAG系統(tǒng)的檢索步驟至關(guān)重要。在utils.py中，我們使用此向量存儲來創(chuàng)建一個檢索器：
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})然后，在我們的 RAG 鏈中使用此檢索器，以根據(jù)用戶的查詢獲取相關(guān)的文檔塊。
通過將我們的向量存儲操作集中在utils.py中，我們保持了關(guān)注點的清晰分離，并在未來需要時更容易替換或升級我們的向量存儲實現(xiàn)。
utils.py文件是我們使用LangChain實現(xiàn)檢索增強型生成（RAG）系統(tǒng)核心的地方。此文件設(shè)置了語言模型、檢索器和RAG鏈。讓我們分解其關(guān)鍵組件：
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from typing import List
from langchain_core.documents import Document
import os
from chroma_utils import vectorstore
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})
output_parser = StrOutputParser()在這里，我們導(dǎo)入必要的 LangChain 組件，并使用我們之前創(chuàng)建的 Chroma vectorstore 設(shè)置我們的檢索器。我們還初始化了一個字符串輸出解析器，用于處理語言模型的輸出。
contextualize_q_system_prompt = (
"Given a chat history and the latest user question "
"which might reference context in the chat history, "
"formulate a standalone question which can be understood "
"without the chat history. Do NOT answer the question, "
"just reformulate it if needed and otherwise return it as is."
)
contextualize_q_prompt = ChatPromptTemplate.from_messages([
("system", contextualize_q_system_prompt),
MessagesPlaceholder("chat_history"),
("human", "{input}"),
])
qa_prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful AI assistant. Use the following context to answer the user's question."),
("system", "Context: {context}"),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}")
])我們定義了兩個主要提示：
contextualize_q_prompt：用于根據(jù)聊天記錄重新構(gòu)建用戶的問題。qa_prompt：用于根據(jù)檢索到的上下文和聊天歷史記錄生成最終答案。def get_rag_chain(model="gpt-4o-mini"):
llm = ChatOpenAI(model=model)
history_aware_retriever = create_history_aware_retriever(llm, retriever, contextualize_q_prompt)
question_answer_chain = create_stuff_documents_chain(llm, qa_prompt)
rag_chain = create_retrieval_chain(history_aware_retriever, question_answer_chain)
return rag_chain此函數(shù)創(chuàng)建我們的 RAG 鏈：
ChatOpenAI在我們的main.py中，我們在聊天端點使用了這個RAG鏈：
@app.post("/chat", response_model=QueryResponse)
def chat(query_input: QueryInput):
# ... (other code)
rag_chain = get_rag_chain(query_input.model.value)
answer = rag_chain.invoke({
"input": query_input.question,
"chat_history": chat_history
})['answer']
# ... (rest of the function)這顯示了如何使用用戶指定的模型實例化 RAG 鏈，以及如何使用用戶的問題和聊天記錄調(diào)用。
通過將我們的 LangChain 邏輯集中在?(通過將LangChain邏輯集中在utils.py中，我們保持了關(guān)注點的清晰分離，并使得未來修改或擴展RAG系統(tǒng)變得更加容易。這種模塊化方法使我們能夠輕松嘗試不同的模型、檢索器或鏈結(jié)構(gòu)，而不會影響應(yīng)用程序的其他部分。)utils.py?中，我們保持了清晰的關(guān)注點分離，并使得將來更容易修改或擴展我們的 RAG 系統(tǒng)。這種模塊化方法使我們能夠輕松地試驗不同的模型、檢索器或鏈結(jié)構(gòu)，而不會影響應(yīng)用程序的其余部分。langchain_
在本教程中，我們逐步了解了如何使用FastAPI和LangChain構(gòu)建一個生產(chǎn)就緒的檢索增強生成（RAG）聊天機器人。讓我們回顧一下我們所完成的工作，并討論一些關(guān)鍵收獲和可能的下一步行動。
此架構(gòu)允許可在生產(chǎn)環(huán)境中部署可擴展、可維護和可擴展的 RAG 系統(tǒng)。
構(gòu)建生產(chǎn)就緒型 RAG 聊天機器人涉及的不僅僅是將語言模型連接到文檔存儲。它需要仔細考慮數(shù)據(jù)流、錯誤處理、可擴展性和用戶體驗。我們構(gòu)建的系統(tǒng)提供了堅實的基礎(chǔ)，可以進行調(diào)整和擴展以滿足特定的業(yè)務(wù)需求。（我們所構(gòu)建的系統(tǒng)提供了一個堅實的基礎(chǔ)，可以根據(jù)特定的業(yè)務(wù)需求進行適應(yīng)和擴展。）
隨著AI和自然語言處理技術(shù)的不斷發(fā)展，像這樣的系統(tǒng)對于創(chuàng)建智能的、支持上下文感知的應(yīng)用將變得越來越重要。通過了解RAG系統(tǒng)的原理和組件，您將在自己的項目中構(gòu)建和改進這項技術(shù)方面做好準備。
請記住，成功的RAG系統(tǒng)的關(guān)鍵不僅在于各個組件本身，還在于它們?nèi)绾螀f(xié)同工作以創(chuàng)建無縫、智能的交互。基于實際使用的持續(xù)測試、監(jiān)控和精煉對于確保您的RAG聊天機器人的長期成功和有效性至關(guān)重要。
為了幫助您進一步了解和實施這個 RAG 聊天機器人系統(tǒng)，我準備了一些額外的資源：
在 FutureSmart AI，我們專注于根據(jù)您的特定需求構(gòu)建定制的自然語言處理 （NLP） 解決方案。我們的專業(yè)知識不僅限于 RAG 系統(tǒng)，還包括：
我們已經(jīng)成功地為各個行業(yè)實施了這些技術(shù)，幫助企業(yè)利用 AI 的強大功能來增強其運營和用戶體驗。
無論您是希望實施像我們在本教程中構(gòu)建的 RAG 系統(tǒng)，還是有更具體的 NLP 需求，我們 FutureSmart AI 的團隊都可以幫助您將 AI 愿望變?yōu)楝F(xiàn)實。（無論您是想實施像本教程中構(gòu)建的RAG系統(tǒng)，還是有更具體的NLP需求，F(xiàn)utureSmart AI的團隊都在這里幫助您將AI愿景變?yōu)楝F(xiàn)實。）
原文鏈接：https://blog.futuresmart.ai/building-a-production-ready-rag-chatbot-with-fastapi-and-langchain
微信截圖_17439935106814-1.webp)
