在當今數字化時代，企業和教育機構每天都會收到海量的咨詢問題。無論是客戶支持、銷售團隊的提問，還是內部員工的咨詢，手動回復這些問題不僅耗時費力，還容易出現回答不一致的情況。而基于人工智能的查詢解答系統，能夠快速、準確且高效地提供答案，極大地提升了工作效率和用戶體驗。

今天，我們就來聊聊如何利用LangChain、ChromaDB和CrewAI構建一個基于檢索增強生成（RAG）的智能查詢解答系統。這個系統不僅能自動處理各種問題，還能生成精準的回答，幫助企業和教育機構更好地服務用戶。

為什么我們需要AI驅動的查詢解答系統？

在傳統的業務流程中，手動回復客戶或學員的咨詢問題是一個極其低效的過程。客戶希望得到即時的回復，而企業則需要快速獲取準確的信息來做出決策。AI驅動的查詢解答系統通過自動化處理這些問題，不僅減輕了人工負擔，還能提供一致且高質量的回答。

在客戶服務領域，AI系統可以自動回復常見問題，提升客戶滿意度；在銷售和市場營銷中，它可以實時提供產品細節和客戶洞察；在金融、醫療、教育和電商等行業，AI驅動的查詢處理能夠確保操作順暢，提升用戶體驗。

深入了解RAG工作流程

在動手構建系統之前，我們先來了解一下檢索增強生成（RAG）系統是如何工作的。RAG架構主要分為三個關鍵階段：索引、檢索和生成。

1. 構建向量存儲（文檔處理與存儲）

系統首先需要處理和存儲相關文檔，以便能夠快速檢索。具體步驟如下：

• 文檔切分：將大型文檔切分為更小的文本塊，以便高效檢索。
• 嵌入模型：利用基于AI的嵌入模型將這些文本塊轉換為向量表示。
• 向量存儲：將向量化的數據索引并存儲在數據庫（如ChromaDB）中，以便快速查找。

2. 查詢處理與檢索

當用戶提交問題時，系統會先檢索相關數據，然后再生成回答。具體步驟如下：

• 用戶查詢輸入：用戶提交問題或請求。
• 向量化：利用嵌入模型將查詢轉換為數值向量。
• 搜索與檢索：系統在向量存儲中搜索最相關的文本塊并檢索出來。

3. 增強與回答生成

為了生成準確的回答，系統會將檢索到的數據與原始查詢結合。具體步驟如下：

• 增強查詢：將檢索到的文檔塊與原始查詢結合。
• LLM處理：利用大型語言模型（LLM）根據查詢和檢索到的上下文生成最終回答。
• 最終回答：系統向用戶提供一個準確且富有上下文的回答。

構建基于RAG的查詢解答系統

接下來，我們將通過一個實際案例，展示如何構建一個基于RAG的查詢解答系統。這個系統將高效地回答學員的問題，幫助他們更好地學習。

選擇合適的數據用于查詢解答

在構建RAG系統之前，最重要的就是數據。一個結構良好的知識庫是關鍵，因為回答的準確性和相關性完全依賴于數據的質量。以下是一些適合不同類型用途的數據：

• 客戶支持數據：常見問題解答（FAQ）、故障排除指南、產品手冊和過去的客戶互動記錄。
• 銷售與市場數據：產品目錄、價格詳情、競爭對手分析和客戶咨詢記錄。
• 內部知識庫：公司政策、培訓文檔和標準操作流程（SOP）。
• 財務與法律文件：合規指南、財務報告和監管政策。
• 用戶生成內容：論壇討論、聊天記錄和反饋表單，這些都能提供真實的用戶問題。

在我們的學員查詢解答系統中，我們嘗試了多種數據類型，最終發現使用課程視頻的字幕是最有效的方法。字幕提供了與學員問題直接相關的結構化和詳細內容，能夠快速生成相關答案。

構建查詢解答系統的架構

在動手編寫代碼之前，我們需要先規劃系統的架構。系統需要完成以下三個主要任務：

1. 從字幕文件（SRT）中提取并存儲課程內容。
2. 根據學員的查詢檢索相關的課程材料。
3. 利用AI驅動的代理生成結構化的回答。

為了實現這些功能，我們將系統分為三個組件：

• 字幕處理：從SRT文件中提取文本，處理并將其嵌入存儲到ChromaDB中。
• 檢索：根據學員的查詢搜索并檢索相關的課程材料。
• 查詢回答代理：利用CrewAI生成結構化且準確的回答。

實現步驟

現在，我們已經規劃好了系統的架構，接下來就是動手實現。

1. 導入必要的庫

構建AI驅動的學習支持系統，首先需要導入一些關鍵的庫：

import pysrt
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from crewai import Agent, Task, Crew
import pandas as pd
import ast
import os
import time
from tqdm import tqdm

這些庫的作用如下：

• pysrt：用于從SRT字幕文件中提取文本。
• RecursiveCharacterTextSplitter：將大段文本切分為更小的塊，以便更好地檢索。
• Document：表示結構化的文本文檔。
• OpenAIEmbeddings：將文本轉換為數值向量，用于相似性搜索。
• Chroma：將嵌入存儲在向量數據庫中，便于高效檢索。
• CrewAI（Agent、Task、Crew）：定義處理學員查詢的AI代理。
• pandas：以DataFrame形式處理結構化數據。
• ast：將基于字符串的數據結構解析為Python對象。
• os：提供系統級操作，如讀取環境變量。
• tqdm：在長時間運行的任務中顯示進度條。

2. 設置環境

為了使用OpenAI的API進行嵌入，我們需要加載API密鑰并配置模型設置。

步驟1：從本地文件中讀取API密鑰

with open('/home/janvi/Downloads/openai.txt', 'r') as file:
    openai_api_key = file.read()

步驟2：將API密鑰存儲為環境變量

os.environ['OPENAI_API_KEY'] = openai_api_key

步驟3：指定OpenAI模型

os.environ["OPENAI_MODEL_NAME"] = 'gpt-4o-mini'

通過這些配置，我們可以確保系統能夠高效地處理和存儲嵌入。

3. 提取并存儲字幕數據

字幕文件中包含了視頻講座的寶貴信息，是AI檢索系統中結構化內容的豐富來源。有效地提取和處理字幕數據，能夠讓我們在回答學員問題時快速檢索到相關信息。

步驟1：從SRT文件中提取文本

我們使用??pysrt??庫從SRT文件中提取文本，并將其組織成結構化的形式，以便進一步處理和存儲。

def extract_text_from_srt(srt_path):
    """從SRT字幕文件中提取文本"""
    subs = pysrt.open(srt_path)
    text = " ".join(sub.text for sub in subs)
    return text

由于課程可能包含多個字幕文件，我們需要系統地組織和迭代這些文件，以便無縫提取文本。

course_folders = {
    "深度學習入門（使用PyTorch）": "C:\M\Code\GAI\Learn_queries\Subtitle_Introduction_to_Deep_Learning_Using_Pytorch",
    "構建生產級RAG系統（使用LlamaIndex）": "C:\M\Code\GAI\Learn_queries\Subtitle of Building Production-Ready RAG systems using LlamaIndex",
    "LangChain入門（構建生成式AI應用與代理）": "C:\M\Code\GAI\Learn_queries\Subtitle_introduction_to_langchain_using_agentic_ai"
}

course_srt_files = {}

for course, folder_path in course_folders.items():
    srt_files = []
    for root, _, files in os.walk(folder_path):
        srt_files.extend(os.path.join(root, file) for file in files if file.endswith(".srt"))
    if srt_files:
        course_srt_files[course] = srt_files

這些提取的文本將成為我們AI驅動學習支持系統的基礎，使我們能夠進行高級檢索和查詢解答。

步驟2：將字幕存儲到ChromaDB

接下來，我們將課程字幕存儲到ChromaDB中，包括文本切分、嵌入生成、持久化存儲和成本估算。

（1）為ChromaDB設置持久化目錄

??persist_directory??是一個文件夾路徑，用于保存存儲的數據。這樣即使程序重新啟動，嵌入數據也能保留下來。

persist_directory = "./subtitles_db"

（2）將文本切分為更小的塊

大型文檔（如整個課程字幕）可能會超出嵌入的標記限制。為了處理這種情況，我們使用??RecursiveCharacterTextSplitter??將文本切分為更小的、有重疊的塊，以提高搜索精度。

text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)

每個塊的長度為1000個字符，為了在塊之間保留上下文，我們將前一個塊的200個字符包含在下一個塊中。這種重疊有助于保留重要細節，提高檢索精度。

（3）初始化OpenAI嵌入和ChromaDB向量存儲

我們需要將文本轉換為數值向量表示，以便進行相似性搜索。OpenAI的嵌入功能允許我們將課程內容編碼為可以高效搜索的格式。

embeddings = OpenAIEmbeddings(openai_api_key=openai_api_key)

這里，??OpenAIEmbeddings()??使用我們的OpenAI API密鑰初始化嵌入模型，確保每段文本都能轉換為高維向量表示。

（4）初始化ChromaDB

現在，我們將這些向量嵌入存儲到ChromaDB中。

vectorstore = Chroma(
    collection_name="course_materials",
    embedding_functinotallow=embeddings,
    persist_directory=persist_directory
)

??collection_name="course_materials"???在ChromaDB中創建了一個專門的集合，用于組織所有與課程相關的嵌入。??embedding_functinotallow=embeddings???指定了OpenAI嵌入用于將文本轉換為數值向量。??persist_directory=persist_directory??確保所有存儲的嵌入在程序重新啟動后仍然可用。

（5）估算存儲課程數據的成本

在將文檔添加到向量數據庫之前，估算標記使用成本是非常重要的。由于OpenAI按每1000個標記收費，我們需要提前估算成本，以便高效管理開支。

COST_PER_1K_TOKENS = 0.0001  # 每1000個標記的成本（使用'text-embedding-ada-002'模型）
TOKENS_PER_CHUNK_ESTIMATE = 750  # 每1000字符塊的估計標記數

total_tokens = 0
total_cost = 0

start_time = time.time()

??COST_PER_1K_TOKENS=0.0001???定義了使用OpenAI嵌入時每1000個標記的成本。??TOKENS_PER_CHUNK_ESTIMATE=750???估計每個1000字符塊包含約750個標記。??total_tokens???和??total_cost???變量用于跟蹤整個執行過程中處理的數據量和產生的成本。??start_time??變量記錄了開始時間，用于測量整個過程的耗時。

（6）檢查并添加課程到ChromaDB

我們希望避免重新處理已經存儲在向量數據庫中的課程。因此，我們先查詢ChromaDB，檢查課程是否已經存在。如果不存在，我們則提取并存儲其字幕數據。

for course, srt_list in course_srt_files.items():
    existing_docs = vectorstore._collection.get(where={"course": course})
    if not existing_docs['ids']:
        srt_texts = [extract_text_from_srt(srt) for srt in srt_list]
        course_text = "\n\n\n\n".join(srt_texts)
        doc = Document(page_cnotallow=course_text, metadata={"course": course})
        chunks = text_splitter.split_documents([doc])

字幕通過??extract_text_from_srt()???函數提取，多個字幕文件通過??\n\n\n\n???連接，以提高可讀性。創建了一個??Document???對象，存儲完整的字幕文本及其元數據。最后，使用??text_splitter.split_documents()??將文本切分為更小的塊，以便高效處理和檢索。

（7）估算標記使用量和成本

在將塊添加到ChromaDB之前，我們估算成本。

chunk_count = len(chunks)
batch_tokens = chunk_count * TOKENS_PER_CHUNK_ESTIMATE
batch_cost = (batch_tokens / 1000) * COST_PER_1K_TOKENS
total_tokens += batch_tokens
total_cost += batch_cost

??chunk_count???表示切分后的塊數量。??batch_tokens???根據塊數量估算總標記數。??batch_cost???計算當前課程的處理成本。??total_tokens???和??total_cost??累加每次處理的值，以跟蹤整體處理量和開支。

（8）將塊添加到ChromaDB

vectorstore.add_documents(chunks)
print(f"已添加課程：{course} (塊數：{chunk_count}, 成本：${batch_cost:.4f})")

處理后的塊被存儲到ChromaDB中，以便高效檢索。程序會顯示添加的塊數和估算的處理成本。

如果課程已經存在，則會顯示以下信息：

print(f"課程已存在：{course}")

一旦所有課程處理完成，我們計算并顯示最終結果。

end_time = time.time()

print(f"\n課程嵌入更新完成！??")
print(f"總處理塊數：{total_tokens // TOKENS_PER_CHUNK_ESTIMATE}")
print(f"估算總標記數：{total_tokens}")
print(f"估算總成本：${total_cost:.4f}")
print(f"總耗時：{end_time - start_time:.2f}秒")

??end_time - start_time??計算總處理時間。系統會顯示處理的塊數、估算的標記使用量、總成本以及整個嵌入過程的總結。

4. 查詢并回答學員問題

一旦字幕存儲到ChromaDB中，系統需要一種方式來檢索相關內容，以便在學員提交問題時提供答案。這個檢索過程通過相似性搜索實現，它能夠識別與輸入問題最相關的存儲文本段。

工作原理

• 查詢輸入：學員提交與課程相關的問題。
• 按課程過濾：系統確保檢索僅限于相關課程材料。
• ChromaDB中的相似性搜索：將查詢轉換為嵌入，ChromaDB檢索最相似的存儲文本塊。
• 返回頂部結果：系統選擇最相關的三個文本段。
• 格式化輸出：檢索到的文本被格式化并呈現為進一步處理的上下文。

def retrieve_course_materials(query: str, course):
    """按課程名稱檢索課程材料"""
    filter_dict = {"course": course}
    results = vectorstore.similarity_search(query, k=3, filter=filter_dict)
    return "\n\n".join([doc.page_content for doc in results])

例如：

course_name = "深度學習入門（使用PyTorch）"
question = "什么是梯度下降？"
context = retrieve_course_materials(query=question, course=course_name)
print(context)

從輸出中可以看到，ChromaDB通過相似性搜索，根據課程名稱和問題檢索到最相關的信息。

為什么使用相似性搜索？

• 語義理解：與關鍵詞搜索不同，相似性搜索能夠找到與查詢語義相關的文本。
• 高效檢索：系統無需掃描整個文檔，只需檢索最相關的部分。
• 提升答案質量：通過按課程過濾并按相關性排序，學員能夠獲得高度針對性的內容。

這種機制確保學員提交問題時，能夠從存儲的課程材料中獲得相關且上下文準確的信息。

5. 實現AI查詢回答代理

檢索到相關課程材料后，下一步是利用AI驅動的代理生成有意義的回答。我們使用CrewAI定義一個智能代理，負責分析查詢并生成結構化的回答。

步驟1：定義代理

查詢回答代理通過清晰的角色和背景故事來指導其行為，以便更好地回答學員的問題。

query_answer_agent = Agent(
    role="學習支持專家",
    goal="您需要為學員提供最準確的回答",
    backstory="""
    您是一家專注于數據科學、機器學習和生成式AI的在線教育公司學員查詢解答部門的負責人。您負責回答學員關于課程內容、作業、技術問題和行政問題的咨詢。您禮貌、圓滑，并且對可以改進的地方負有責任感。
    """,
    verbose=False
)

在代碼塊中，我們首先定義了代理的角色為“學習支持專家”，因為它充當虛擬助教的角色，回答學員的問題。然后，我們定義了目標，確保代理在回答時優先考慮準確性和清晰性。最后，我們將??verbose???設置為??False??，這樣在不需要調試時，執行過程將保持安靜。這種清晰定義的代理角色確保回答既有幫助性，又結構化，且符合教育平臺的語氣。

步驟2：定義任務

定義了代理之后，我們需要為其分配任務。

query_answering_task = Task(
    descriptinotallow="""
    盡您所能回答學員的問題。盡量保持回答簡潔，不超過100個單詞。
    這是問題：{query}

    這是從課程字幕中提取的相關內容，僅在需要時使用：{relevant_content}。
    由于這些內容是從課程字幕中提取的，可能存在拼寫錯誤，請在回答中糾正這些錯誤。

    這是與學員之前的對話記錄：{thread}。
    在對話中，以“學員”開頭的是學員的問題，以“支持”開頭的是您的回答。請根據之前的對話適當調整您的回答。

    這是學員的全名：{learner_name}。
    如果不確定學員的名字，直接用“嗨”開頭。
    在回答的結尾添加一些適當的、鼓勵性的安慰語句，例如“希望您覺得有幫助”、“希望這些信息有用。繼續努力！”、“很高興能幫到您！隨時聯系我。”等。

    如果您不確定答案，請注明：“抱歉，我不確定這個問題的答案，我會稍后回復您。”
    """,
    expected_output="簡潔準確的回答",
    agent=query_answer_agent
)

接下來，我們來分解分配給AI的任務。處理學員的查詢時，??{query}???代表學員的問題。回答應簡潔（不超過100個單詞）且準確。如果需要使用課程內容，??{relevant_content}??是從存儲在ChromaDB中的字幕中提取的，AI必須在回答中糾正任何拼寫錯誤。

如果存在之前的對話，??{thread}???有助于保持連貫性。學員的問題以“學員”開頭，而之前的回答以“支持”開頭，這使得代理能夠提供與上下文相關的回答。通過??{learner_name}??實現個性化——代理會用學員的名字稱呼他們，如果不確定名字，就簡單地用“嗨”開頭。

為了使回答更具吸引力，AI會在結尾添加一句積極的結束語，比如“希望您覺得有幫助！”或者“隨時聯系我。”如果AI不確定答案，它會明確說明：“抱歉，我不確定這個問題的答案，我會稍后回復您。”這種方法確保了回答的禮貌性、清晰性和結構化，提升了學員的參與度和信任感。

步驟3：初始化CrewAI實例

現在我們已經定義了代理和任務，接下來初始化CrewAI，它能夠動態處理學員的查詢。

response_crew = Crew(
    agents=[query_answer_agent],
    tasks=[query_answering_task],
    verbose=False
)

??agents=[query_answer_agent]???將“學習支持專家”代理添加到團隊中。??tasks=[query_answering_task]???將查詢回答任務分配給這個代理。設置??verbose=False??可以保持輸出簡潔，除非需要調試。CrewAI能夠動態處理多個學員的查詢，使系統具有可擴展性和高效性，能夠動態處理查詢。

步驟4：為多個學員的查詢生成回答

設置好AI代理后，我們需要動態處理存儲在結構化數據集中的學員查詢。

以下代碼處理存儲在CSV文件中的學員查詢，并使用AI代理生成回答。它首先加載包含學員查詢、課程詳情和對話線程的數據集。??reply_to_query??函數提取相關細節，如學員姓名、課程名稱和當前查詢。如果存在之前的對話，它會提取出來以提供上下文。如果查詢包含圖片，則會跳過。然后，它從ChromaDB中檢索相關的課程材料，并將查詢、相關內容和之前的對話發送給AI代理，以生成結構化的回答。

df = pd.read_csv('C:\M\Code\GAI\Learn_queries\filtered_data_top3_courses.csv')

def reply_to_query(df, index=1):
    learner_name = df.iloc[index]["thread_starter"]
    course_name = df.iloc[index]["course"]
    if df.iloc[index]['number_of_replies'] > 1:
        thread = ast.literal_eval(df.iloc[index]["modified_thread"])
    else:
        thread = []
    question = df.iloc[index]["current_query"]
    if df.iloc[index]['has_image'] == True:
        return" "
    context = retrieve_course_materials(query=question, course=course_name)
    response_result = response_crew.kickoff(inputs={"query": question, "relevant_content": context, "thread": thread, "learner_name": learner_name})
    print('Q: ', question)
    print('\n')
    print('A: ', response_result)
    print('\n\n')

測試該函數時，我們為一個查詢（??index=1??）執行它：

reply_to_query(df, index=1)

從輸出中可以看到，它能夠正常工作，僅針對一個索引生成回答。

現在，我們通過所有查詢進行迭代，處理每一個查詢，同時處理可能出現的錯誤。這確保了查詢解答過程的高效自動化，能夠動態處理多個學員的查詢。

for i in range(len(df)):
    try:
        reply_to_query(df, index=i)
    except:
        print("索引號出錯：", i)
        continue

為什么這一步很重要？

• 自動化查詢處理：系統能夠高效處理多個學員的查詢。
• 確保上下文相關性：回答基于檢索到的課程材料和之前的對話生成。
• 可擴展性：該方法允許AI代理動態處理并回答數千個查詢。
• 提升學習支持體驗：學員能夠收到個性化且數據驅動的回答。

這一步確保每個學員的查詢都能被分析、結合上下文并有效回答，從而提升整體學習體驗。

輸出示例

從輸出中可以看到，回答查詢的過程已經實現自動化，首先是問題，然后是回答。

未來改進方向

為了進一步提升基于RAG的查詢解答系統，我們可以考慮以下改進方向：

1. 常見問題及其解答：在查詢解答框架內實現一個結構化的FAQ系統，能夠即時回答常見問題，減少對人工支持的依賴。
2. 圖像處理能力：增加分析和提取圖像（如截圖、圖表或掃描文檔）相關信息的能力，將使系統在教育和客戶支持領域更具 versatility。
3. 改進圖像列布爾值：完善圖像列檢測的邏輯，更準確地識別和處理基于圖像的查詢。
4. 語義切塊和不同的切塊技術：嘗試不同的切塊策略，如語義切塊、固定長度分割和混合方法，可以提高檢索精度和回答的上下文理解能力。

總結

這個基于RAG的查詢解答系統利用LangChain、ChromaDB和CrewAI，高效地實現了學員支持的自動化。它從課程字幕中提取文本，將其嵌入存儲到ChromaDB中，并通過相似性搜索檢索相關內容。CrewAI代理處理查詢，參考之前的對話，并生成結構化的回答，確保回答的準確性和個性化。

該系統提升了可擴展性、檢索效率和回答質量，使自主學習更加互動。未來的改進方向包括多模態支持、更好的檢索優化和增強的回答生成。通過自動化查詢解答，這個系統簡化了學習支持流程，為學員提供了更快、更具上下文意識的回答，提升了整體參與度。

希望這篇文章能幫助你更好地理解如何構建一個基于RAG的智能查詢解答系統。如果你有任何問題或想法，歡迎在評論區留言，我們一起探討！

目前市面上也有很多現成管理平臺，如Dify、RagFlow、AnythinLLM等，通過低代碼的方式就可以直接實現自主搭建智能問答系統。感興趣的都可以去看看哈！

本文轉載自公眾號Halo咯咯    作者：基咯咯

原文鏈接：??https://mp.weixin.qq.com/s/vQiL3RjAU7--QE-EyXrarg??