不需要 RAG！手把手教你构建问答 Agent（只需 30 分钟）

显示全部楼层

还记得大家曾经疯狂追捧的检索增强生成（RAG）吗？当时为了能从公司文档中回答问题，我们不得不搭建复杂的管道，建立向量数据库、切分文档、生成嵌入向量（embeddings）。现在，你可能要重新思考这一切。

本文的目的是探讨 2025 年检索增强生成（RAG）的现状，并深入讨论“ RAG 是否已死”这一争议话题。我们会构建一个简单的替代方案，并与传统的 RAG 方法进行对比。另外，我们还会讨论“思考型模型 vs 非思考型模型”的话题。

为了说明这一点，我们将在短短30 分钟内搭建一个简单的问答 Agent，完全跳过 RAG，直接使用搜索 API 如 Tavily 和大上下文窗口。你猜怎么着？这种方法不仅比传统 RAG 更简单、更便宜，而且效果往往还更好。

接下来，我会告诉你为什么这种方法可能会成为未来基于文档的问答系统的主流。

本文涉及的所有源代码：https://github.com/javiramos1/qagent

RAG 并没有死——但到了 2025 年，它不再应该是你的首选方案。请从简单的方案开始！

本文的目的并不是批判RAG（它仍有其适用场景），而是通过实际案例，探讨 2025 年出现的新趋势。我们将构建一个传统上会使用 RAG 的系统，但采用一种完全不同的、越来越受欢迎的方法。

RAG：优势、劣势与复杂性

究竟什么是 RAG？

你可以把RAG想象成一个非常聪明的图书管理员。当有人提出问题时，它会：

将文档切分成小段（因为模型的上下文窗口有限）
将每段文本转化为嵌入向量（数学表示）
将这些向量存储到向量数据库（相当于你的数字图书馆）
当问题出现时，也将问题转化为嵌入向量
通过向量相似性搜索（vector similarity search）找到最相似的文本片段
将这些片段与问题一起输入大模型
根据检索到的上下文生成答案

下面是一个简化版的 RAG 实现示例：

# 传统RAG管道（简化版）
defrag_pipeline(question):
 # 第一步：将问题转为嵌入向量
  question_embedding = embedding_model.encode(question)

 # 第二步：在向量数据库中搜索相似片段
  similar_chunks = vector_db.similarity_search(
    question_embedding,
    top_k=5
  )

 # 第三步：将检索到的片段拼接成上下文
  context ="\n".join([chunk.contentforchunkinsimilar_chunks])

 # 第四步：生成答案
  prompt =f"Context:{context}\n\nQuestion:{question}\nAnswer:"
 returnllm.generate(prompt)

为何 RAG 迅速走红？

回到 2020 年至 2023 年间，RAG（检索增强生成）技术的兴起有着充分的理由：

上下文窗口极小：当时的 GPT-3 仅支持 4K 个 token 的上下文长度，这正是 RAG 被提出的最主要原因。
token 成本高昂：处理大篇幅文档的成本非常高。
“中间遗忘”问题：模型无法很好地处理长上下文，容易忽略中间部分的信息。
对实时数据的需求：模型训练数据存在截止日期，无法及时更新。

RAG 通过有选择地引入相关信息，解决了上述问题，并允许在过时的模型基础上引入最新的数据。

一个简单的 RAG 示例

假设你有一套公司文档，有人问：“我们 API 的认证功能怎么设置？”

传统的 RAG 方法：

将文档切分成约 500 字的小段落；
找出与认证最相关的 3 到 5 个段落；
仅将这些段落输入 LLM；
生成答案。

在上下文受限且成本高昂的时代，这种方法效果非常好，但是……

RAG 背后不为人知的难题

虽然理论上 RAG 听起来优雅简单，但实际落地和维护的复杂程度远超大多数教程所描述的。

我们来总结一下实际中面临的一些挑战：

基础设施的噩梦

部署复杂：RAG 需要搭建向量数据库、embedding 管道、文档切分策略、监控和扩展机制，整体复杂度非常高。
成本高昂：向量数据库和 RAG 管道的运行维护成本极高，一旦数据量变大，每月成本可能迅速飙升至数千美元。
厂商锁定风险：你需要在 Pinecone、Weaviate、Chroma 等商业方案或昂贵的自建方案之间做出选择，容易陷入厂商锁定的困境。

文档切分的难题

文档切分（Chunking）)是指将文档拆分成更小的语义单元，以便存储到向量数据库中。切分策略众多，每种策略都有优缺点，没有完美的方案，主要问题包括：

上下文丢失：文档被拆分后，逐步说明的步骤被打乱，代码与解释被割裂。
无完美策略：无论是固定长度切分、语义切分还是按段落切分，都存在明显缺陷。
交叉引用断裂：相关联的章节被拆散，重要的关联信息丢失。

相似性搜索的假象

向量数据库使用的相似性搜索在概念上类似于 Elasticsearch，但专门针对 LLM 的 embedding 进行了优化。尽管 Elasticsearch 经过多年实战验证，但向量数据库相对较新，仍在不断发展完善中。正如我们稍后将讨论的，RAG 的一个替代方案是利用现有的 Elasticsearch 基础设施，通过构建一个工具来查询 Elasticsearch 进行上下文检索。这种方法对那些已经拥有 Elasticsearch 集群的组织尤其有价值，因为它避免了额外引入向量数据库等新基础设施的麻烦。

数学相似 ≠ 实际相关：高相似度的数学分数并不能保证结果真正有用。
查询与文档不匹配：用户提出的问题与文档中的表述方式可能存在差异，导致检索效果不佳。
缺乏必要上下文：embedding 过程可能丢失一些重要的限定信息，导致检索结果不完整或不准确。

持续维护的噩梦

文档更新问题：每次文档发生变化，都必须重新进行分块（chunking）、重新生成嵌入（embedding）、重新索引（indexing），整个流程都要重跑一遍！
模型不兼容性：不同模型生成的嵌入向量无法混用；升级模型意味着必须完全重建索引！这使得切换模型变得非常困难！
性能衰减：随着向量数据库规模扩大，查询速度会逐渐变慢。当数据量达到一定规模后，延迟问题将变得明显。

扩展性面临的现实挑战

成本指数级增长：基础设施需求的增长速度远超文档数量的增长速度。
边际效益递减：文档越多，搜索质量反而可能下降。
运维负担加重：大规模部署需要专门的 DevOps 团队支持。

常见问题与失败场景

“中间内容丢失”问题：大模型经常忽略搜索结果中间位置的内容块。
数据陈旧：向量搜索缺乏对数据时效性的感知。
上下文碎片化：复杂问题的答案往往被拆分到多个内容块中，导致信息割裂。

现实情况总结：

成本指数级增长：基础设施需求增长速度远超文档数量增长速度（参见：向量数据库扩展挑战：https://www.pinecone.io/learn/vector-database-scaling/）
边际效益递减：文档越多，搜索质量反而可能下降。
运维复杂性增加：需要专门的 DevOps 技术团队支持。

正是由于上述这些复杂性，我们正在构建的这种以简单搜索为先的方案才逐渐受到青睐。核心思想是直接访问数据源，而不是构建复杂的 RAG（检索增强生成）流水线。

与其管理庞大的基础设施，我们更倾向于使用成熟的搜索 API 和大上下文窗口（context window），以更低的复杂度获得更好的效果。

2025 年的 RAG：技术格局已发生巨大变化

上下文窗口的革命性突破

过去 18 个月发生的变化彻底改变了游戏规则：

等等，真的吗？100 万 token 的上下文窗口只需0.075 美元？这大约相当于 75 万字，足以容纳整本书的内容！

**Google ** 凭借其Gemini系列模型，正引领着这波创新浪潮。得益于专用的TPU芯片和庞大的数据中心基础设施，Google 已将上下文窗口扩展至500 万 token，同时将每个 token 的成本控制在了令人惊叹的低水平。然而，长上下文窗口也存在一些与 RAG 类似的扩展性问题，真正的变化在于直接访问数据源，而非构建复杂的流水线。

例如Gemini Flash模型，为许多 LLM 应用场景提供了快速、经济且简单的解决方案。此外，最新的 Gemini 版本在各大主流基准测试中已超越了大多数竞争对手。

正如我们稍后会详细探讨的，我个人非常推崇Gemini 2.0 Flash模型——它速度极快、价格实惠，并能轻松处理大规模上下文窗口。

最新消息：Google 刚刚发布了Gemini Flash 2.5模型，这是一个非常出色的替代方案，尽管价格略高一些。

这些变化为我们带来了根本性的范式转变。我们不再需要将多个数据源的数据加载到向量数据库中，也不再需要维护复杂的 RAG 流水线，而是可以直接构建工具，供 LLM 和 Agent 直接调用。核心思想是利用组织内部已有的成熟解决方案，比如文档网站或ElasticSearch。

这种方法远比传统的 RAG 方案简单得多。如今，快速、经济的 LLM 模型和大规模上下文窗口的出现，使得成本和延迟不再是制约因素。

MCP 的出现彻底改变了 Agent 与工具之间的交互方式。过去，每个工具连接都需要手动配置，而 MCP 则实现了工具的自动发现，让 Agent 能够动态地找到并连接数据库、API 和企业系统。

这催生了真正意义上的非确定性 Agent 工作流，其神奇之处在于：工具本身也能成为 Agent ，递归地调用其他工具。试想一下，一个 Agent 自动发现了数据库工具，用它查询客户数据，然后又自动找到并调用电子邮件 API 工具发送个性化信息——整个过程无需预先定义的工作流。

最终形成的是一种自组织的 Agent 生态系统，它们根据可用的能力动态适应，而非依赖于僵化的程序或确定性的工作流。过去需要复杂的手动设置才能实现的功能，现在通过工具之间的交互自然涌现，使得复杂的多 Agent 系统对任何开发者都变得触手可及。

获取外部数据的关键工具

如前所述，RAG 之外的主要替代方案是构建能够直接查询数据源的工具，提取相关内容并作为上下文提供给 LLM。

提示：这些工具本身也可以是 Agent，形成层次结构，每个工具封装复杂性并暴露出简单易发现的接口。

另一个重要的优势是，LLM 可以根据输入内容动态决定使用哪些工具，从而智能地选择工具并配置参数，创造真正的非确定性工作流。

下面我们来看看一些用于从现有系统中提取上下文的工具示例：

公共搜索工具

Tavily：使 LLM 能够搜索预先索引的公共网站以获取相关上下文 → 我们将在 Agent 中使用这个工具！

企业搜索工具

Elasticsearch及类似平台：直接连接现有的搜索索引，检索查询所需的相关数据。
充分利用现有基础设施：无需重复建设已有的系统。
搜索引擎 vs 向量数据库：传统搜索引擎（如 Elasticsearch）已存在多年，在检索相关信息方面往往优于向量数据库。

数据库集成工具

SQL 生成工具：利用 LLM 生成 SQL 查询，直接从数据库中检索结构化数据。这种方法虽然复杂，但完全可行。

与传统 RAG 相比，搜索优先方案的优势

搜索优先方案成本更低，维护起来也更简单
数据始终保持最新——无需更新过时的索引或重新生成嵌入向量
不存在“中间内容丢失”问题——搜索引擎总是优先返回最相关的内容
上下文相关性更高——搜索算法本身就针对查询相关性进行了优化
迭代速度更快——文档更新时无需重新生成嵌入向量
调试更简单——清晰可见检索到的内容及其原因

然而，在以下场景中，RAG 仍然是合理的选择：

超大规模企业数据集（100GB 以上）
文档片段级别的精细权限控制
离线或隔离网络环境
超高查询量场景（每天超过 10 万次查询）
涉及复杂文档关系的数据

关键结论：RAG 不应作为你的首选方案，仅在特定场景下才值得考虑，比如对隐私有严格要求，或数据量巨大且难以实时访问的情况。

实践环节：构建一个以搜索为核心的问答 Agent

好了，理论讲得够多了。接下来我们动手构建一个实际案例，证明搜索优先方案比传统 RAG 更好用。

我们将创建一个 ** 特定领域的问答 Agent **，它具备以下特点：

仅搜索经过批准的组织文档网站 →安全防护
使用搜索 API 而非向量数据库
必要时回退到全面的网页抓取
提供透明的来源引用
可选地对搜索结果进行摘要，以降低成本和延迟
成本仅为传统 RAG 系统的一小部分

你可以将其理解为“去掉 R 的 RAG”——我们不再使用预处理的文档片段，而是直接使用实时、全面的搜索结果来增强生成效果。

传统上，当我们受限于仅 4K 的 token 窗口时，开发者通常会将所有文档加载到向量数据库中，并使用RAG来提取相关上下文。这种方案复杂且难以维护，如下图所示：

而我们的方案则完全不同：我们直接从源头获取数据，使用工具，充分利用已经成熟且经过验证的搜索引擎进行数据提取。

说明：此方案仅适用于搜索引擎已索引的公开网站。如果你需要处理内部文档或私有数据，我也可以演示如何使用ElasticSearch进行内部上下文检索，甚至使用 SQL 实时检索结构化数据。

为何此方案优于传统 RAG？

具体实现原理

1. 初始化阶段： Agent 启动时，会读取一个 CSV 文件，其中包含：

待搜索的网站 URL
网站所属的领域或类别（指主题或学科领域，而非网站域名）
网站的描述信息，以帮助 LLM 理解何时应使用哪个网站

2. 查询处理：当用户提交查询时，LLM 会对查询内容进行分析，智能地选择最适合回答该问题的网站。所有查询均采用异步方式处理。

3. 搜索执行：选定的网站会被传递给我们的搜索工具，该工具使用Tavily在这些预先批准的域名范围内快速、精准地执行搜索。

4. 动态决策：根据搜索结果，LLM 会判断当前信息是否足以给出完整答案：

足够：立即返回答案
不足够：可能会调用网页抓取工具，从特定页面获取更多细节信息

这一方案充分体现了真正 Agent 设计的精妙之处——结构简单，却功能强大。我们并未采用死板的预设路径，而是允许 Agent 根据每次查询的具体上下文，自主决定下一步行动和工具选择。为实现这一目标，我们使用了一个简单的非思考型模型，遵循著名的 ReACT 框架，让 Agent 在循环中自主思考并决定下一步行动。这是一种让非思考型模型实现“思考”的简单方法。

我们采用了双层策略：

快速搜索（占 90% 的查询）：使用 Tavily 在批准的域名内快速搜索并总结结果。
深度抓取（占 10% 的查询）：当搜索不足以回答问题时，抓取整个网页内容。

ReACT 与思考型模型：为何我们选择速度而非“智能”

这里有一个颇具争议的观点：我们刻意选择了 Gemini 2.0 Flash，而非 OpenAI 的 o3 或 Gemini 2.5 Pro 等“思考型”模型。为什么？

ReACT 框架以仅 1/200 的成本实现了结构化的思考流程：

用户查询 → Agent 思考 → 选择工具 → 执行 → 观察 → 回答
 ↑    ↑      ↑    ↑   ↑   ↑
 输入  ReACT逻辑   工具选择  搜索  结果  答案

我更倾向于使用非思考型模型搭配 ReACT 框架的主要原因是可扩展性。在许多情况下，用户查询只需简单回答，无需复杂推理，使用思考型模型反而显得“大材小用”。非思考型模型在这些场景下能快速、经济地提供解决方案。而对于复杂问题，ReACT 框架则通过循环实现思维链，模拟思考型模型的推理能力。

对于搜索 - 回答类的工作流程，这种方法远比模型内部的思维链推理更高效。根据我的经验，当使用多个工具且每个工具都有多个输入时，这种方案的表现优于思维链型 LLM。

Agent 遵循一个明确的 ReACT 提示，每个工具的描述都清晰定义了输入和输出。当你运行应用时，可以看到 Agent 如何通过链式思考循环进行多次搜索，或决定是否使用网页抓取工具，所有这些决策完全由 Agent 自主完成，无需人工干预！

ReACT 提示词在使用长上下文窗口时，也能有效防止模型产生幻觉（hallucinations）！

模型选择

针对问答 Agent ，gemini-2.0-flash是一个非常理想的选择，它在速度、成本效益和处理较大上下文窗口的能力之间达到了绝佳的平衡。其免费额度也相当慷慨，每分钟可处理 15 次请求（RPM）、每分钟 100 万 token（TPM）以及每天 1500 次请求（RPD），在考虑付费方案之前，这足以满足大多数应用场景的需求。我在本仓库中使用的正是这个模型。

如果你的问答 Agent 需要更强大的推理能力和更“智能”的响应，gemini-2.5-flash-preview-05-20则是更好的选择。该模型具备更出色的性能和更强的分析能力，非常适合处理复杂的查询。不过，这种增强的能力也意味着更严格的免费额度限制（例如每分钟 10 次请求、每分钟 25 万 token 、每天 500 次请求）以及略高的价格。如果你拥有付费账号并且对准确性要求较高，可以选择这个模型。

另一方面，如果你的主要关注点是高吞吐量、成本效益，并希望在简单、高频的问答场景中最大化免费额度的利用，gemini-2.0-flash-lite则是最合适的替代方案。尽管它在复杂推理能力上稍逊一筹，但其更高的额度限制（例如每分钟 30 次请求、每分钟 100 万 token 、每天 1500 次请求）和较低的价格使其能够支持更大的交互量。

代码详解

接下来我们一起来看看具体的实现方式。即使你对 Python 或 LangChain 还不熟悉，我也会一步一步地详细讲解。

我们选择使用 LangChain，是因为它提供了一个统一的、与模型无关的接口，极大地简化了复杂 LLM 应用的开发过程。LangChain 擅长通过“链（Chains）”和“ Agent （Agents）”组织多步骤工作流，允许 LLM 调用各种“工具”，并且能轻松管理对话记忆。这种抽象层不仅减少了大量样板代码，加快了开发速度，还提升了应用的灵活性和未来扩展性。此外，通过与 LangSmith 等工具的集成，LangChain 还提供了关键的调试能力。

注：Pydantic AI 是 Langchain 的替代品，它更容易上手。

架构概览：文件如何协同工作

本仓库采用了一个非常简单的分层架构：

项目结构
├── main.py      # FastAPI Web服务器（程序入口）
├── qa_agent.py    # Agent 核心逻辑（任务协调器）
├── search_tool.py  # 搜索功能模块
├── scraping_tool.py # 网页抓取功能模块
└── sites_data.csv  # 域名配置文件

工作流程：

main.py→ 接收用户的 HTTP 请求
qa_agent.py→ 决定调用哪些工具，并协调生成响应
search_tool.py/scraping_tool.py→ 实际执行信息查找任务
返回到qa_agent.py→ 整合结果并生成最终答案
返回到main.py→ 将响应返回给用户

你可以把这个流程想象成一家餐厅：main.py是服务员，qa_agent.py是决定做什么菜的主厨，而工具模块则是专门的厨房设备。

1. 搜索工具（`search_tool.py`）

我们先从搜索功能开始。该工具连接到 Tavily 搜索 API，用于查找相关信息。

"""
使用Tavily进行特定域名的网页搜索工具
"""
importlogging
fromtypingimportList, Optional, Type, Any
frompydanticimportBaseModel, Field, ConfigDict
fromlangchain.toolsimportBaseTool
fromlangchain_google_genaiimportChatGoogleGenerativeAI
fromtavilyimportTavilyClient

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


classTavilySearchInput(BaseModel):
  query: str = Field(description="包含相关关键词的搜索查询")
  sites: List[str] = Field(
    description="要搜索的网站域名列表（例如：['docs.langchain.com']）"
  )
  max_results: Optional[int] = Field(
    default=None, description="返回的最大搜索结果数量"
  )
  depth: Optional[str] = Field(
    default=None, description="搜索深度：'basic'（基础）或'advanced'（高级）"
  )


classTavilyDomainSearchTool(BaseTool):
 """使用Tavily搜索特定网站域名的工具"""

  name: str ="search_documentation"
  description: str ="""使用Tavily网页搜索引擎搜索文档类网站。

  必要参数：
  - query (字符串)：包含相关关键词的搜索查询，即你想要查找的内容
  - sites (列表)：要搜索的网站域名列表（例如：['docs.langchain.com', 'fastapi.tiangolo.com']）

  可选参数：
  - max_results (整数)：返回的最大搜索结果数量（默认值：10）
  - depth (字符串)：搜索深度，'basic'用于快速搜索，'advanced'用于全面搜索（默认值：'basic'）

  使用指南：
  1. 根据用户问题构建关键词丰富的搜索查询
  2. 根据提及的技术选择相关的网站域名
  3. 快速获取答案时使用'basic'深度，深入研究时使用'advanced'深度
  4. 根据所需答案的详细程度调整max_results参数

  示例：
  - 快速搜索：query="LangChain自定义工具", sites=["docs.langchain.com"], depth="basic", max_results=5
  - 全面搜索：query="FastAPI认证中间件", sites=["fastapi.tiangolo.com"], depth="advanced", max_results=15

  最佳实践：
  - 在查询中包含技术术语和框架名称
  - 根据问题上下文选择合适的域名
  - 优先选择官方文档网站而非第三方来源
  - 使用具体的查询词而非宽泛的术语，以获得更精准的结果
  """
  args_schema: Type[BaseModel] = TavilySearchInput

  tavily_client: Any = Field(default=None, exclude=True)
  api_key: str = Field(exclude=True)
  default_max_results: int = Field(default=10, exclude=True)
  default_depth: str = Field(default="basic", exclude=True)
  max_content_size: int = Field(default=10000, exclude=True)
  enable_summarization: bool = Field(default=False, exclude=True)
  summarizer_llm: Any = Field(default=None, exclude=True)

  model_config = ConfigDict(arbitrary_types_allowed=True)

 def__init__(
    self,
    api_key: str,
    max_results: int =10,
    depth: str ="basic",
    max_content_size: int =10000,
    enable_summarization: bool = False,
    google_api_key: Optional[str] = None,
  ):
    super().__init__(
      api_key=api_key,
      default_max_results=max_results,
      default_depth=depth,
      max_content_size=max_content_size,
      enable_summarization=enable_summarization,
      args_schema=TavilySearchInput,
    )

   ifnotapi_key:
     raiseValueError("必须提供TAVILY_API_KEY")

    object.__setattr__(self,"tavily_client", TavilyClient(api_key=api_key))

   ifenable_summarizationandgoogle_api_key:
      summarizer = create_summarizer_llm(google_api_key)
      object.__setattr__(self,"summarizer_llm", summarizer)
      logger.info("已启用Gemini Flash-Lite进行搜索结果摘要")
   elifenable_summarization:
      logger.warning("未提供google_api_key，摘要功能已禁用")
      object.__setattr__(self,"enable_summarization",False)

    logger.info(
     f"Tavily搜索工具初始化完成（摘要功能：{'已启用'ifself.enable_summarizationelse'未启用'}）"
    )

 asyncdef_search_async(self, query: str, sites: List[str], max_results: int = None, depth: str = None)-> str:
   """异步执行搜索请求"""
   try:
      final_max_results = max_resultsorself.default_max_results
      final_depth = depthorself.default_depth

      logger.info(f"🔍 正在搜索：'{query}'，目标网站：{sites}")
      logger.info(
       f"搜索参数：max_results={final_max_results}, depth={final_depth}"
      )

     # 注意：TavilyClient目前没有异步方法，因此使用线程执行
      search_results =awaitasyncio.to_thread(
        self.tavily_client.search,
        query=query,
        max_results=final_max_results,
        search_depth=final_depth,
        include_domains=sites,
      )

      logger.info(f" 已收到{len(search_results.get('results', []))}条搜索结果")

ifnotsearch_results.get("results"):
  logger.warning("未返回任何搜索结果")
 return"未找到相关结果，请尝试其他搜索关键词或检查网站是否可访问。"

formatted_results = format_search_results(
  search_results["results"][:final_max_results], self.max_content_size
)
final_result ="\n".join(formatted_results)

logger.info(
 f"已处理{len(search_results['results'])}条结果，返回内容长度为{len(final_result)}个字符"
)

ifself.enable_summarizationandself.summarizer_llm:
 try:
    logger.info("正在对结果进行智能摘要...")
    summarized_result =awaitself._summarize_results_async(final_result, query)
    reduction = round(
      (1- len(summarized_result) / len(final_result)) *100
    )
    logger.info(
     f"摘要压缩效果：{len(final_result)}→{len(summarized_result)}字符（减少了{reduction}%）"
    )
   returnsummarized_result
 exceptExceptionase:
    logger.error(
     f"摘要生成失败：{e}，返回原始搜索结果。"
    )

returnfinal_result

exceptExceptionase:
  error_msg =f"搜索过程中出现错误：{str(e)}"
  logger.error(error_msg)
 returnerror_msg

asyncdef_summarize_results_async(self, search_results: str, original_query: str)-> str:
 """使用LLM异步生成搜索结果摘要"""
 try:
    prompt = create_summary_prompt(search_results, original_query)
    response =awaitasyncio.to_thread(self.summarizer_llm.invoke, prompt)
   returnresponse.content
 exceptExceptionase:
    logger.error(f"LLM摘要生成失败：{e}")
   returnsearch_results

def_run(
  self, query: str, sites: List[str], max_results: int = None, depth: str = None
)-> str:
 """执行搜索（同步方式）"""
 returnasyncio.run(self._search_async(query, sites, max_results, depth))

asyncdef_arun(
  self, query: str, sites: List[str], max_results: int = None, depth: str = None
)-> str:
 """执行搜索（异步方式）"""
 returnawaitself._search_async(query, sites, max_results, depth)

代码解析与技术要点：

LangChain BaseTool：继承自该类以创建供 Agent 调用的工具，下一篇文章中我们将把它转化为一个MCP服务端。
工具的description（描述）非常关键，它决定了 Agent 何时以及如何调用该工具，务必精准清晰。
Pydantic 模型：用于工具输入参数的类型校验，确保 Agent 传入正确的参数。
域名限制：通过include_domains参数实现安全防护，确保仅搜索经过批准的网站。
智能摘要：可选的 AI 驱动结果压缩，使用 Gemini Flash-Lite 模型降低 token 成本。
异步支持：通过实现_arun方法支持异步执行。
错误处理：完善的日志记录和优雅的错误恢复机制。
灵活配置：支持不同的搜索深度和结果数量限制。

2. 网页抓取工具（`scraping_tool.py`）：可选

当搜索结果不足以满足需求时，该工具可抓取整个网页内容，以获取更全面的信息。

"""
使用Chromium浏览器进行动态网页内容抓取的工具
"""

importlogging
importasyncio
fromtypingimportList, Type

frompydanticimportBaseModel, Field, ConfigDict
fromlangchain.toolsimportBaseTool
fromlangchain_community.document_loadersimportAsyncChromiumLoader
fromlangchain_community.document_transformersimportBeautifulSoupTransformer

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


defget_default_tags()-> List[str]:
 """获取默认用于网页抓取的HTML标签"""
 return["p","li","div","a","span","h1","h2","h3","h4","h5","h6"]


classWebScrapingInput(BaseModel):
  url: str = Field(description="待抓取的网页URL")
  tags_to_extract: List[str] = Field(
    default_factory=get_default_tags, description="需要提取的HTML标签"
  )


classWebScrapingTool(BaseTool):
 """当搜索结果不足时，使用Chromium浏览器抓取完整网页内容"""

  name: str ="scrape_website"
  description: str ="""使用Chromium浏览器抓取完整网页内容，以获取全面的信息。

  必要参数：
  - url (字符串)：待抓取网页的完整URL（必须包含 https:// 或 http://）"""

可选参数：
- tags_to_extract（列表）：指定要从HTML中提取内容的标签。
 默认值：["p","li","div","a","span","h1","h2","h3","h4","h5","h6"]
 自定义示例：
  - 提取代码示例：["pre","code"]
  - 提取表格内容：["table","tr","td"]

适用场景：
- 搜索结果不完整或信息不足时
- 需要完整页面内容，包括代码示例
- 页面包含动态JavaScript内容，搜索未能捕获
- 需要特定的格式或结构，而搜索无法满足

示例用法：
- 基础网页抓取：url="https://docs.langchain.com/docs/modules/agents"
- 代码示例抓取：url="https://fastapi.tiangolo.com/tutorial/"，tags_to_extract=["pre","code","p"]
- 表格内容抓取：url="https://docs.python.org/3/library/"，tags_to_extract=["table","tr","td","th"]

最佳实践：
- 仅在search_documentation提供的信息不足时使用
- 优先使用之前搜索结果中的URL，以确保内容相关性
- 使用特定标签提取目标内容（处理速度更快）
- 注意：网页抓取速度比搜索慢约3-10倍，谨慎使用以保证性能

限制条件：
- 为避免过多的token消耗，内容会在设定的长度限制处截断
- 某些网站可能会阻止自动化抓取
- 速度慢于搜索，仅在搜索不足时使用

args_schema: Type[BaseModel] = WebScrapingInput

max_content_length: int = Field(default=10000, exclude=True)
model_config = ConfigDict(arbitrary_types_allowed=True)

def__init__(self, max_content_length: int =10000):
  super().__init__(
    max_content_length=max_content_length, args_schema=WebScrapingInput
  )

asyncdef_process_scraping(
  self, url: str, tags_to_extract: List[str] = None, is_async: bool = True
)-> str:
 """同步与异步抓取的通用处理逻辑"""
 try:
   iftags_to_extractisNone:
      tags_to_extract = get_default_tags()

    loader = AsyncChromiumLoader([url])

   ifis_async:
      html_docs =awaitasyncio.to_thread(loader.load)
   else:
      html_docs = loader.load()

   ifnothtml_docs:
     returnf"无法加载来自{url}的内容"

    bs_transformer = BeautifulSoupTransformer()

   ifis_async:
      docs_transformed =awaitasyncio.to_thread(
        bs_transformer.transform_documents,
        html_docs,
        tags_to_extract=tags_to_extract,
      )
   else:
      docs_transformed = bs_transformer.transform_documents(
        html_docs,
        tags_to_extract=tags_to_extract,
      )

   ifnotdocs_transformed:
     returnf"未能从{url}提取任何内容"

    content = docs_transformed[0].page_content

   iflen(content) > self.max_content_length:
      content = (
        content[: self.max_content_length] +"\n\n...（内容已截断）"
      )

   returnf"""
**抓取的网站：**{url}
**提取的内容：**

{content}

**注意：**以上为完整网页内容，便于全面分析。
"""

 exceptExceptionase:
   returnf"网页抓取出错，网址：{url}，错误信息：{str(e)}"

def_run(self, url: str, tags_to_extract: List[str] = None)-> str:
 """同步网页抓取"""
 returnasyncio.run(
    self._process_scraping(url, tags_to_extract, is_async=False)
  )

asyncdef_arun(self, url: str, tags_to_extract: List[str] = None)-> str:
 """异步网页抓取"""
 returnawaitself._process_scraping(url, tags_to_extract, is_async=True)

实现原理解析：

AsyncChromiumLoader：使用真实的 Chromium 浏览器，处理 JavaScript 动态加载的网页内容。
BeautifulSoupTransformer：智能地从 HTML 中提取干净的文本内容。
选择性标签提取：仅提取有意义的内容标签，忽略导航栏、广告等无关内容。
内容长度限制：自动截断过长的页面内容，控制 token 成本。注意：此处也可以考虑使用摘要技术。
错误处理机制：优雅地处理网络故障、JavaScript 错误和解析问题。
异步支持：内置异步方法，适合生产环境扩展性需求。

3. 问答 Agent （`qa_agent.py`）——智能大脑

这是整个系统的核心所在。 Agent 负责决定使用哪些工具，并协调整个对话流程。

"""
具备领域特定网页搜索能力的问答 Agent
"""

```python
importlogging
importpandasaspd
fromtypingimportList, Dict, Any, Optional

fromlangchain.agentsimportAgentExecutor, create_structured_chat_agent
fromlangchain_google_genaiimportChatGoogleGenerativeAI
fromlangchain.promptsimportChatPromptTemplate, MessagesPlaceholder
fromlangchain.schemaimportBaseMessage, HumanMessage, AIMessage

fromsearch_toolimportTavilyDomainSearchTool
fromscraping_toolimportWebScrapingTool

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

defcreate_system_prompt(knowledge_sources_md: str, domains: List[str])-> str:
 """根据知识源创建系统提示词"""
 returnf"""你是一名专门用于搜索特定文档网站的问答代理。

可用知识源（按类别/领域/主题划分，每个类别均包含网站及其描述）：
{knowledge_sources_md}

使用说明：
1. 对于任何问题，始终首先使用search_documentation工具进行搜索；
2. 分析用户问题，确定相关的领域/主题/类别；
3. 根据问题涉及的技术或主题选择合适的网站；
4. 若搜索结果不足以完整回答问题，则使用scrape_website工具抓取搜索结果中最相关的URL；
5. 你只能回答与以下可用知识源相关的问题：{domains}；
6. 若问题超出可用知识源范围，请勿回答，并建议用户你可以回答哪些主题的问题。

工具使用策略：
- 首选：使用search_documentation快速查找相关信息；
- 次选：若搜索结果不完整、不清晰或无法提供足够信息回答问题，则使用scrape_website抓取搜索结果中最有希望的URL；
- 为提高效率，始终优先使用搜索，仅在搜索结果无相关信息时才使用抓取。

规则：
- 回答要全面、详细且有帮助；
- 尽可能引用来源；
- 仅在搜索结果无法回答问题时才使用抓取；
- 使用抓取时，选择之前搜索结果中最相关的URL。

你可以使用以下工具：

{{tools}}

使用JSON对象指定工具，需提供action键（工具名称）和action_input键（工具输入）。

有效的"action"值："Final Answer"或{{tool_names}}

每次仅提供一个动作，格式如下：

{{{{
 "action": "$TOOL_NAME",
 "action_input": "$INPUT"
}}}}

请遵循以下格式：

问题：需要回答的问题
思考：考虑前后步骤
动作：

$JSON_BLOB

观察：动作结果
...（重复 思考/动作/观察 若干次）
思考：我知道如何回答了
动作：

{{{{
 "action": "Final Answer",
 "action_input": "你的回答"
}}}}

开始吧！提醒你务必始终以有效的单一动作JSON对象进行回应。必要时使用工具，若适合直接回答则直接回应，若有不清楚之处则请求用户澄清。格式为动作：```$JSON_BLOB```然后观察


class DomainQAAgent:
  """根据用户查询在特定领域网站中搜索的问答代理"""

  def __init__(
    self,
    csv_file_path: str = "sites_data.csv",
    config: Optional[Dict[str, Any]] = None,
  ):
    if config is None:
      raise ValueError("必须提供配置参数")

    self.config = config
    self.sites_df = load_sites_data(csv_file_path)
    self.llm = create_llm(config)
    self.search_tool = create_search_tool(config)
    self.scraping_tool = create_scraping_tool(config)
    self.chat_history: List[BaseMessage] = []
    self.agent_executor = self._create_agent()

    logger.info(f"代理已初始化，共加载了{len(self.sites_df)}个网站")

  def _create_agent(self) -> AgentExecutor:
    """创建带有工具和提示词的结构化聊天代理"""
    knowledge_sources_md, domains = build_knowledge_sources_text(self.sites_df)
    system_message = create_system_prompt(knowledge_sources_md, domains)

prompt = ChatPromptTemplate.from_messages(
  [
    ("system", system_message),
    MessagesPlaceholder(variable_name="chat_history", optional=True),
    (
     "human",
     "{input}\n\n{agent_scratchpad}(提醒：无论如何都要以JSON格式回应)"
     "\n重要提示：调用工具时，务必保持JSON格式不变，使用action和action_input字段，并将函数参数放入action_input字段中。",
    ),
  ]
)

agent = create_structured_chat_agent(
  llm=self.llm, tools=[self.search_tool, self.scraping_tool], prompt=prompt
)

returnAgentExecutor(
  agent=agent,
  tools=[self.search_tool, self.scraping_tool],
  verbose=True,
  max_iterations=10, # 限制迭代次数，防止无限循环
  return_intermediate_steps=True,
  handle_parsing_errors=True, # 优雅处理解析错误
)

asyncdefachat(self, user_input: str)-> str:
 """异步处理用户输入"""
 try:
    logger.info(f"正在处理输入：{user_input}")

    agent_input = {
     "input": user_input,
     "chat_history": (
        self.chat_history[-5:]ifself.chat_historyelse[]
      ), # 限制上下文历史为最近5条
    }

   # 异步调用Agent执行器
    response =awaitself.agent_executor.ainvoke(agent_input)
    answer = response.get("output","抱歉，我无法处理您的请求。")

   # 更新对话历史
    self.chat_history.extend(
      [HumanMessage(content=user_input), AIMessage(content=answer)]
    )

   returnanswer

 exceptExceptionase:
    error_msg =f"发生错误：{str(e)}"
    logger.error(error_msg)
   returnerror_msg

defreset_memory(self):
 """重置对话记忆"""
  self.chat_history.clear()
  logger.info("对话记忆已重置")

LangChain使得 Agent 代码非常简洁，大部分代码都是 Prompt 定义，未来可将 Prompt 单独抽取到文件中。

这里发生了什么？

结构化聊天 Agent（Structured Chat Agent）：我们使用比基础 ReAct Agent 更可靠的结构化聊天 Agent 来处理复杂的工具调用。我发现 LangChain 中的create_structured_chat_agent方法结合 ReAct 框架和非思考型模型，是大多数需要人机交互的 Agent 的最佳选择。
Prompt 设计：我们使用 ReAct 框架，并结合特定的指令和安全防护措施，确保 Agent 仅回答特定领域或类别的问题。我们解析 CSV 内容，并将其传入系统 Prompt。
ChatGoogleGenerativeAI：针对 Gemini 模型的优化封装，具备完善的错误处理机制。
动态 Prompt 生成：系统 Prompt 根据 CSV 提供的知识源动态调整。
MessagesPlaceholder：实现对话记忆和上下文感知。
智能工具策略：明确指示 Agent 何时进行搜索、何时进行网页抓取。
生产安全性：设置迭代次数限制、错误处理机制和全面的日志记录。
异步支持：专为高性能 Web 应用而设计。

4. FastAPI 服务器（`main.py`）- Web 应用的前端入口

我们使用 FastAPI 创建了一个生产级的 Web API，用户可通过 HTTP 请求与 Agent 交互。

"""
领域特定问答Agent的FastAPI应用程序

从.env文件中读取环境变量，并使用它们初始化问答Agent。
提供一个chat端点，允许用户与Agent进行对话。
"""

importlogging
importos
importuuid
fromcontextlibimportasynccontextmanager
fromtypingimportDict, Any

fromfastapiimportFastAPI, HTTPException, Cookie, Response
frompydanticimportBaseModel
importuvicorn
fromdotenvimportload_dotenv

fromqa_agentimportDomainQAAgent

load_dotenv()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


defget_int_env(key: str, default: int)-> int:
 """从环境变量中解析整数，若失败则使用默认值"""
 try:
   returnint(os.getenv(key, default))
 exceptValueError:
    logger.warning(f"{key}无效，使用默认值：{default}")
   returndefault


defget_float_env(key: str, default: float)-> float:
 """从环境变量中解析浮点数，若失败则使用默认值"""
 try:
   returnfloat(os.getenv(key, default))
 exceptValueError:
    logger.warning(f"{key}无效，使用默认值：{default}")
   returndefault

defvalidate_api_keys():
 """验证必要的API密钥是否已配置"""
  google_api_key = os.getenv("GOOGLE_API_KEY")
  tavily_api_key = os.getenv("TAVILY_API_KEY")

 ifnotgoogle_api_keyorgoogle_api_key =="your_google_api_key_here":
   raiseValueError("必须设置环境变量 GOOGLE_API_KEY")

 ifnottavily_api_keyortavily_api_key =="your_tavily_api_key_here":
   raiseValueError("必须设置环境变量 TAVILY_API_KEY")

 returngoogle_api_key, tavily_api_key


defbuild_config()-> Dict[str, Any]:
 """从环境变量构建配置字典"""
  google_api_key, tavily_api_key = validate_api_keys()

  search_depth = os.getenv("SEARCH_DEPTH","basic")
 ifsearch_depthnotin["basic","advanced"]:
    logger.warning(f"无效的 SEARCH_DEPTH 值 '{search_depth}'，使用默认值：basic")
    search_depth ="basic"

 return{
   "google_api_key": google_api_key,
   "tavily_api_key": tavily_api_key,
   "max_results": get_int_env("MAX_RESULTS",10),
   "search_depth": search_depth,
   "max_content_size": get_int_env("MAX_CONTENT_SIZE",10000),
   "max_scrape_length": get_int_env("MAX_SCRAPE_LENGTH",10000),
   "enable_search_summarization": os.getenv(
     "ENABLE_SEARCH_SUMMARIZATION","false"
    ).lower() =="true",
   "llm_temperature": get_float_env("LLM_TEMPERATURE",0.1),
   "llm_max_tokens": get_int_env("LLM_MAX_TOKENS",3000),
   "request_timeout": get_int_env("REQUEST_TIMEOUT",30),
   "llm_timeout": get_int_env("LLM_TIMEOUT",60),
  }


deflog_config(config: Dict[str, Any]):
 """以易读格式记录配置信息（不包含API密钥）"""
  safe_config = {k: vfork, vinconfig.items()ifnotk.endswith("_api_key")}
  logger.info("已加载配置信息：")
 forkey, valueinsafe_config.items():
    logger.info(f" {key}:{value}")


defcreate_config()-> Dict[str, Any]:
 """创建并验证完整的配置信息"""
 try:
    config = build_config()
    log_config(config)
    logger.info("环境变量验证完成")
   returnconfig
 exceptExceptionase:
    logger.error(f"环境变量验证失败：{str(e)}")
   raise


@asynccontextmanager
asyncdeflifespan(app: FastAPI):
 """初始化和清理问答代理的生命周期管理"""
 try:
    logger.info("正在初始化问答代理...")
    config = create_config()
   # 初始化空的会话存储，而非单一代理实例
    app.state.user_sessions = {}
    app.state.config = config
    logger.info("会话存储初始化成功")
 exceptExceptionase:
    logger.error(f"会话存储初始化失败：{str(e)}")
   raise

 yield

  logger.info("正在关闭会话存储...")
 # 清理所有代理实例
 ifhasattr(app.state,'user_sessions'):
   forsession_idinlist(app.state.user_sessions.keys()):
      logger.info(f"正在清理会话{session_id}")
    app.state.user_sessions.clear()


app = FastAPI(
  title="领域问答代理API",
  description="使用Tavily和Langchain在特定领域内搜索信息的问答代理",
  version="1.0.0",
  lifespan=lifespan,
)


defget_or_create_agent(session_id: str)-> DomainQAAgent:
 """获取已有的代理实例，或为会话创建新的代理实例"""
 ifnothasattr(app.state,"user_sessions"):
   raiseHTTPException(status_code=500, detail="会话存储未初始化")

 ifsession_idnotinapp.state.user_sessions:
    logger.info(f"为会话{session_id}创建新的代理实例")
    app.state.user_sessions[session_id] = DomainQAAgent(config=app.state.config)

 returnapp.state.user_sessions[session_id]


classChatRequest(BaseModel):
  message: str
  reset_memory: bool =False


classChatResponse(BaseModel):
  response: str
  status: str ="success"
  session_id: str


@app.get("/health")
asyncdefhealth_check():
 """健康检查接口，包含会话存储状态"""
 return{
   "message":"领域问答代理API运行正常",
   "status":"healthy",
   "version":"1.0.0",
   "active_sessions": len(app.state.user_sessions)ifhasattr(app.state,"user_sessions")else0,
  }


@app.post("/chat", response_model=ChatResponse, summary="与问答代理进行对话")
asyncdefchat(
  request: ChatRequest,
  response: Response,
  session_id: str = Cookie(None)
):
 """通过问答代理处理用户提问"""
 # 如果没有会话ID，则生成新的会话ID
 ifnotsession_id:
    session_id = str(uuid.uuid4())
    response.set_cookie(
      key="session_id",
      value=session_id,
      httponly=True,
      secure=True,
      samesite="lax",
      max_age=3600# 会话有效期1小时
    )

  logger.info(f"正在处理会话{session_id}的聊天请求")

 # 获取或创建当前会话的代理实例
  agent = get_or_create_agent(session_id)

 ifrequest.reset_memory:
    agent.reset_memory()
    logger.info(f"会话{session_id}请求重置记忆")

  response_text =awaitagent.achat(request.message)
  logger.info(f"成功处理会话{session_id}的聊天请求")

 returnChatResponse(
    response=response_text,
    status="success",
    session_id=session_id
  )

@app.post("/reset", summary="重置会话记忆")
asyncdefreset_memory(session_id: str = Cookie(None)):
 """重置当前会话的记忆"""
 ifnotsession_id:
   raiseHTTPException(status_code=400, detail="当前无有效会话")

  agent = get_or_create_agent(session_id)
  agent.reset_memory()
  logger.info(f"已通过接口重置会话{session_id}的记忆")
 return{"message":"会话记忆已重置","status":"success"}


if__name__ =="__main__":
  uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True, log_level="info")

这里到底发生了什么？

FastAPI：现代化的 Python Web 框架，自动生成 OpenAPI 文档。
Pydantic 模型：类型安全的请求 / 响应验证与序列化。
依赖注入：清晰的架构设计，便于错误处理。
生命周期管理：Agent 在启动时初始化一次，并在请求之间持续存在。
环境变量验证：对所有配置参数进行全面的验证。
Cookie 支持：API 使用安全的 HTTP Cookie 为每个用户维护独立的会话记忆。当用户首次发起请求时，系统自动生成一个唯一的会话 ID（UUID），并将其存储在安全的 Cookie 中。每个会话 ID 对应一个独立的 Agent 实例，确保不同用户的对话历史互不干扰，即使同时使用 API 也不会混淆。
生产级特性：健康检查、完善的日志记录、错误处理和监控。
异步支持：完全异步架构，确保高负载下的高性能表现。

整体流程：完整的运作机制

当用户提出问题时，系统的运作流程如下：

HTTP 请求→ 用户向/chat端点发送 POST 请求，携带消息内容。
数据校验→ Pydantic 验证请求格式和 Agent 的可用性。
Agent 处理→qa_agent.py接收问题及对话历史。
结构化推理→ Agent 根据系统提示分析问题，决定处理策略。
工具选择→ 通常首先调用search_tool.py快速获取结果。
受限域搜索→ 通过 Tavily API 仅搜索预先批准的域名。
结果评估→ Agent 评估搜索结果质量，判断是否足够回答问题。
可选网页抓取→ 如有必要，调用scraping_tool.py抓取最相关网页内容。
答案整合→ Agent 将所有信息整合成连贯的回答。
记忆更新→ 更新对话历史，以便后续问题的上下文理解。
HTTP 响应→ 将格式化的响应返回给客户端。

这种架构展示了如何利用 ReAct 框架，以系统化的推理方式，让像 Gemini Flash 这样成本低廉的“非推理型”模型，在结构化任务上胜过昂贵的“推理型”模型。基于工具的架构确保了结果的可靠性和可追溯性，非常适合基于公开网站的生产级问答系统！

如何扩展？

你可以持续添加更多的工具：如 SQL 生成与执行、ElasticSearch、S3 访问等等，可能性无限。只需构建更多工具并通过 MCP 暴露即可！但在此之前，建议先查看 MCP 的仓库列表，很可能其他人已经实现了你需要的集成。

这种设计的核心思想是直接访问数据源，而非构建复杂的数据管道，因为大模型对噪声或未清理的数据有很好的容错能力。

接下来，让我们看看 Agent 的实际表现！以下是运行方法：

配置你的知识源

要配置你的 Agent 能够搜索哪些网站，你需要编辑sites_data.csv文件。这个 CSV 文件包含三个关键列，用于告诉代理可用的资源：

CSV 文件结构：

第 1 列（领域 / 分类）：主题或领域（例如："python"、"web-development"、"machine-learning"）
第 2 列（网站 URL）：具体的网站域名（例如："python.langchain.com"、"docs.python.org"）
第 3 列（描述）：清晰说明该网站包含哪些内容，以及在什么情况下使用该网站。

示例条目：

python,python.langchain.com,LangChain官方Python文档，包含构建LLM应用程序的指南、教程和API参考
web-development,developer.mozilla.org,全面的Web开发文档，涵盖HTML、CSS、JavaScript和Web API

注意：
描述信息非常重要，因为 Agent 正是根据描述来判断某个网站是否有助于回答用户的问题。务必具体说明每个网站涵盖的主题和信息类型。

你可以根据需要添加任意数量的网站。Agent 会根据用户的查询和你提供的描述，智能地选择合适的网站进行搜索。

获取 API 密钥：

你需要前往 Tavily 和 Google AI Studio 获取 API 密钥。

获取 Tavily API 密钥：

访问 tavily.com 并注册一个免费账户。登录后，进入仪表盘或 API 页面，即可找到你的 API 密钥。Tavily 通常提供较为宽裕的免费额度，每月可进行数千次搜索，足够用于测试和小型项目。

获取 Gemini API 密钥：

访问 ai.google.dev，使用你的 Google 账户登录。登录后，找到 "Get API Key" 按钮或进入 API 密钥管理页面。你可以创建一个新项目，然后生成你的 API 密钥。Google 的 Gemini API 也提供了相当充裕的免费额度，每月可进行大量请求。

获得这两个密钥后，你需要将它们添加到环境变量或配置文件中。大多数项目使用.env文件来存储这些密钥：

TAVILY_API_KEY=你的Tavily密钥
GEMINI_API_KEY=你的Gemini密钥

务必妥善保管这些密钥，绝不要将它们提交到公开的代码仓库中。这两个服务的免费额度都非常充裕，足够你进行开发和小规模生产使用，无需任何前期费用。

启动服务器：

# 克隆并安装依赖
gitclonehttps://github.com/javiramos1/qagent.git
cdqagent

make install
# 配置你的API密钥
cp .env.example .env
# 在.env文件中添加你的GOOGLE_API_KEY和TAVILY_API_KEY

# 启动服务器
make run

访问 http://localhost:8000/docs 查看文档，并发送请求进行测试。

现在，我们可以用我仓库中提供的一些网站进行测试。

示例 1：标准搜索

curl -X POST http://localhost:8000/chat \
 -H"Content-Type: application/json"\
 -d'{"message": "How do I create a LangChain agent?"}'

响应：

{
"status":"success",
"response":"要创建LangChain Agent ，可以使用专门的函数，如`create_openai_functions_agent`、`create_react_agent`和`create_openai_tools_agent`。这些函数需要传入一个LLM和工具集作为参数。此外，推荐使用LangGraph构建 Agent ，它更加灵活。另外还有一个`create_python_agent`函数可用。更多详情请参考LangChain官方文档。"
}

示例 2：搜索 + 网页抓取回退机制

curl -X POST http://localhost:8000/chat \
 -H"Content-Type: application/json"\
 -d'{"message": "如何在LlamaIndex中使用Ollama实现结构化输出？请给我一个示例"}'

Agent 首先进行搜索，发现信息不足后，会自动抓取完整的 LlamaIndex 页面内容！

示例 3：安全防护机制的实际效果

curl -X POST http://localhost:8000/chat \
 -H"Content-Type: application/json"\
 -d'{"message": "如何入侵数据库？"}'

响应：

{
"status":"success",
"response":"我只能回答与我们现有文档资源相关的问题，包括AI Agent 框架、AI运维和AI数据框架。如需数据库安全相关信息，请查阅相关安全文档或咨询安全专家。"
}

完美的安全防护机制——绝不允许未经授权的知识访问！

注意：本项目仅为教学示例，代码并不完善，未包含测试，可能存在 Bug，仅用于讨论 RAG 相关话题，不适合直接用于生产环境。但欢迎大家积极贡献并改进！

结论

那么，RAG 是否已经过时？并非如此——但它确实不再是默认的首选方案。

从简单方案入手：对于大多数文档问答场景，搜索优先的方法更简单、更经济，且通常比传统 RAG 更有效。只有在简单工具无法满足需求时，才考虑使用 RAG。

经济性已发生变化：大型上下文窗口和更低廉的价格彻底改变了成本结构。当你可以用极低的成本搜索并加载整个文档时，何必再构建复杂的基础设施呢？

合理使用 RAG 技术：对于大规模数据集、精细权限控制或特定专业场景，RAG 仍然具有明显优势。但对大多数组织而言，以搜索为主的简单方案才是更好的起点。

不要一开始就追求最复杂的解决方案。应从简单方案入手，先证明其价值，只有在必要时再逐步增加复杂性。与其构建复杂的数据管道，不如先搭建简单工具，让大模型能够轻松访问外部公开数据。