LangChain RAG 學習筆記：從文檔加載到問答服務

我在先前的隨筆中分享過用Dify低代碼平臺來實現問答系統，也有幾篇隨筆是通過不同的方式來訪問大模型。本篇將使用LangChain來做對應的實現。相關代碼主要是通過Trae，它可以幫助你快速的了解了基本使用 LangChain 構建 RAG的方法，包括從文檔加載、向量存儲到問答接口實現，整個過程涉及多個關鍵環節。
雖然(ran)借助大模型以及Trae，給我(wo)們(men)提供了另外(wai)一種生(sheng)成(cheng)代(dai)(dai)碼和(he)學習代(dai)(dai)碼的(de)(de)(de)(de)方(fang)式(shi)，但(dan)其目前還是需要人工來(lai)參與的(de)(de)(de)(de)，尤其是版本的(de)(de)(de)(de)變化(hua)導(dao)致引入的(de)(de)(de)(de)包和(he)接(jie)口的(de)(de)(de)(de)調用方(fang)式(shi)都(dou)發(fa)生(sheng)了很多變化(hua)，所以這就需要一個(ge)根據生(sheng)成(cheng)的(de)(de)(de)(de)代(dai)(dai)碼不(bu)斷的(de)(de)(de)(de)去調試(shi)和(he)修正。本文里貼出的(de)(de)(de)(de)代(dai)(dai)碼也是經歷過這個(ge)過程之后總結下來(lai)的(de)(de)(de)(de)。

RAG 系統整體架構

首先回憶一下RAG 系統的核心思想，是將用戶查詢與知識庫中的相關信息進行匹配，再結合大語言模型生成準確回答。
這里我將一(yi)套 RAG 系(xi)統通(tong)分成以(yi)下幾(ji)個模塊：

1. 文檔加載與處理
2. 文本分割與嵌入
3. 向量存儲管理
4. 檢索功能實現
5. 問答生成服務
6. 接口部署

這幾個(ge)模塊完成了后(hou)端(duan)模塊的(de)建立。實際項目中會考慮更多的(de)模塊，比如大(da)模型(xing)的(de)選擇(ze)和部署(shu)，向(xiang)量(liang)數據庫的(de)選擇(ze)，知識庫的(de)準備，前端(duan)頁面的(de)搭建等，這些將不作為本文描(miao)述的(de)重點。

本文(wen)代碼(ma)，關于大(da)(da)模(mo)型(xing)的選擇(ze)，我們將基于 DashScope 提供的嵌入模(mo)型(xing)和(he)大(da)(da)語言模(mo)型(xing)，結合 LangChain 和(he) Chroma 向(xiang)量數據(ju)庫來實現整個系統。

這里我(wo)歷經過(guo)(guo)一(yi)些莫名其妙的(de)(de)(de)磨難，比如剛開(kai)始(shi)我(wo)選擇本(ben)地的(de)(de)(de)Ollama部(bu)署(shu)，包括(kuo)向(xiang)量模(mo)型都(dou)(dou)是(shi)(shi)在(zai)本(ben)地。但是(shi)(shi)在(zai)測(ce)試的(de)(de)(de)過(guo)(guo)程中，發現(xian)召(zhao)(zhao)(zhao)回的(de)(de)(de)結(jie)(jie)果(guo)(guo)很(hen)離譜。比如我(wo)投喂了(le)(le)勞(lao)動法和交通法的(de)(de)(de)內容，然后(hou)問(wen)一(yi)個勞(lao)動法相關的(de)(de)(de)問(wen)題，比如哪些節假日應(ying)該安排休假，結(jie)(jie)果(guo)(guo)召(zhao)(zhao)(zhao)回的(de)(de)(de)結(jie)(jie)果(guo)(guo)中有好多是(shi)(shi)交通法的(de)(de)(de)內容。剛開(kai)始(shi)我(wo)以(yi)為是(shi)(shi)向(xiang)量模(mo)型的(de)(de)(de)問(wen)題，于是(shi)(shi)在(zai)CherryStudio里，構建同樣(yang)的(de)(de)(de)知識庫(ku)，使(shi)用同樣(yang)的(de)(de)(de)向(xiang)量嵌(qian)入模(mo)型，召(zhao)(zhao)(zhao)回測(ce)試的(de)(de)(de)結(jie)(jie)果(guo)(guo)很(hen)符合預(yu)期(qi)。后(hou)來在(zai)LangChain里又嘗(chang)試過(guo)(guo)更換向(xiang)量數據庫(ku)，以(yi)及(ji)更改距離算法，召(zhao)(zhao)(zhao)回的(de)(de)(de)結(jie)(jie)果(guo)(guo)都(dou)(dou)達不(bu)到預(yu)期(qi)。直到有一(yi)天(tian)，本(ben)地部(bu)署(shu)的(de)(de)(de)嵌(qian)入模(mo)型突然不(bu)工作了(le)(le)（真的(de)(de)(de)好奇(qi)怪，同樣(yang)的(de)(de)(de)模(mo)型在(zai)windows和macos都(dou)(dou)有部(bu)署(shu)，突然間就(jiu)都(dou)(dou)不(bu)能(neng)訪問(wen)了(le)(le)，至今原因不(bu)明。），于是(shi)(shi)嘗(chang)試更換到在(zai)線的(de)(de)(de)Qwen的(de)(de)(de)大模(mo)型，召(zhao)(zhao)(zhao)回測(ce)試終于復合預(yu)期(qi)了(le)(le)。

吐槽完(wan)畢，接下來進(jin)入正題：

1. 文檔加載與向量庫構建

文檔加載是 RAG 系統的基礎，需要處理不同格式的文檔并將其轉換為向量存儲。這里我檢索的是所有txt和docx文件。
所有的知識庫文件都放在knowledge_base文件夾下，向量數據庫存儲在chroma_db下。
知識庫為了測試召回方便，我投喂了法律相關的內容，主要有勞動法和道路安全法，同時也投喂了一些自己造的文檔。
向量數據庫這(zhe)里用(yong)(yong)到(dao)的是chroma，其(qi)調用(yong)(yong)方法相對簡單(dan)，不(bu)需(xu)要額(e)外(wai)安裝配(pei)置什么。同(tong)時也可以選(xuan)擇比如(ru)FAISS，Milvus甚至PostgreSQL，但(dan)這(zhe)些向量庫需(xu)要單(dan)獨的部署和配(pei)置，過(guo)程稍微復雜一點。所以這(zhe)篇文章的向量庫選(xuan)擇了(le)Chroma。

核心代碼實現

def load_documents_to_vectorstore(
    document_dir: str = "./RAG/knowledge_base",
    vectorstore_dir: str = "./RAG/chroma_db",
    embedding_model: str = "text-embedding-v1",
    dashscope_api_key: Optional[str] = None,
    chunk_size: int = 1000,
    chunk_overlap: int = 200,
    collection_name: str = "my_collection",
) -> bool:
    # 文檔目錄檢查
    if not os.path.exists(document_dir):
        logger.error(f"文檔目錄不存在: {document_dir}")
        return False

    # 加載不同格式文檔
    documents = []
    # 加載 txt
    txt_loader = DirectoryLoader(document_dir, glob="**/*.txt", loader_cls=TextLoader)
    documents.extend(txt_loader.load())
    # 加載 docx
    docx_loader = DirectoryLoader(document_dir, glob="**/*.docx", loader_cls=Docx2txtLoader)
    documents.extend(docx_loader.load())

    # 文本分割
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=["\n\n", "\n", " ", ""],
    )
    splits = text_splitter.split_documents(documents)

    # 初始化嵌入模型
    embeddings = DashScopeEmbeddings(model=embedding_model, dashscope_api_key=dashscope_api_key)
    
    # 探測嵌入維度，避免維度沖突
    probe_vec = embeddings.embed_query("dimension probe")
    emb_dim = len(probe_vec)
    collection_name = f"{collection_name}_dim{emb_dim}"
    
    # 創建向量存儲
    vectorstore = Chroma.from_documents(
        documents=splits,
        embedding=embeddings,
        collection_name=collection_name,
        persist_directory=persist_dir,
    )
    vectorstore.persist()
    return True

關鍵技術點解析

1.** 文檔加載 **：使用 DirectoryLoader 批量加載目錄中(zhong)的 TXT 和 DOCX 文檔，可根據需求擴展支持 PDF 等(deng)其他格式

2.** 文本分割 **：采用 RecursiveCharacterTextSplitter 進行文本分割，關鍵參(can)數(shu)：

• chunk_size：文本塊大小
• chunk_overlap：文本塊重疊部分，確保上下文連貫性
• separators：分割符列表，優先使用段落分隔

3.** 嵌入處理 **：

• 使用 DashScope 提供的嵌入模型生成文本向量
• 自動探測嵌入維度，避免不同模型間的維度沖突
• 為不同模型創建獨立的存儲目錄，確保向量庫兼容性

4.** 數據寫入(ru) ** 使用(yong)的是from_documents方(fang)法(fa)。這里(li)(li)如(ru)果嵌入(ru)模型不(bu)可用(yong)的話，會(hui)卡(ka)死在(zai)這里(li)(li)。

2. 向量庫構建與檢索功能

向量庫是 RAG 系統的(de)核心(xin)組件(jian)，負責(ze)高效存儲和檢索文本向量。

向量庫構建函數

def build_vectorstore(
    vectorstore_dir: str = "./RAG/chroma_db",
    embedding_model: str = "text-embedding-v4",
    dashscope_api_key: Optional[str] = None,
    collection_name_base: str = "my_collection",
) -> Tuple[Chroma, DashScopeEmbeddings, int, str]:
    # 獲取API密鑰
    if dashscope_api_key is None:
        dashscope_api_key = os.getenv("DASHSCOPE_API_KEY")
    
    # 初始化嵌入模型
    embeddings = DashScopeEmbeddings(model=embedding_model, dashscope_api_key=dashscope_api_key)

    # 探測嵌入維度與持久化目錄
    probe_vec = embeddings.embed_query("dimension probe")
    emb_dim = len(probe_vec)
    collection_name = f"{collection_name_base}_dim{emb_dim}"
    model_dir_tag = embedding_model.replace(":", "_").replace("/", "_")
    persist_dir = os.path.join(vectorstore_dir, model_dir_tag)

    # 加載向量庫
    vs = Chroma(
        persist_directory=persist_dir,
        embedding_function=embeddings,
        collection_name=collection_name,
    )
    return vs, embeddings, emb_dim, persist_dir

檢索功能實現

def retrieve_context(
    question: str,
    k: int,
    vectorstore: Chroma,
) -> List[str]:
    """使用向量庫檢索 top-k 文檔內容，返回文本片段列表"""
    docs = vectorstore.similarity_search(question, k=k)
    chunks: List[str] = []
    for d in docs:
        src = d.metadata.get("source", "<unknown>")
        text = d.page_content.strip().replace("\n", " ")
        chunks.append(f"[source: {src}]\n{text}")
    return chunks

技術要點說明

1.** 向量(liang)庫兼容性處理 **：

• 為不同嵌入模型創建獨立目錄
• 集合名包含維度信息，避免維度沖突
• 自動探測嵌入維度，確保兼容性

2.** 檢(jian)索實(shi)現 **：

• 使用 similarity_search 進行向量相似度檢索
• 返回包含來源信息的文本片段
• 可通過調整 k 值控制返回結果數量，CherryStudio默認是5，所以在這里我也用這個值。

注：similarity_search不返(fan)回相似度信(xin)息，如果需(xu)要這個信(xin)息，需(xu)要使用similarity_search_with_relevance_scores。

3. 問答功能實現

問答(da)功(gong)能是(shi)(shi)(shi) RAG 系統的(de)核心應(ying)用，大體的(de)流程就是(shi)(shi)(shi)結合檢(jian)索到的(de)上下文和(he)大語言模型生成(cheng)回答(da)。如(ru)果你已(yi)經知道(dao)了(le)如(ru)何在(zai)Dify中進行類似操作，那(nei)么這(zhe)部分(fen)代碼理解上就會容易些，尤其是(shi)(shi)(shi)在(zai)用戶提示詞部分(fen)，思(si)路都是(shi)(shi)(shi)一樣的(de)。

問答核心函數

def answer_question(
    question: str,
    top_k: int = 5,
    embedding_model: str = "text-embedding-v4",
    chat_model: str = os.getenv("CHAT_MODEL", "qwen-turbo"),
    dashscope_api_key: Optional[str] = None,
    vectorstore_dir: str = "./RAG/chroma_db",
    temperature: float = 0.2,
    max_tokens: int = 1024,
) -> Tuple[str, List[str]]:
    # 構建向量庫
    vs, embeddings, emb_dim, persist_dir = build_vectorstore(
        vectorstore_dir=vectorstore_dir,
        embedding_model=embedding_model,
        dashscope_api_key=dashscope_api_key,
    )

    # 檢索上下文
    context_chunks = retrieve_context(question, k=top_k, vectorstore=vs)
    sources = []
    for c in context_chunks:
        # 提取來源信息
        if c.startswith("[source: "):
            end = c.find("]\n")
            if end != -1:
                sources.append(c[len("[source: "):end])
    context_str = "\n\n".join(context_chunks)

    # 構造提示詞
    system_prompt = (
        "你是一個嚴謹的問答助手。請基于提供的檢索上下文進行回答，"
        "不要編造信息，若上下文無答案請回答：我不知道。"
    )
    user_prompt = (
        f"問題: {question}\n\n"
        f"檢索到的上下文(可能不完整，僅供參考):\n{context_str}\n\n"
        "請給出簡潔、準確的中文回答，并在需要時引用關鍵點。"
    )

    # 調用大語言模型生成答案
    dashscope.api_key = dashscope_api_key
    gen_kwargs = {
        "model": chat_model,
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        "result_format": "message",
        "temperature": temperature,
        "max_tokens": max_tokens,
    }

    resp = Generation.call(**gen_kwargs)
    answer = _extract_answer_from_generation_response(resp)
    return answer.strip(), sources

關鍵技術點

1.** 提示詞設計 **：

• 系統提示詞明確回答約束（基于上下文、不編造信息）
• 用戶提示詞包含問題和檢索到的上下文
• 明確要求簡潔準確的中文回答

2.** 模(mo)型調用(yong)參數 **：

• temperature：控制輸出隨機性，低溫度值生成更確定的結果，對于問答系統這個值推薦接近0。如果是生成詩詞類應用則推薦接近1.
• max_tokens：限制回答長度
• result_format：指定輸出格式，便于解析

3.** 結果處理 **：

• 從模型響應中提取答案文本
• 收集并返回來源信息，提高回答可信度

4. 構建 HTTP 服務接口

為(wei)(wei)了方便(bian)使用，我們(men)可以(yi)將(jiang)問答功能封裝為(wei)(wei) HTTP 服務，這樣更方便(bian)將(jiang)服務集成到其它應用環境中。

HTTP 服務實現

class QAHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        parsed = urllib.parse.urlparse(self.path)
        if parsed.path != "/qa":
            self.send_response(HTTPStatus.NOT_FOUND)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(json.dumps({"error": "Not Found"}).encode("utf-8"))
            return

        qs = urllib.parse.parse_qs(parsed.query)
        question = (qs.get("question") or [None])[0]
        top_k = int((qs.get("top_k") or [5])[0])
        embedding_model = (qs.get("embedding_model") or [os.getenv("EMBEDDING_MODEL", "text-embedding-v4")])[0]
        chat_model = (qs.get("chat_model") or [os.getenv("CHAT_MODEL", "qwen-turbo")])[0]

        if not question:
            self.send_response(HTTPStatus.BAD_REQUEST)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(json.dumps({"error": "Missing 'question' parameter"}).encode("utf-8"))
            return

        try:
            answer, sources = answer_question(
                question=question,
                top_k=top_k,
                embedding_model=embedding_model,
                chat_model=chat_model,
                dashscope_api_key=os.getenv("DASHSCOPE_API_KEY"),
                vectorstore_dir=os.getenv("VECTORSTORE_DIR", "./RAG/chroma_db"),
            )
            payload = {
                "question": question,
                "answer": answer,
                "sources": sources,
                "top_k": top_k,
                "embedding_model": embedding_model,
                "chat_model": chat_model,
                "status": "ok",
            }
            self.send_response(HTTPStatus.OK)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(json.dumps(payload, ensure_ascii=False).encode("utf-8"))
        except Exception as e:
            logger.error(f"請求處理失敗: {e}")
            self.send_response(HTTPStatus.INTERNAL_SERVER_ERROR)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(json.dumps({"error": "internal_error", "message": str(e)}).encode("utf-8"))

def run_server(host: str = "0.0.0.0", port: int = int(os.getenv("PORT", "8000"))):
    httpd = HTTPServer((host, port), QAHandler)
    logger.info(f"QA 服務已啟動: //localhost:{port}/qa?question=...")
    httpd.serve_forever()

通(tong)過這個(ge)http接口，就可以(yi)供其它(ta)應用(yong)進行調用(yong)，比(bi)如如下我用(yong)Trae生成的(de)前端：

服務特點

1.** 接口設計 ：提供 /qa 端點，支持通過 URL 參數指定問題和模型參數
2. 錯誤處理 ：對缺失參數、服務錯誤等情況返回適當的 HTTP 狀態碼
3. 靈活性 ：支持動態指定 top_k、嵌入模型和聊天模型
4. 易用性 **：返回包含問(wen)題、答案(an)、來源和(he)模型信(xin)息的 JSON 響應(ying)

5. 系統測試與驗證

為確保檢(jian)索的結(jie)果(guo)復(fu)合預期，建議單獨實現召回測試功(gong)能，驗證檢(jian)索效果(guo)：

def recall(
    query: str,
    top_k: int = 5,
    vectorstore_dir: str = "./RAG/chroma_db",
    embedding_model: str = "text-embedding-v4",
    dashscope_api_key: Optional[str] = None,
) -> None:
    vs = build_vectorstore(
        vectorstore_dir=vectorstore_dir,
        embedding_model=embedding_model,
        dashscope_api_key=dashscope_api_key,
    )

    logger.info(f"執行相似度檢索: k={top_k}, query='{query}'")
    docs = vs.similarity_search(query, k=top_k)

    print("\n=== Recall Results ===")
    for i, d in enumerate(docs, start=1):
        src = d.metadata.get("source", "<unknown>")
        snippet = d.page_content.strip().replace("\n", " ")
        if len(snippet) > 500:
            snippet = snippet[:500] + "..."
        print(f"[{i}] source={src}\n    {snippet}\n")

通過召回測試，可以直觀地查看檢索到的文本片段，評估檢索質量，為調整文本分割參數和檢索參數提供依據。
當然召回測試，除了能(neng)在(zai)調用大模型前(qian)提前(qian)看到準(zhun)確度，也(ye)能(neng)在(zai)測試過程中(zhong)，節省大模型調用的成(cheng)本(ben)消(xiao)耗。

總結與展望

本文(wen)匯總(zong)了基于(yu)LangChain 構建 RAG 系統的簡單實現，從(cong)文(wen)檔(dang)加載、向量(liang)存(cun)儲(chu)到問答(da)服務(wu)實現。后續可以從(cong)以下幾個(ge)方(fang)面進行(xing)改進：

1. 支持更多文檔格式（PDF、Markdown 等）
2. 實現更高級的檢索策略（混合檢索、重排序等）
3. 替換向量數據庫
4. 更改相似度算法
5. 增加緩存機制，提高服務響應速度
6. 實現批量處理和增量更新功能
7. 增加用戶認證和權限管理

本文所有代碼可以在以下地址找到：