LangGraph Agentic RAG 教學：建構智能檢索代理系統 簡介 Agentic RAG（Retrieval-Augmented Generation）是一個強大的架構，結合了檢索增強生成和代理系統的概念。這個系統能夠智能地決定何時需要從知識庫中檢索資訊，何時可以直接回應用戶。本教學將逐步帶你了解如何使用 LangGraph 建構一個完整的 Agentic RAG 系統。 概述 在這個教學中，我們將建立一個檢索代理，能夠： 自動判斷是否需要從向量資料庫中檢索上下文 評估檢索文件的相關性 必要時重寫查詢以獲得更好的結果 基於檢索到的內容生成準確的回答 系統架構圖 graph TD    A[開始] --> B[生成查詢或回應]    B --> C{需要檢索？}    C -->|是| D[檢索工具]    C -->|否| E[結束]    D --> F{文件相關？}    F -->|是| G[生成答案]    F -->|否| H[重寫問題]    H --> B    G --> E 環境設置 首先，安裝所需的套件： pip install -U --quiet langgraph "langchain[openai]" langchain-community langchain-text-splitters 設置 API 金鑰： import getpass import os def _set_env(key: str):    if key not in os.environ:        os.environ[key] = getpass.getpass(f"{key}:") _set_env("OPENAI_API_KEY") 步驟 1：預處理文件 1.1 載入文件 使用 WebBaseLoader 從網路載入文件： from langchain_community.document_loaders import WebBaseLoader urls = [    "https://lilianweng.github.io/posts/2024-11-28-reward-hacking/",    "https://lilianweng.github.io/posts/2024-07-07-hallucination/",    "https://lilianweng.github.io/posts/2024-04-12-diffusion-video/", ] docs = [WebBaseLoader(url).load() for url in urls] 1.2 分割文件 將文件分割成較小的區塊以便索引： from langchain_text_splitters import RecursiveCharacterTextSplitter docs_list = [item for sublist in docs for item in sublist] text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(    chunk_size=100, chunk_overlap=50 ) doc_splits = text_splitter.split_documents(docs_list) 步驟 2：建立檢索工具 2.1 創建向量存儲 使用 OpenAI 嵌入創建記憶體向量存儲： from langchain_core.vectorstores import InMemoryVectorStore from langchain_openai import OpenAIEmbeddings vectorstore = InMemoryVectorStore.from_documents(    documents=doc_splits, embedding=OpenAIEmbeddings() ) retriever = vectorstore.as_retriever() 2.2 創建檢索工具 將檢索器包裝成工具： from langchain.tools.retriever import create_retriever_tool retriever_tool = create_retriever_tool(    retriever,    "retrieve_blog_posts",    "搜尋並返回關於 Lilian Weng 部落格文章的資訊。", ) 步驟 3：構建圖形節點 3.1 生成查詢或回應節點 創建決定是否需要檢索的節點： from langgraph.graph import MessagesState from langchain.chat_models import init_chat_model response_model = init_chat_model("openai:gpt-4", temperature=0) def generate_query_or_respond(state: MessagesState):    """呼叫模型生成回應。根據問題決定是否使用檢索工具。"""    response = (        response_model        .bind_tools([retriever_tool])        .invoke(state["messages"])    )    return {"messages": [response]} 3.2 評分文件節點 創建評估文件相關性的節點： from pydantic import BaseModel, Field from typing import Literal class GradeDocuments(BaseModel):    """使用二元分數評分文件相關性"""    binary_score: str = Field(        description="相關性分數：'yes' 表示相關，'no' 表示不相關"    ) grader_model = init_chat_model("openai:gpt-4", temperature=0) def grade_documents(    state: MessagesState, ) -> Literal["generate_answer", "rewrite_question"]:    """判斷檢索到的文件是否與問題相關"""    question = state["messages"][0].content    context = state["messages"][-1].content        # 評分邏輯...    if score == "yes":        return "generate_answer"    else:        return "rewrite_question" 3.3 重寫問題節點 當文件不相關時，重寫原始問題： def rewrite_question(state: MessagesState):    """重寫原始用戶問題"""    messages = state["messages"]    question = messages[0].content    # 重寫邏輯...    return {"messages": [{"role": "user", "content": response.content}]} 3.4 生成答案節點 基於檢索到的內容生成最終答案： def generate_answer(state: MessagesState):    """生成答案"""    question = state["messages"][0].content    context = state["messages"][-1].content    # 生成答案邏輯...    return {"messages": [response]} 步驟 4：組裝圖形 將所有節點組合成完整的工作流程： from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode, tools_condition workflow = StateGraph(MessagesState) # 添加節點 workflow.add_node("generate_query_or_respond", generate_query_or_respond) workflow.add_node("retrieve", ToolNode([retriever_tool])) workflow.add_node("rewrite_question", rewrite_question) workflow.add_node("generate_answer", generate_answer) # 添加邊 workflow.add_edge(START, "generate_query_or_respond") # 條件邊 workflow.add_conditional_edges(    "generate_query_or_respond",    tools_condition,    {        "tools": "retrieve",        END: END,    }, ) workflow.add_conditional_edges(    "retrieve",    grade_documents, ) workflow.add_edge("generate_answer", END) workflow.add_edge("rewrite_question", "generate_query_or_respond") # 編譯圖形 graph = workflow.compile() 步驟 5：運行 Agentic RAG 使用完整的系統來處理查詢： # 執行查詢 for chunk in graph.stream(    {        "messages": [            {                "role": "user",                "content": "Lilian Weng 對獎勵駭客的類型說了什麼？",            }        ]    } ):    for node, update in chunk.items():        print(f"來自節點 {node} 的更新")        update["messages"][-1].pretty_print()        print("\n") 實際應用範例 以下是一個完整的使用範例： # 初始化系統 agentic_rag = workflow.compile() # 處理複雜查詢 response = agentic_rag.invoke({    "messages": [        {            "role": "user",            "content": "解釋 Lilian Weng 文章中關於強化學習中獎勵駭客的主要觀點",        }    ] }) # 顯示最終答案 print(response["messages"][-1].content) 最佳實踐 文件分塊策略：根據內容類型調整 chunk_size 和 chunk_overlap 評分閾值：調整文件相關性評分的標準 重寫策略：使用更智能的提示來改進問題重寫 錯誤處理：添加適當的異常處理機制 監控和日誌：使用 LangSmith 追蹤系統性能 擴展建議 多語言支援：添加多語言文件處理能力 多源檢索：整合多個知識來源 快取機制：實現查詢和回答的快取 回饋循環：加入用戶回饋來改進系統 安全措施：添加內容過濾和安全檢查 結論 Agentic RAG 系統結合了智能決策和檢索增強生成的優勢，能夠提供更準確、更相關的回答。透過 LangGraph 的圖形架構，我們可以輕鬆地構建和維護這樣的複雜系統。 這個系統的主要優勢包括： 智能判斷何時需要外部資訊 自動評估和改進檢索品質 靈活的工作流程控制 易於擴展和客製化 希望這個教學能幫助你建立自己的 Agentic RAG 系統！