肥仔教程网

SEO 优化与 Web 开发技术学习分享平台

2025年09月21日

Agentic问答系统:自适应RAG+LangGraph = 智能问答系统最强组合!

Adaptive RAG（自适应检索增强生成）是一种先进的 RAG 策略，它智能地将动态查询分析与自我纠错机制相结合，以优化响应准确性。

1. 引言

检索增强生成（RAG）彻底革新了我们构建能够访问外部知识并进行推理的 AI 系统的方式。然而，随着应用程序的复杂性不断增加，传统 RAG 方法的局限性日益凸显。

如今，我们正在经历一种从线性、固定管道到智能、自适应系统的转变，这些系统可以根据查询复杂度和上下文动态调整其检索与生成策略。

在本文中，我们将深入探讨 Agentic RAG 系统，特别是如何使用 LangGraph 与 Google Gemini 实现 Adaptive RAG。在实现之前，我们先来理解什么是 Adaptive RAG。

2. Adaptive RAG（自适应检索增强生成）

Adaptive RAG 是一种先进的 RAG 策略，它将动态查询分析和主动/自我修正机制智能结合在一起。

Adaptive RAG 的核心理念：并非所有查询都具有相同复杂度。研究表明，现实世界中的查询具有多种复杂程度：

  • 简单查询：如 “巴黎是哪个国家的首都？” —— 直接由 LLM 回答即可
  • 多跳查询：如 “占领 Malakoff 的人是什么时候来到 Philipsburg 所在地区的？” —— 需要 4 个推理步骤。

上图示比较了不同检索增强 LLM 策略在问答任务中的应用方式：

（A）单步方法：对简单查询高效，但无法处理多跳推理。
（B）多步方法：适用于复杂查询，但在简单查询上引入了不必要的计算成本。
（C）自适应方法：通过查询复杂度分类器，选择最适合的策略，可灵活切换单步、多步甚至无需检索方式。

核心问题，如上图所示：

  • （A）单步方法：适合简单查询，但不能支持复杂推理
  • （B）多步方法：支持复杂推理，但对简单查询效率低
  • （C）Adaptive 方法：通过查询分类器，将每个查询引导至最适策略

2.1 理解 Adaptive RAG 工作流程

从实现图中可以看到，Adaptive RAG 通过基于图的状态管理系统，构建了一个智能决策流程：

1. 查询路由与分类

系统从一个训练好的复杂度分类器开始，分析每一个输入问题。这不仅仅是关键词匹配，而是通过模型进行的复杂度评估，用于判断：

  • 是否需要检索？（模型内知识是否足够？）
  • 如果需要检索，复杂度属于哪一层级？
  • 应采用的最优策略：无需检索 / 单步检索 / 多跳推理

2. 动态知识获取策略

基于复杂度分类，系统将查询路由至以下策略之一：

  • 基于索引的检索：适用于知识库可回答的问题
  • 网页搜索：本地知识无法覆盖或需要最新信息时
  • 无需检索：模型的参数知识即可回答的问题

3. 多阶段质量保障机制

系统在多个关键节点设置了评估机制：

  • 文档相关性评估：通过置信度评分判断检索结果质量
  • 幻觉检测机制：验证生成内容是否基于已检索证据
  • 答案质量评估：确保最终答案真正解决用户问题

3. 我们要构建什么？

本文将带你构建一个能够智能决策的信息检索与路由机制的高级 RAG 系统。该系统具备：

  • 智能查询分析机制
  • 严谨的评估框架
  • 支持多源信息（本地知识库 + Web）的自适应架构

我们将对 LangChain 原有代码结构进行重构，目标包括：提升可读性、易维护性、开发者体验。

实现组件包括：

  • LangGraph：管理有状态的复杂工作流
  • Google Gemini：主语言模型
  • 向量数据库：文档检索效率保障
  • 网页搜索集成：实时信息检索
  • 全面评估框架：保障质量

4. 项目结构

在正式编码前，先来理解整个项目目录：

building-adaptive-rag/
├── graph/
│   ├── chains/
│   │   ├── tests/
│   │   │   ├── __init__.py
│   │   │   └── test_chains.py
│   │   ├── __init__.py
│   │   ├── answer_grader.py
│   │   ├── generation.py
│   │   ├── hallucination_grader.py
│   │   ├── retrieval_grader.py
│   │   └── router.py
│   ├── nodes/
│   │   ├── __init__.py
│   │   ├── generate.py
│   │   ├── grade_documents.py
│   │   ├── retrieve.py
│   │   └── web_search.py
│   ├── __init__.py
│   ├── consts.py
│   ├── graph.py
│   └── state.py
├── static/
│   ├── LangChain-logo.png
│   ├── Langgraph Adaptive Rag.png
│   └── graph.png
├── .env
├── .gitignore
├── ingestion.py
├── main.py
├── model.py
├── README.md
└── requirements.txt

整个项目采用模块化架构，每个组件有明确职责：

  • graph/ 中管理核心逻辑：
    • chains/：各类 LLM 操作
    • nodes/：工作流程的节点
    • graph.py：协调所有节点的主图逻辑

4.1 使用 UV 进行环境配置

我们将使用 UV（一种快速的 Python 包管理器）搭建开发环境。

安装 UV：

curl -LsSf https://astral.sh/uv/install.sh | sh

创建项目目录与虚拟环境：

mkdir building-adaptive-rag
cd building-adaptive-rag
uv venv --python 3.10
source .venv/bin/activate  # Windows 下使用 .venv\\Scripts\\activate

创建 requirements.txt，添加以下依赖：

beautifulsoup4
langchain-community
tiktoken
langchainhub
langchain
langgraph
tavily-python
langchain-openai
python-dotenv
black
isort
pytest
langchain-chroma
langchain-tavily==0.1.5
langchain_aws
langchain_google_genai

安装依赖：

uv pip install -r requirements.txt

配置环境变量，创建 .env 文件：

GOOGLE_API_KEY=your_tavily_api_key_here
TAVILY_API_KEY=your_tavily_api_key_here  # 网页搜索功能所需
LANGCHAIN_API_KEY=your_langchain_api_key_here  # 可选，用于跟踪
LANGCHAIN_TRACING_V2=true                # 可选
LANGCHAIN_ENDPOINT=https://api.smith.langchain.com  # 可选
LANGCHAIN_PROJECT=agentic-rag            # 可选

5. 实现指南

第 1 步：定义状态管理系统和核心常量

在 graph/state.py 文件中，我们将定义工作流运行时的状态。这些状态将记录对话历史、用户问题、生成答案、检索到的文档等信息。

此外，在 graph/consts.py 中，我们定义所有核心常量，如提示模板（prompt）、命令常量和其他重复使用的字符串。

第 2 步：定义聊天模型与嵌入模型

我们将在 model.py 文件中设置基础模型。

# from langchain_openai import ChatOpenAI
from langchain_aws import ChatBedrock
# from langchain_openai import OpenAIEmbeddings
from langchain_google_genai import GoogleGenerativeAIEmbeddings


# llm_model = ChatOpenAI(temperature=0)
llm_model =  ChatBedrock(model_id="us.anthropic.claude-sonnet-4-20250514-v1:0", region_name="us-west-2", temperature=0)   


embed_model = GoogleGenerativeAIEmbeddings(model="models/text-embedding-004")
  • ChatGoogleGenerativeAI 用作主聊天模型（LLM）
  • GoogleGenerativeAIEmbeddings 用于生成文档嵌入
  • TavilySearchResults 是集成的网页搜索工具（基于 Tavily）

这些组件后续将在不同节点中被调用。

第 3 步：构建文档摄取管道

文件：ingestion.py

from dotenv import load_dotenv
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_chroma import Chroma
from langchain_community.document_loaders import WebBaseLoader
from model import embed_model


load_dotenv()


urls = [
    "https://lilianweng.github.io/posts/2023-06-23-agent/",
    "https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",
    "https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/",
]


docs = [WebBaseLoader(url).load() for url in urls]
docs_list = [item for sublist in docs for item in sublist]


text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=250, chunk_overlap=0
)


doc_splits = text_splitter.split_documents(docs_list)


embed = embed_model


# Create vector store with documents
vectorstore = Chroma.from_documents(
    documents=doc_splits,
    collection_name="rag-chroma",
    embedding=embed,
    persist_directory="./.chroma",
)


# Create retriever
retriever = vectorstore.as_retriever()

我们创建一个摄取脚本，它：

  1. 从指定目录读取文档（.md、.txt 等）
  2. 将文档分割成合适的 chunk
  3. 为每个 chunk 创建嵌入
  4. 存入向量数据库（如 Chroma）

在摄取后，文档就可以供后续检索使用。

第 4 步：构建查询路由链（Router）

文件：graph/chains/router.py

from typing import Literal


from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from model import llm_model


class RouteQuery(BaseModel):
    """Route a user query to the most relevant datasource."""


    datasource: Literal["vectorstore", "websearch"] = Field(
        ...,
        description="Given a user question choose to route it to web search or a vectorstore.",
    )


llm = llm_model


structured_llm_router = llm.with_structured_output(RouteQuery)


system = """You are an expert at routing a user question to a vectorstore or web search.
The vectorstore contains documents related to agents, prompt engineering, and adversarial attacks.
Use the vectorstore for questions on these topics. For all else, use web-search."""
route_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "{question}"),
    ]
)


question_router = route_prompt | structured_llm_router

这是系统智能决策的第一步 —— 判断该如何处理用户查询。

我们使用 LLM 提示模板，构建一个链来输出下列三种操作之一：

  • route_to_retrieval：查询需要文档支持
  • route_to_search：需要访问 Web 搜索结果
  • route_to_generation：直接生成，不需要检索

这个路由器会成为整个 LangGraph 中的决策节点。

第 5 步：创建文档检索评分器（Retrieval Grader）

文件：graph/chains/retrieval_grader.py

from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from model import llm_model


llm = llm_model


class GradeDocuments(BaseModel):
    """Binary score for relevance check on retrieved documents."""


    binary_score: str = Field(
        description="Documents are relevant to the question, 'yes' or 'no'"
    )




structured_llm_grader = llm.with_structured_output(GradeDocuments)


system = """You are a grader assessing relevance of a retrieved document to a user question. \n 
    If the document contains keyword(s) or semantic meaning related to the question, grade it as relevant. \n
    Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question."""


grade_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "Retrieved document: \n\n {document} \n\n User question: {question}"),
    ]
)


retrieval_grader = grade_prompt | structured_llm_grader

这个评分器用于评估从向量数据库中检索到的文档是否真正与用户查询相关。如果置信度低，系统将选择从 Web 重新获取信息。

其核心逻辑是：如果文档评分不够高，我们就尝试网页搜索。

第 6 步：构建幻觉检测系统（Hallucination Grader）

文件：graph/chains/hallucination_grader.py

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableSequence
from pydantic import BaseModel, Field
from model import llm_model


llm =  llm_model


class GradeHallucinations(BaseModel):
    """Binary score for hallucination present in generation answer."""


    binary_score: bool = Field(
        description="Answer is grounded in the facts, 'yes' or 'no'"
    )




structured_llm_grader = llm.with_structured_output(GradeHallucinations)


system = """You are a grader assessing whether an LLM generation is grounded in / supported by a set of retrieved facts. \n 
     Give a binary score 'yes' or 'no'. 'Yes' means that the answer is grounded in / supported by the set of facts."""


hallucination_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "Set of facts: \n\n {documents} \n\n LLM generation: {generation}"),
    ]
)


hallucination_grader: RunnableSequence = hallucination_prompt | structured_llm_grader

幻觉检测是 Adaptive RAG 的关键步骤之一。我们将生成的答案与提供的文档进行比较，并判断：

  • 答案是否真实基于检索内容？
  • 是否存在幻觉（即模型“编造”内容）？

这个评分器能显著提升生成的可靠性。

第 7 步：创建答案质量评分器（Answer Grader）

文件：graph/chains/answer_grader.py

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableSequence
from pydantic import BaseModel, Field
from model import llm_model




class GradeAnswer(BaseModel):


    binary_score: bool = Field(
        description="Answer addresses the question, 'yes' or 'no'"
    )


llm =  llm_model


structured_llm_grader = llm.with_structured_output(GradeAnswer)


system = """You are a grader assessing whether an answer addresses / resolves a question \n 
     Give a binary score 'yes' or 'no'. Yes' means that the answer resolves the question."""
answer_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "User question: \n\n {question} \n\n LLM generation: {generation}"),
    ]
)


answer_grader: RunnableSequence = answer_prompt | structured_llm_grader

我们使用一个 LLM 链来评估最终答案是否真正解决了用户的问题。如果评分较低，系统将尝试重新检索、重新生成，直到达到满意的质量。

第 8 步：构建生成链（Final Generator）

文件：graph/chains/generation.py

from langchain import hub
from langchain_core.output_parsers import StrOutputParser
from model import llm_model


llm = llm_model


prompt = hub.pull("rlm/rag-prompt")


generation_chain = prompt | llm | StrOutputParser()

这是最终用于生成用户答案的 LLM 链。它接收以下输入：

  • 检索到的文档（可为空）
  • 原始用户查询

输出结构化格式的答案（可以含有来源引用等）。

第 9 步：实现检索节点（Retrieve Node）

文件：graph/nodes/retrieve.py

from typing import Any, Dict


from graph.state import GraphState
from ingestion import retriever




def retrieve(state: GraphState) -> Dict[str, Any]:
    print("---RETRIEVE---")
    question = state["question"]


    documents = retriever.invoke(question)
    return {"documents": documents, "question": question}

此节点从向量数据库中执行文档检索。

注意：它仅在路由器认为“需要检索”时被调用。

第 10 步：创建文档评分节点（Grade Documents Node）

文件：graph/nodes/grade_documents.py

from typing import Any, Dict


from graph.chains.retrieval_grader import retrieval_grader
from graph.state import GraphState




def grade_documents(state: GraphState) -> Dict[str, Any]:
    """
    Determines whether the retrieved documents are relevant to the question
    If any document is not relevant, we will set a flag to run web search


    Args:
        state (dict): The current graph state


    Returns:
        state (dict): Filtered out irrelevant documents and updated web_search state
    """


    print("---CHECK DOCUMENT RELEVANCE TO QUESTION---")
    question = state["question"]
    documents = state["documents"]


    filtered_docs = []
    web_search = False
    for d in documents:
        score = retrieval_grader.invoke(
            {"question": question, "document": d.page_content}
        )
        grade = score.binary_score
        if grade.lower() == "yes":
            print("---GRADE: DOCUMENT RELEVANT---")
            filtered_docs.append(d)
        else:
            print("---GRADE: DOCUMENT NOT RELEVANT---")
            web_search = True
            continue
    return {"documents": filtered_docs, "question": question, "web_search": web_search}

此节点结合步骤 5、6、7 中的所有评分器，用于动态判断：

  • 是否继续使用当前文档？
  • 是否需要尝试其他策略（如 Web 搜索）？
  • 是否要重新生成答案？

它是整个系统“自我纠错”机制的核心。

第 11 步：构建网页搜索节点（Web Search Node）

文件：graph/nodes/web_search.py

from typing import Any, Dict
from dotenv import load_dotenv
from langchain.schema import Document
from langchain_tavily import TavilySearch
from graph.state import GraphState


load_dotenv()


web_search_tool = TavilySearch(max_results=3)


def web_search(state: GraphState) -> Dict[str, Any]:
    print("---WEB SEARCH---")
    question = state["question"]


    # Initialize documents - this was the missing part!
    documents = state.get("documents", [])  # Get existing documents or empty list


    tavily_results = web_search_tool.invoke({"query": question})["results"]
    joined_tavily_result = "\n".join(
        [tavily_result["content"] for tavily_result in tavily_results]
    )
    web_results = Document(page_content=joined_tavily_result)


    # Add web results to existing documents (or create new list if documents was empty)
    if documents:
        documents.append(web_results)
    else:
        documents = [web_results]


    return {"documents": documents, "question": question}


if __name__ == "__main__":
    web_search(state={"question": "agent memory", "documents": None})

如果评分系统认为当前文档不足以支持答案，系统将调用此节点进行实时 Web 搜索。我们使用 Tavily API 来完成搜索。

第 12 步：实现生成节点（Generate Node）

文件：graph/nodes/generate.py

from typing import Any, Dict


from graph.chains.generation import generation_chain
from graph.state import GraphState




def generate(state: GraphState) -> Dict[str, Any]:
    print("---GENERATE---")
    question = state["question"]
    documents = state["documents"]


    generation = generation_chain.invoke({"context": documents, "question": question})
    return {"documents": documents, "question": question, "generation": generation}

此节点用于触发最终答案的生成。它接收所有检索与评分信息，并调用之前定义的生成链。

第 13 步：构建完整图形工作流（LangGraph）

文件：graph/graph.py

from dotenv import load_dotenv
from langgraph.graph import END, StateGraph


from graph.chains.answer_grader import answer_grader
from graph.chains.hallucination_grader import hallucination_grader
from graph.chains.router import RouteQuery, question_router
from graph.consts import GENERATE, GRADE_DOCUMENTS, RETRIEVE, WEBSEARCH
from graph.nodes import generate, grade_documents, retrieve, web_search
from graph.state import GraphState


load_dotenv()


def decide_to_generate(state):
    print("---ASSESS GRADED DOCUMENTS---")


    if state["web_search"]:
        print(
            "---DECISION: NOT ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, INCLUDE WEB SEARCH---"
        )
        return WEBSEARCH
    else:
        print("---DECISION: GENERATE---")
        return GENERATE


def grade_generation_grounded_in_documents_and_question(state: GraphState) -> str:
    print("---CHECK HALLUCINATIONS---")
    question = state["question"]
    documents = state["documents"]
    generation = state["generation"]


    score = hallucination_grader.invoke(
        {"documents": documents, "generation": generation}
    )


    if hallucination_grade := score.binary_score:
        print("---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---")
        print("---GRADE GENERATION vs QUESTION---")
        score = answer_grader.invoke({"question": question, "generation": generation})
        if answer_grade := score.binary_score:
            print("---DECISION: GENERATION ADDRESSES QUESTION---")
            return "useful"
        else:
            print("---DECISION: GENERATION DOES NOT ADDRESS QUESTION---")
            return "not useful"
    else:
        print("---DECISION: GENERATION IS NOT GROUNDED IN DOCUMENTS, RE-TRY---")
        return "not supported"




def route_question(state: GraphState) -> str:
    """
    Route question to web search or RAG.


    Args:
        state (dict): The current graph state


    Returns:
        str: Next node to call
    """


    print("---ROUTE QUESTION---")
    question = state["question"]
    source: RouteQuery = question_router.invoke({"question": question})


    if source.datasource == WEBSEARCH:
        print("---ROUTE QUESTION TO WEB SEARCH---")
        return WEBSEARCH
    elif source.datasource == "vectorstore":
        print("---ROUTE QUESTION TO RAG---")
        return RETRIEVE




workflow = StateGraph(GraphState)


workflow.add_node(RETRIEVE, retrieve)
workflow.add_node(GRADE_DOCUMENTS, grade_documents)
workflow.add_node(GENERATE, generate)
workflow.add_node(WEBSEARCH, web_search)


workflow.set_conditional_entry_point(
    route_question,
    {
        WEBSEARCH: WEBSEARCH,
        RETRIEVE: RETRIEVE,
    },
)


# workflow.set_entry_point(RETRIEVE)


workflow.add_edge(RETRIEVE, GRADE_DOCUMENTS)
workflow.add_conditional_edges(
    GRADE_DOCUMENTS,
    decide_to_generate,
    {
        WEBSEARCH: WEBSEARCH,
        GENERATE: GENERATE,
    },
)


workflow.add_conditional_edges(
    GENERATE,
    grade_generation_grounded_in_documents_and_question,
    {
        "not supported": GENERATE,
        "useful": END,
        "not useful": WEBSEARCH,
    },
)
workflow.add_edge(WEBSEARCH, GENERATE)
workflow.add_edge(GENERATE, END)


app = workflow.compile()


app.get_graph().draw_mermaid_png(output_file_path="graph.png")

这是最核心的部分，我们在这里：

  • 使用 LangGraph 构建整个节点状态机
  • 定义每个节点的状态传递方式
  • 实现路由逻辑（如条件跳转、循环）
  • 构建一个复杂的、基于状态的自适应处理图

这一部分将综合前面所有节点与评分器。

第 14 步：创建主应用入口

文件：main.py

from dotenv import load_dotenv


load_dotenv()


from graph.graph import app


def format_response(result):
    """Format the response from the graph for better readability"""
    if isinstance(result, dict) and "generation" in result:
        return result["generation"]
    elif isinstance(result, dict) and "answer" in result:
        return result["answer"]
    else:
        # Fallback to string representation
        return str(result)




def main():
    print("=" * 60)
    print(" Advanced RAG Chatbot")
    print("=" * 60)
    print("Welcome! Ask me anything or type 'quit', 'exit', or 'bye' to stop.")
    print("-" * 60)


    while True:
        try:
            # Get user input
            user_question = input("\n You: ").strip()


            # Check for exit commands
            if user_question.lower() in ['quit', 'exit', 'bye', 'q']:
                print("\n Goodbye! Thanks for chatting!")
                break


            # Skip empty inputs
            if not user_question:
                print("Please enter a question.")
                continue


            # Show processing indicator
            print("\n Bot: Thinking...")


            # Process the question through the graph
            result = app.invoke(input={"question": user_question})


            # Format and display the response
            response = format_response(result)
            print(f"\n Bot: {response}")


        except KeyboardInterrupt:
            print("\n\n Goodbye! Thanks for chatting!")
            break
        except Exception as e:
            print(f"\n Sorry, I encountered an error: {str(e)}")
            print("Please try asking your question again.")




if __name__ == "__main__":
    main()

这个文件将作为系统的入口，处理：

  • 用户输入
  • 调用 LangGraph 的执行逻辑
  • 控制整个查询 - 检索 - 生成流程

它会输出最终回答，并打印中间评分与调试信息。

第 15 步：全面测试套件

文件：graph/chains/tests/test_chains.py

from pprint import pprint
import pytest


from dotenv import load_dotenv


load_dotenv()




from graph.chains.generation import generation_chain
from graph.chains.hallucination_grader import (GradeHallucinations,
                                               hallucination_grader)
from graph.chains.retrieval_grader import GradeDocuments, retrieval_grader
from graph.chains.router import RouteQuery, question_router
from ingestion import retriever




def test_retrival_grader_answer_yes() -> None:
    question = "agent memory"
    docs = retriever.invoke(question)


    doc_txt = docs[1].page_content


    res: GradeDocuments = retrieval_grader.invoke(
        {"question": question, "document": doc_txt}
    )


    assert res.binary_score == "yes"




def test_retrival_grader_answer_no() -> None:
    question = "agent memory"
    docs = retriever.invoke(question)


    # Skip test if no documents are returned
    # if len(docs) == 0:
    #     pytest.skip(f"No documents returned for query: {question}")


    doc_txt = docs[1].page_content


    res: GradeDocuments = retrieval_grader.invoke(
        {"question": "how to make pizaa", "document": doc_txt}
    )


    assert res.binary_score == "no"




def test_generation_chain() -> None:
    question = "agent memory"
    docs = retriever.invoke(question)
    generation = generation_chain.invoke({"context": docs, "question": question})
    pprint(generation)




def test_hallucination_grader_answer_yes() -> None:
    question = "agent memory"
    docs = retriever.invoke(question)


    generation = generation_chain.invoke({"context": docs, "question": question})
    res: GradeHallucinations = hallucination_grader.invoke(
        {"documents": docs, "generation": generation}
    )
    assert res.binary_score




def test_hallucination_grader_answer_no() -> None:
    question = "agent memory"
    docs = retriever.invoke(question)


    res: GradeHallucinations = hallucination_grader.invoke(
        {
            "documents": docs,
            "generation": "In order to make pizza we need to first start with the dough",
        }
    )
    assert not res.binary_score




def test_router_to_vectorstore() -> None:
    question = "agent memory"


    res: RouteQuery = question_router.invoke({"question": question})
    assert res.datasource == "vectorstore"




def test_router_to_websearch() -> None:
    question = "how to make pizza"


    res: RouteQuery = question_router.invoke({"question": question})
    assert res.datasource == "websearch"

我们为所有 LLM 链和节点编写了测试用例。测试包括：

  • 路由链是否能正确分类查询
  • 评分器是否能准确评估文档质量
  • 幻觉检测是否有效识别幻觉

这些测试确保我们的系统具有稳定性和可扩展性。

6. 系统运行方式

通过将所有组件（链、节点、评分器、搜索器）连接在一个有状态图中，系统具备以下能力：

6.1 智能查询路由

能够判断用户问题是否需要外部知识支持，并选择合适策略：

  • 无需检索（模型内部知识）
  • 向量数据库检索
  • Web 搜索

6.2 多级质量检查

每个步骤的输出都经过以下机制验证：

  • 文档相关性评分器：检索内容是否匹配问题？
  • 幻觉检测器：答案是否基于文档？有没有“编造”？
  • 答案质量评分器：最终答案是否真正解决问题？

6.3 自我纠错与多轮推理

系统根据评分结果自动调整路径：

  • 若文档质量差 → 重检索 / 网页搜索
  • 若答案不佳 → 重新生成 / 调整上下文
  • 可以在多个候选答案之间跳转、对比、合并

6.4 工作流完全可视化

借助 LangGraph，整个系统结构、状态转移、节点流动都可以图形化展示，易于调试与优化。

7. 总结

我们构建了一个完整的 Adaptive RAG 系统，具备以下核心能力：

模块

描述

查询路由器

识别查询复杂度，智能选择执行路径

检索器

从向量数据库或 Web 获取文档

生成器

基于文档和用户问题生成最终答案

幻觉检测器

确保答案与文档一致，避免胡编乱造

质量评分器

多轮判断答案是否达标，必要时重新生成

LangGraph 图

管理状态、控制流转，支持多跳、多路径、多状态决策

相比传统 RAG，Adaptive RAG 提供了更强的弹性、更高的准确率、更好的用户体验。

此外，模块高度解耦、可测试性强，便于后续迁移到其他 LLM、向量数据库或前端框架。

8. 局限性与下一步计划

尽管我们构建了一个强大的系统，但仍有以下待改进点：

1.成本问题

评分器和生成器每一步都调用 LLM，计算资源开销较大。

改进方向：

  • 替换部分评分器为轻量模型（如 BGE、DistilBERT）
  • 使用 rule-based 策略优化调用频次

2.Latency 延迟

多步流程带来响应延迟，影响交互体验。

改进方向：

  • 并行处理某些评分流程
  • 引入缓存机制：对相似问题/文档复用处理结果

3.查询分类准确率

当前查询分类器由 LLM 执行，准确性受限于提示词与模型能力。

改进方向：

  • 使用微调模型进行查询分类
  • 利用语义向量聚类 + 模板映射判断查询意图

9. 下一步计划

  1. 引入 Feedback 回路：让用户反馈反哺评分器
  2. 多语言支持：将系统泛化为可多语种适配架构
  3. 增强 UI：构建基于 Streamlit 或 Next.js 的可视化界面

你现在拥有了一个生产级别、支持自我评估与纠错的 Adaptive RAG 系统结构。可以在你的问答应用、企业知识管理、搜索系统中落地使用。

标签：vector at 

作者:fzmama , 分类:技术文章 , 浏览:14 , 评论:0

控制面板
您好，欢迎到访网站！
  查看权限
网站分类
最新留言