Agentic RAG 系统的崛起

news/2025/1/15 12:58:11/文章来源:https://www.cnblogs.com/tunancbq/p/18672796

探秘智能检索新境界：Agentic RAG 系统的崛起

📖阅读时长：10分钟

🕙发布时间：2025-01-15

近日热文：全网最全的神经网络数学原理（代码和公式）直观解释
欢迎关注知乎和公众号的专栏内容
LLM架构专栏
知乎LLM专栏
知乎【柏企】
公众号【柏企科技说】【柏企阅文】
在人工智能飞速发展的当下，信息检索与处理技术不断革新。今天，让我们一同深入探究一项极具创新性的技术成果——基于 LangGraph 开发的 RAG Research Multi - Agent 工具，它代表着智能检索领域的重大突破。

一、引言：Naive RAG 与 Agentic RAG 的交锋

在项目推进过程中，传统的 Naive RAG 方法逐渐显露出诸多不足。首先，它缺乏对复杂查询的理解能力，无法将复杂问题拆解为多个可处理的子步骤，只能在单一层次上处理查询，难以通过对各步骤的分析得出统一结论。其次，Naive RAG 管道缺少响应验证步骤以及处理幻觉的机制，这使得它无法在出现错误时通过生成新响应来纠正问题。再者，Naive RAG 系统不支持根据工作流条件使用工具、调用外部 API 或与数据库进行交互。

为攻克这些难题，多智能体 RAG 研究系统应运而生。基于智能体的框架具备诸多强大功能：

路由与工具使用：路由智能体能够对用户查询进行分类，并将流程导向合适的节点或工具。这一功能支持基于上下文的决策，例如判断文档是否需要完整总结、是否需要更详细信息，或者问题是否超出范围等。
规划子步骤：复杂查询往往需要分解为更小、更易处理的步骤。借助智能体框架，从查询出发，能够生成一系列执行步骤，在探索查询不同方面的同时得出结论。例如，当查询要求比较文档中两个不同部分时，基于智能体的方法可以识别这种比较需求，分别检索两个来源，并在最终响应中进行对比分析。
反思与错误纠正：除了简单的响应生成，基于智能体的方法还能添加验证步骤，以解决潜在的幻觉、错误或无法准确回答用户查询的响应。这也使得集成包含人工参与的自我纠正机制成为可能，将人工输入融入自动化流程。在可靠性至关重要的企业应用中，这种功能使基于智能体的 RAG 系统成为更强大、可靠的解决方案。
共享全局状态：智能体工作流共享全局状态，简化了多步骤间的状态管理，这对于维持多智能体流程不同阶段的一致性至关重要。

二、项目全景：构建智能检索的蓝图

（一）系统流程概览

该系统的运行流程包含多个关键步骤：

分析与路由查询（自适应 RAG）：用户查询首先被分类并路由到相应节点。系统据此决定是进入下一步（“研究计划生成”）、向用户请求更多信息，还是在查询超出范围时立即响应。
研究计划生成：系统根据请求的复杂程度生成详细的、逐步的研究计划，并返回解决用户问题所需的具体步骤列表。
研究子图：针对研究计划中的每个步骤，系统调用子图。子图通过大语言模型（LLM）生成两个查询，然后使用集成检索器（结合相似性搜索、BM25 和最大边际相关（MMR）算法）检索与这些查询相关的文档。接着，通过基于 Cohere 的重排序步骤进行上下文压缩，最终为所有步骤提供得分最高的前 k 个相关文档。
生成步骤：工具基于相关文档，通过 LLM 生成答案。
幻觉检查（带人工参与的自我纠正 RAG）：系统会分析生成的答案，判断其是否得到提供的上下文支持，并涵盖所有方面。若检查不通过，图形工作流将被中断，提示用户生成修订答案或结束流程。

（二）向量数据库的构建

在构建向量数据库时，采用了基于段落的分块方法，借助 Docling 和 LangChain 库实现。具体过程如下：

文档解析：对于结构复杂的 PDF 文件，尤其是包含复杂表格布局的文档，选择合适的解析工具至关重要。许多库在处理这类文档时精度不足，而开源库 Docling 则表现出色。它能够实现简单高效的文档解析，并支持将文档导出为所需格式，可从多种常用文档格式（如 PDF、DOCX、PPTX、XLSX、图像、HTML、AsciiDoc 和 Markdown）读取并导出为 Markdown 和 JSON 格式。Docling 对 PDF 文档的理解全面，包括表格结构、阅读顺序和页面布局，同时还支持对扫描 PDF 进行光学字符识别（OCR）。使用以下代码进行文档格式转换：

from docling.document_converter import DocumentConverter
logger.info("Starting document processing.")
converter = DocumentConverter()
markdown_document = converter.convert(source).document.export_to_markdown()

文本分块：将 PDF 中的文本转换为 Markdown 格式后，基于标题并使用 MarkdownHeaderTextSplitter 进行分块，得到一个包含 332 个 Document 对象（LangChain Document）的列表。示例代码如下：

from langchain_text_splitters import MarkdownHeaderTextSplitter
headers_to_split_on = [("#", "Header 1"),("##", "Header 2")
]
markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on)
docs_list = markdown_splitter.split_text(markdown_document)

向量存储构建：使用 Chroma 构建向量数据库，将句子存储为向量嵌入，并在其中进行搜索。将持久化数据库存储在本地目录 “db_vector” 中，示例代码如下：

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
embd = OpenAIEmbeddings()
vectorstore_from_documents = Chroma.from_documents(documents=docs_list,collection_name="rag - chroma - google - v1",embedding=embd,persist_directory='db_vector'
)

（三）主图构建

系统包含两个关键图形：

研究子图：负责生成不同的查询，用于从向量数据库中检索和重新排名前 k 个文档。
主图：包含主要工作流程，如分析用户查询、生成完成任务所需的步骤、生成响应以及通过人工参与机制检查幻觉。

主图构建过程涉及定义多种状态和节点：

状态定义
- InputState：包含用户与智能体之间交换的消息列表。
- AgentState：继承自 InputState，包含路由器对用户查询的分类、研究计划中的步骤列表、智能体可参考的检索文档列表以及幻觉评分。

from dataclasses import dataclass, field
from typing import Annotated, Literal, TypedDict
from langchain_core.documents import Document
from langchain_core.messages import AnyMessage
from langgraph.graph import add_messages
from utils.utils import reduce_docs@dataclass(kw_only=True)
class InputState:messages: Annotated[list[AnyMessage], add_messages]@dataclass(kw_only=True)
class AgentState(InputState):router: Router = field(default_factory=lambda: Router(type="general", logic=""))steps: list[str] = field(default_factory=list)documents: Annotated[list[Document], reduce_docs] = field(default_factory=list)hallucination: GradeHallucinations = field(default_factory=lambda: GradeHallucinations(binary_score="0"))

节点定义与连接
- 分析与路由查询节点：该函数用于分析用户查询并更新 AgentState 中的路由器变量，根据查询分类确定下一步操作。例如，对于问题 “Retrieve the data center PUE efficiency value in Dublin in 2019”，可能返回如下分类结果：

{"logic": "This is a specific question about the environmental efficiency of a data center in Dublin in 2019, which relates to the Environmental Report.","type": "environmental"
}

- **其他节点**：还包括 “create_research_plan”（创建研究计划）、“ask_for_more_info”（请求更多信息）、“respond_to_general_query”（响应一般查询）、“conduct_research”（进行研究）、“respond”（响应）、“check_hallucinations”（检查幻觉）等节点，通过特定的条件和逻辑相互连接，形成完整的工作流。

主图的构建代码如下：

from langgraph.graph import END, START, StateGraph
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
builder = StateGraph(AgentState, input=InputState)
builder.add_node(analyze_and_route_query)
builder.add_edge(START, "analyze_and_route_query")
builder.add_conditional_edges("analyze_and_route_query", route_query)
builder.add_node(create_research_plan)
builder.add_node(ask_for_more_info)
builder.add_node(respond_to_general_query)
builder.add_node(conduct_research)
builder.add_node("respond", respond)
builder.add_node(check_hallucinations)
builder.add_conditional_edges("check_hallucinations", human_approval, {"END": END, "respond": "respond"})
builder.add_edge("create_research_plan", "conduct_research")
builder.add_conditional_edges("conduct_research", check_finished)
builder.add_edge("respond", "check_hallucinations")
graph = builder.compile(checkpointer=checkpointer)

（四）应用程序构建

在应用程序层面，定义了异步函数来处理查询。用户输入查询后，系统会按照主图定义的流程进行处理，并在必要时与用户进行交互。例如，当幻觉检查不通过时，提示用户决定是否重新生成响应。核心代码如下：

async def process_query(query):inputState = InputState(messages=query)async for c, metadata in graph.astream(input=inputState, stream_mode="messages", config=thread):if c.additional_kwargs.get("tool_calls"):print(c.additional_kwargs.get("tool_calls")[0]["function"].get("arguments"), end="", flush=True)if c.content:time.sleep(0.05)print(c.content, end="", flush=True)if len(graph.get_state(thread)[-1]) > 0:if len(graph.get_state(thread)[-1][0].interrupts) > 0:response = input("\nThe response may contain uncertain information. Retry the generation? If yes, press 'y': ")if response.lower() == 'y':async for c, metadata in graph.astream(Command(resume=response), stream_mode="messages", config=thread):if c.additional_kwargs.get("tool_calls"):print(c.additional_kwargs.get("tool_calls")[0]["function"].get("arguments"), end="")if c.content:time.sleep(0.05)print(c.content, end="", flush=True)

三、成果见证：实力验证与对比

为检验系统性能，使用了一份关于谷歌环境可持续性战略的年度报告进行测试。以复杂问题 “Retrieve the data center PUE efficiency values in Singapore 2nd facility in 2019 and 2022. Also retrieve regional average CFE in Asia pacific in 2023” 为例，系统成功生成了正确的结果，通过了幻觉检查。系统生成的步骤如下：

“Look up the PUE efficiency values for the Singapore 2nd facility for the years 2019 and 2022.”
“Find the regional average CFE for the Asia Pacific region in 2023.”
最终生成的文本为：
“The Power Usage Effectiveness (PUE) for the Singapore 2nd facility in 2019 is not available, as the data for that year is not provided. However, the PUE for 2022 is 1.21.”
“The regional average Carbon - Free Energy (CFE) in the Asia Pacific for 2023 is 12%.”

与之对比，当在 ChatGPT 上上传相同的 PDF 文件并提出相同查询时，ChatGPT 返回的结果存在错误且出现幻觉。这充分凸显了本系统在复杂信息检索和准确回答方面的优势。