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

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

RAG 系統整體架構

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

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

這幾(ji)個模(mo)(mo)塊完成(cheng)了后(hou)端模(mo)(mo)塊的(de)建立。實際項目(mu)中會考慮更多的(de)模(mo)(mo)塊，比如(ru)大模(mo)(mo)型的(de)選(xuan)擇和部署，向(xiang)量數(shu)據庫的(de)選(xuan)擇，知識庫的(de)準備(bei)，前端頁面(mian)的(de)搭建等，這些將(jiang)不作(zuo)為本文描述的(de)重點(dian)。

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

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

吐槽完畢，接下來進入(ru)正(zheng)題：

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

文檔加載是 RAG 系統的基礎，需要處理不同格式的文檔并將其轉換為向量存儲。這里我檢索的是所有txt和docx文件。
所有的知識庫文件都放在knowledge_base文件夾下，向量數據庫存儲在chroma_db下。
知識庫為了測試召回方便，我投喂了法律相關的內容，主要有勞動法和道路安全法，同時也投喂了一些自己造的文檔。
向量數據庫這(zhe)里用到的是chroma，其調用方(fang)法相對簡單，不需要(yao)(yao)額外安裝配(pei)置(zhi)什么。同時(shi)也可以選擇(ze)比如FAISS，Milvus甚至(zhi)PostgreSQL，但這(zhe)些(xie)向量庫需要(yao)(yao)單獨(du)的部署和配(pei)置(zhi)，過程稍微復雜(za)一點。所以這(zhe)篇文章的向量庫選擇(ze)了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 批(pi)量加載目(mu)錄中的 TXT 和 DOCX 文檔，可根據需求擴展支持 PDF 等其(qi)他(ta)格式

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

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

3.** 嵌入處理 **：

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

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

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

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

向量庫構建函數

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.** 向量庫兼容性處(chu)理 **：

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

2.** 檢索實現 **：

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

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

3. 問答功能實現

問(wen)答(da)功能(neng)是(shi)(shi) RAG 系統的核心應用，大體的流程(cheng)就(jiu)是(shi)(shi)結合檢索(suo)到(dao)的上(shang)下文(wen)和大語言模(mo)型生(sheng)成回答(da)。如果你已經(jing)知道了如何(he)在Dify中進行類(lei)似操作，那(nei)么這部(bu)分代碼(ma)理解上(shang)就(jiu)會(hui)容(rong)易些，尤其(qi)是(shi)(shi)在用戶提示(shi)詞部(bu)分，思路都是(shi)(shi)一樣的。

問答核心函數

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.** 模型調(diao)用參數 **：

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

3.** 結果處(chu)理 **：

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

4. 構建 HTTP 服務接口

為了方(fang)便使(shi)用(yong)，我們可以將問答(da)功能(neng)封裝為 HTTP 服務，這(zhe)樣更方(fang)便將服務集(ji)成到其它應(ying)用(yong)環(huan)境中。

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()

通過這(zhe)個http接口(kou)，就可以(yi)供其它應用進行調用，比如如下我(wo)用Trae生成的前端：

服務特點

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

5. 系統測試與驗證

為確保檢索(suo)的結果(guo)復合預期(qi)，建議(yi)單獨實現(xian)召回測試(shi)功(gong)能，驗(yan)證檢索(suo)效果(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")

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

總結與展望

本文(wen)匯總(zong)了基于LangChain 構建 RAG 系統的(de)簡單實(shi)現(xian)，從文(wen)檔(dang)加載、向(xiang)量(liang)存儲到問答(da)服務(wu)實(shi)現(xian)。后續(xu)可以從以下(xia)幾(ji)個方面進(jin)行改(gai)進(jin)：

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

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