Agentic Rag Evolved：建立动态

Rifx.Online
Large Language Models , Natural Language Processing , AI Applications
16 Feb, 2025

大家好，欢迎回到我们的检索增强生成 (RAG) 中系列文章。如果这是您第一次来到这里，我的名字是 Joey O’Neill — 我是一名专注于微软Azure和生成式人工智能的云与定制应用顾问，拥有构建全栈RAG应用的经验。

在上一篇文章中，我们通过解决其缺陷和弱点来升级我们的简单RAG管道。我们引入了预处理和后处理技术，以精炼检索到的内容，确保大语言模型生成更准确和更有信息的响应。在今天的文章中，我们将跳过这些步骤，更加关注手头的核心任务：自主RAG管道。作为额外的挑战，您可以尝试将上一篇文章中的高级技术融入此管道，并观察结果如何演变。考虑到这一点，让我们深入了解本文的目标。

介绍与背景

当我们谈论“自主”RAG管道时，目标是设计一个系统，根据用户的输入动态选择最合适的检索和推理过程。这意味着管道在后台自适应其功能，以提供准确、相关的答案，而无需用户手动选择或推理采取哪条路径。

例如，在我的咨询工作中，我经常开发将由各种人使用的RAG管道，包括非技术用户 — 因此在测试期间，我们同样包括非技术领域专家来评估我们的输出。这些用户了解他们的数据，但对底层软件机制并不熟悉。我观察到的一个反复出现的问题是使用类似“总结此文档”的提示。这些类型的提示在标准RAG管道中造成挑战，因为它依赖于语义相似性或基于关键词的搜索从分块索引中检索数据。因此，管道经常提取文章的任意部分，阻止大语言模型形成内容的完整和准确的图像。

一方面，我们可以创建允许用户手动选择使用哪个管道的功能（摘要管道与默认RAG搜索），但这种方法缺乏真正用户友好的体验所需的无缝性和集成性。非技术用户通常不理解不同管道的细微差别，期望他们做出这样的决定可能会导致困惑和低效。相反，目标应该是允许用户自然地提出任何问题，而系统智能地确定最佳路径或RAG流，以提供最准确和相关的答案。这就是自主人工智能的力量所在 — 使用大语言模型不仅回答问题，还推理使用哪个管道或方法。这消除了摩擦，确保即使没有技术专长的用户也能够轻松获得最佳结果。通过自动化这个决策过程，我们可以提供一个直观、智能且能够处理多样化用户需求的产品。

在本教程中，我们将使用LangGraph构建一个实验性的ReAct自主RAG流，以解决上述问题。我们的解决方案将具有两个不同的RAG管道：一个专门用于总结单个文档，另一个作为默认的语义搜索管道，根据用户的查询天真地检索内容。大语言模型将负责推理调用哪个管道，顺畅地适应用户的意图，并根据检索到的内容提供答案。最后，我们将评估输出，识别此方法中的任何缺陷，并讨论潜在的增强措施，以改善未来迭代中的结果。和往常一样，完成的代码以及任何补充材料已推送到我上面的Git仓库。让我们开始吧！

环境设置

在本教程中，我们将使用以下版本的Python：

使用以下命令创建一个虚拟Python环境：

python -m venv .venv

使用以下命令激活虚拟环境：

对于Windows用户：

.venv\Scripts\activate

对于Mac用户：

source .venv/bin/activate

对于软件包，我们将使用以下内容：

chromadb==0.5.18
langchain-chroma==0.1.4
langchain-core==0.3.28
langchain-openai==0.2.6
langchain-text-splitters==0.3.2
langgraph==0.2.60
python-dotenv==1.0.1

使用pip直接安装软件包，或将其复制到requirements.txt文件中。

让我们创建一个环境文件（.env），以安全地存储环境变量，如API密钥和其他敏感信息。此文件通过允许您将凭据与主代码库分开来保护您的凭据安全。稍后，我们将其添加到.gitignore文件中，以确保在将项目上传到公共存储库时不会暴露任何密钥或秘密。

在.env文件中，创建一个变量以存储您的OpenAI API密钥，并确保使用与下面相同的变量名称。请注意，您需要从OpenAI获取自己的API密钥，并在您的账户上有余额。

OPENAI_API_KEY='<YOUR API KEY HERE>'

创建完成后，保存它，让我们继续进行.gitignore文件。

创建.gitignore文件，并包含以下内容并保存：

*.env
.env
.venv/
__pycache__/

最后，创建您将要处理的Python文件。您可以在.py文件中执行此操作，但我发现使用Python笔记本文件（.ipynb）更容易跟随。

现在，让我们将环境变量加载到Python文件中并保存。这是为了允许访问我们之前保存的OpenAI API密钥。

from dotenv import load_dotenv

load_dotenv()

创建完所有这些文件后，您应该有一个类似于以下内容的目录：

project-folder/
├── .env
├── .gitignore
├── main.ipynb
└── .venv/

我们的环境已经设置好并准备就绪！

设置我们的向量存储

接下来，我们将设置向量存储，它将作为我们RAG管道的上下文来源，存储我们分块、嵌入和索引的文档。在本教程中，我们将使用Chroma数据库的本地非持久性实例。Chroma是一个开源向量数据库，易于初始化，并且非常适合测试，特别是与LangChain和LangGraph配合使用。如果您更喜欢更持久的解决方案或希望使用其他技术，请随意自定义此步骤以满足您的需求。

首先，我们将加载我们的文章并将其存储为变量。对于本教程，我们将使用两篇不同的文档：2024年国情咨文和CBS新闻发布的关于Luigi Mangione的文章。这些文档可在上面的GitHub存储库中找到，但如果您愿意，可以使用自己的文章。

with open("../RAG_Docs/2024_state_of_the_union.txt") as f:
    state_of_the_union = f.read()

with open("../RAG_Docs/2024_12_10_Mangione_CBS_Article.txt") as f:
    mangione_cbs_article = f.read()

现在我们已经将文件的内容读取到我们的环境中，我们可以使用LangChain内置的CharacterTextSplitter将其分块为LangChain格式的文档。

from langchain_text_splitters import CharacterTextSplitter

text_splitter = CharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len
)

state_of_the_union_texts = text_splitter.create_documents([state_of_the_union])
mangione_cbs_article_texts = text_splitter.create_documents([mangione_cbs_article])

这个过程创建了两个LangChain文档的列表，提供了原始文章的有序和重叠分割。然而，这种分割使元数据为空。为了实现基于源文件的未来过滤，我们需要向这些文档添加元数据。幸运的是，这非常简单，可以通过以下方式实现：

for i, doc in enumerate(state_of_the_union_texts):
    doc.metadata = {
        'filename': '2024_state_of_the_union.txt',
        'chunk': i + 1
    }

for i, doc in enumerate(mangione_cbs_article_texts):
    doc.metadata = {
        'filename': '2024_12_10_Mangione_CBS_Article.txt',
        'chunk': i + 1
    }

处理完文档块并添加元数据后，它们现在可以存储在Chroma中。我们将初始化一个Chroma客户端，并使用它来创建LangChain向量存储。对于嵌入模型，我们将利用OpenAI的text-embedding-3-large模型。这确保在我们将文档存储在向量存储中时，其内容将自动编码并与之一起存储。

import chromadb
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

chroma_client = chromadb.Client()
collection = chroma_client.get_or_create_collection("test_collection")

embeddings = OpenAIEmbeddings(model="text-embedding

## ReAct代理模型概述

我不会深入探讨ReAct代理模型，但简而言之，它为大语言模型提供了可以根据描述选择的工具。这使得模型能够通过单独或顺序利用这些工具来逐步执行一个过程，观察每个动作的结果。通过这种迭代推理，模型确定实现所需目标的最佳路径——在这种情况下，提供最准确和相关的响应以满足用户的查询。



_图像来源: [https://newsletter.theaiedge.io/p/introduction-to-langchain-augmenting](https://newsletter.theaiedge.io/p/introduction-to-langchain-augmenting)_

考虑到这一点，下一步是创建函数——或者在LangGraph术语中，工具——以便大语言模型可以使用这些工具进行推理并决定采取哪些行动以检索回答用户查询所需的信息。在这种情况下，我们将定义两个主要工具，以表示两个不同的管道：一个用于提取单个文章的内容，另一个用于在所有文档中执行一般的RAG。此外，我们还将包括一个辅助工具，以返回所有文件名的列表，为模型提供可靠的方式来引用和调用单个文件。

让我们先创建辅助工具，并更清楚地理解我们所追求的工具语法。在生产级程序中，动态查询数据库，从元数据中提取独特的文件名并根据需要返回它们会更好。然而，由于我们在这里的重点是核心任务，我们将使用更简单、不可扩展的方法：返回一个静态的文件名字符串列表，以供大语言模型用于过滤数据库。

```python
from typing import List

def get_file_list() -> List[str]:
    """检索当前在数据库中可用且用户可以访问的文件名列表，作为字符串。使用此工具获取用户可以总结的文件。
    
    Args:
        None
    """
    return ['2024_state_of_the_union.txt', '2024_12_10_Mangione_CBS_Article.txt']

正如你所看到的，我们为返回值包含了类型注解，尽管这个函数不接受任何参数，但我们也会为这些参数提供类型注解。这有助于模型更好地理解调用工具时应提供的数据的预期格式以及它将收到的内容。此外，我们在函数开始时包含了详细的多行字符串。大语言模型使用这个描述来帮助决定何时、何地或是否在任何给定时间使用该工具。这充当了描述符，解释了函数的目的、使用上下文以及它接受的任何参数及其描述。在接下来的工具中，你将看到当我们定义包含参数的函数时，这一机制的实际应用。

对于下一个工具，我们将创建与前一个工具配对的RAG函数，以返回单个文章的内容以供总结。该函数接受一个文件名，从向量存储中过滤出所有与该文件名对应的块，然后返回内容块的连接字符串。我们包括一个错误检查，以确保文件名有效，并且大语言模型没有出错。如果文件名无效，该函数将引发错误，提示大语言模型需要推理出错误并重试。

请注意，这种方法在生产就绪的应用程序中有其局限性。具体来说，你应该考虑文件大小和长度相对于你所使用的大语言模型服务或模型的上下文窗口。处理上下文长度可能涉及截断文章、将其总结为较小的块，或实施其他策略来解决该问题。然而，对于这个例子，我们知道文件长度在可接受的范围内，因此在这里不会成为问题。

def get_file_content_by_name(filename: str) -> str:
    """检索并总结指定文档的内容。
    
    Args:
        filename: 一个字符串，表示要检索内容的文件名。
    """
    
    valid_files = ['2024_state_of_the_union.txt', '2024_12_10_Mangione_CBS_Article.txt']
    if (filename not in valid_files):
        return "错误：无效的文件名... 请重试..."

    content_list = vector_store.get(where={'filename': filename})['documents']
    content = "\n".join(content_list)

    return content

对于我们的最后一个工具，我们将创建一个通用的、默认的简单RAG上下文检索函数。该函数使用用户的查询在向量数据库中执行语义相似性搜索，然后返回所有相关块的连接字符串，以供大语言模型的上下文使用。

def default_rag(query: str) -> str:
    """根据查询在向量数据库中搜索相关的信息块。
    
    Args:
        query: 用户的查询，将用于返回相似的内容块。
    """
    results = vector_store.similarity_search(
        query,
        k=3
    )
    content_list = [f"* {res.page_content} [{res.metadata}]" for res in results]
    content = "\n".join(content_list)
    return content

现在我们已经创建了所有必要的工具供大语言模型选择，我们将实例化大语言模型并使用LangChain的内置bind_tools函数绑定这些工具。

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

tools = [get_file_list, get_file_content_by_name, default_rag]

llm_with_tools = llm.bind_tools(tools)

现在我们已经将工具绑定到大语言模型，我们可以使用它们创建主要的推理函数。上面图片中显示的“推理器”或“代理”节点。

推理函数（代理节点）

在LangGraph中，有一个状态的概念，这是一个在图中传递的对象，保留执行的值或状态，使得信息能够在节点之间转移。它本质上充当了一个中心值存储，可以在执行过程中被图操作。在我们的案例中，状态保存了用户的查询和来自先前调用的消息历史。它跟踪用户的查询、任何额外的推理和工具调用，最终根据提供的信息得出最终答案。我们使用类型字典来定义可以更改的值，并利用LangGraph内置的add_messages函数来附加新消息，而不是覆盖之前的消息。这确保了每次执行图时不会替换先前的状态。

from typing import Annotated, TypedDict
import operator
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages

class State(TypedDict):
    query: str
    messages: Annotated[list[AnyMessage], add_messages]

然后，我们将这个图状态用作推理器节点的参数。每次调用新工具或执行程序时，推理器节点将利用上面创建的状态进行推理并根据状态的内容采取行动。此函数使模型能够在调用绑定工具的大语言模型时进行推理，或者在它确定已收集到提供最佳可能响应的必要上下文时回答查询。

from langchain_core.messages import HumanMessage, SystemMessage

def reasoner(state):
    query = state["query"]
    messages = state["messages"]

    sys_msg = "你是一个有用的AI代理助手。你检查工具并根据用户的查询决定使用哪个工具。它们可以总结单个文件或在所有文件中搜索RAG配置。根据返回的上下文，你将尽力回答查询。"

    message = HumanMessage(query)
    messages.append(message)
    result = [llm_with_tools.invoke([sys_msg] + messages)]
    return {"messages": result}

现在我们已经将所有必要的功能设置到位，我们可以使用LangGraph设置实际的执行流程！

创建LangGraph

随着所有功能的设置，我们可以使用LangGraph创建一个执行图，允许状态通过各种函数传递。这使得大语言模型能够通过工具和过程进行推理，以便为用户的查询提供最佳答案。

我们首先通过StateGraph初始化工作流，传入我们之前创建的状态。接下来，我们定义节点。在这个例子中，有两个节点：推理器节点和工具节点。推理器节点利用我们之前定义的推理器函数，并充当代理的驱动程序。工具节点是LangGraph库中预定义的节点，管理我们创建的工具。它根据推理器的决策激活适当的工具。

一旦节点被定义，我们就为图添加边。首先，我们将推理器节点连接到图的START，因为它是第一个执行的。然后，我们从推理器节点添加一个条件边到预构建的tools_condition，该条件根据模型的推理决定激活哪个工具。最后，我们将工具节点连接回推理器节点。这个循环边允许大语言模型决定是否需要执行

workflow.add_edge(START, "reasoner")
workflow.add_conditional_edges(
    "reasoner",
    tools_condition
)
workflow.add_edge("tools", "reasoner")

reAct_graph = workflow.compile()

图形现在已设置，并且与之前的图表结构相对应。我们的代理已完全配置，接下来让我们进入下一部分并进行测试！

摘要管道测试

图形完全设置后，我们现在可以通过调用 reAct_graph 并传入我们之前构建的状态来执行它。此状态包括用户的查询和一个空消息列表。

response = reAct_graph.invoke({
    "query": "Summarize the article about Luigi Mangione.",
    "messages": []
})

在这个查询中，我们特别要求系统总结一篇关于 Luigi Mangione 的文章。预期的行为是基于摘要的 RAG 管道工具检索请求文章的内容。让我们使用以下代码打印出结果消息：

for m in response['messages']:
    m.pretty_print()

以下是输出：

================================ 人类消息 =================================
Summarize the article about Luigi Mangione.
================================== AI 消息 ==================================
工具调用：
  get_file_list (call_BpveEyBsyB19QlxgIGQUVRf4)
 调用 ID: call_BpveEyBsyB19QlxgIGQUVRf4
  参数：
================================= 工具消息 =================================
名称: get_file_list

["2024_state_of_the_union.txt", "2024_12_10_Mangione_CBS_Article.txt"]
================================ 人类消息 =================================
Summarize the article about Luigi Mangione.
================================== AI 消息 ==================================
工具调用：
  get_file_content_by_name (call_trTQvT2M9KLGAceAZImSLKeo)
 调用 ID: call_trTQvT2M9KLGAceAZImSLKeo
  参数：
    文件名: 2024_12_10_Mangione_CBS_Article.txt
================================= 工具消息 =================================
名称: get_file_content_by_name

<2024_12_10_Mangione_CBS_Article.txt 内容在这里，但我移除了以便更好查看>

================================ 人类消息 =================================
Summarize the article about Luigi Mangione.
================================== AI 消息 ==================================

Luigi Mangione, 26岁，因枪击 UnitedHealthcare 首席执行官 Brian Thompson 而被控谋杀。他在宾夕法尼亚州阿尔图纳被捕，因一名顾客在麦当劳认出了他。Mangione 一直通过使用信号阻断袋来逃避警方的追捕。被捕时，他身上发现多张假身份证、一支火器以及一张表达对医疗系统不满的便条，暗示了犯罪的潜在动机。

Mangione 的背景包括宾夕法尼亚大学的硕士学位，以及在 TrueCar 担任数据工程师的短暂工作经历。他与一个在房地产和医疗保健领域活跃的马里兰显赫家族有联系。他的家人对他的行为感到震惊。Mangione 目前正在对引渡到纽约提出异议，并对指控表示不认罪。警方担心他的行为对潜在的极端分子产生影响，考虑到他身上发现的手写便条的内容。

这个输出展示了代理为用户查询返回最佳响应的推理和工具使用的顺序。正如预期的那样，它成功执行了文件摘要管道，首先调用了 `get_file_list` 工具，然后利用该工具创建 `get_file_content_by_name` 工具调用的参数，从而检索请求的信息。代理能够在没有用户特定指示的情况下决定这一路径，除了查询之外。

### 默认 RAG 管道测试

接下来，让我们通过询问一个与文章摘要无关的一般问题来测试默认 RAG 管道：

```python
rag_response = reAct_graph.invoke({
    "query": "Who was the CEO of United Healthcare that was killed?",
    "messages": []
})

for m in rag_response['messages']:
    m.pretty_print()

以下是输出：

================================ 人类消息 =================================
Who was the CEO of United Healthcare that was killed?
================================== AI 消息 ==================================
工具调用：
  default_rag (call_LWuYcmNuGNT8TBFeiyU1zCPt)
 调用 ID: call_LWuYcmNuGNT8TBFeiyU1zCPt
  参数：
    查询: CEO of United Healthcare killed
================================= 工具消息 =================================
名称: default_rag

<内容已编辑以减少长度>
[{'chunk': 1, 'filename': '2024_12_10_Mangione_CBS_Article.txt'}]

<内容已编辑以减少长度>
[{'chunk': 8, 'filename': '2024_12_10_Mangione_CBS_Article.txt'}]

<内容已编辑以减少长度>
[{'chunk': 9, 'filename': '2024_12_10_Mangione_CBS_Article.txt'}]
================================ 人类消息 =================================
Who was the CEO of United Healthcare that was killed?
================================== AI 消息 ==================================

被杀的 UnitedHealthcare 首席执行官是 Brian Thompson。他在涉及名为 Luigi Mangione 的嫌疑人的事件中被枪击。

这个结果突显了代理如何处理不明确请求摘要的查询。它没有检索整篇文章的内容，而是高效地从向量存储中检索相关内容块，并通过利用 `default_rag` 工具与用户的查询提供简洁、准确的响应。

### ReAct 代理的局限性与最终备注

重要的是要记住，ReAct 代理方法依赖于大语言模型 (LLM) 来推理执行哪些工具以实现所需结果。这意味着它并不是 100% 可靠的方法。如果您的用例要求绝对可靠性，请考虑使用更确定性的方法。然而，LangGraph 生态系统的一个主要优势是提供 **LangSmith**，这是一个帮助监控生产环境中结果的独立服务。

另一个需要考虑的方面是通过模型微调提高准确性的潜力。微调有助于 LLM 更好地理解用户请求并确定在特定场景中调用哪些工具。例如，在为本文开发代码时，我遇到了一种情况，即询问“谁是 Luigi Mangione？”错误地触发了文章摘要管道，而不是默认 RAG 管道。虽然我通过细化工具描述解决了这个问题，但在某些情况下，微调模型可以提供更强大的解决方案，尤其是在工具具有重叠功能或描述模糊的情况下。

这结束了我们关于 **自主 RAG 与 LangGraph & Python** 的讨论。在下一篇文章中，我们将探讨如何创建一个具有 Web 套接字和流式传输的全栈 RAG 应用程序。

_如需额外帮助，这里有一些用于创建本文的文档和媒体链接：_ 
- [https://klu.ai/glossary/react-agent-model](https://klu.ai/glossary/react-agent-model) 
- [https://langchain-ai.github.io/langgraph/concepts/](https://langchain-ai.github.io/langgraph/concepts/) 
- [https://langchain-ai.github.io/langgraph/how-tos/create-react-agent/](https://langchain-ai.github.io/langgraph/how-tos/create-react-agent/) 
- [https://www.youtube.com/watch?v=pEMhPBQMNjg](https://www.youtube.com/watch?v=pEMhPBQMNjg)