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

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

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

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

RAG 系統整體架構

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

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

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

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

這里我歷經過一些莫名其妙的磨難，比如剛開始我選擇本地的Ollama部署，包括向量模型都是在本地。但是在測試的過程中，發現召回的結果很離譜。比如我投喂了勞動法和交通法的內容，然后問一個勞動法相關的問題，比如哪些節假日應該安排休假，結果召回的結果中有好多是交通法的內容。剛開始我以為是向量模型的問題，于是在CherryStudio里，構建同樣的知識庫，使用同樣的向量嵌入模型，召回測試的結果很符合預期。后來在LangChain里又嘗試過更換向量數據庫，以及更改距離算法，召回的結果都達不到預期。直到有一天，本地部署的嵌入模型突然不工作了（真的好奇怪，同樣的模型在windows和macos都有部署，突然間就都不能訪問了，至今原因不明。），于是嘗試更換到在線的Qwen的大模型，召回測試終于復合預期了。

吐槽完畢，接下來進入正題：

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

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

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

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

3.** 嵌入處理 **：

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

4.** 數據寫入 ** 使用的是from_documents方法。這里如果嵌入模型不可用的話，會卡死在這里。

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

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

向量庫構建函數

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

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

2.** 檢索實現 **：

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

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

3. 問答功能實現

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

問答核心函數

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

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

3.** 結果處理 **：

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

4. 構建 HTTP 服務接口

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

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 服務已啟動: http://localhost:{port}/qa?question=...")
    httpd.serve_forever()

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

服務特點

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

5. 系統測試與驗證

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

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

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

總結與展望

本文匯總了基于LangChain 構建 RAG 系統的簡單實現，從文檔加載、向量存儲到問答服務實現。后續可以從以下幾個方面進行改進：

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

本文所有代碼可以在以下地址找到：
https://github.com/microsoftbi/Langchain_DEMO/tree/main/RAG

posted @ 2025-10-29 11:33 哥本哈士奇(aspnetx) 閱讀(166) 評論(0) 收藏舉報

刷新頁面返回頂部

專注微軟平臺的商業智能解決方案

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

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

RAG 系統整體架構

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

核心代碼實現

關鍵技術點解析

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

向量庫構建函數

檢索功能實現

技術要點說明

3. 問答功能實現

問答核心函數

關鍵技術點

4. 構建 HTTP 服務接口

HTTP 服務實現

服務特點

5. 系統測試與驗證

總結與展望

公告