我要投稿
微信搜一搜，关注“大模型技术共学营”
告别复杂RAG！30分钟教你打造更高效的问答Agent，效果更好成本更低。核心内容： 1. 传统RAG方法的优缺点与复杂性分析 2. 基于搜索API和大上下文窗口的替代方案详解 3. 30分钟快速搭建问答Agent的实战演示
还记得大家曾经疯狂追捧的检索增强生成（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管道（简化版）
def rag_pipeline(question):
# 第一步：将问题转为嵌入向量
question_embedding = embedding_model.encode(question)
# 第二步：在向量数据库中搜索相似片段
similar_chunks = vector_db.similarity_search(
question_embedding,
top_k=5
)
# 第三步：将检索到的片段拼接成上下文
context = "\n".join([chunk.content for chunk in similar_chunks])
# 第四步：生成答案
prompt = f"Context: {context}\n\nQuestion: {question}\nAnswer:"
return llm.generate(prompt)
回到 2020 年至 2023 年间，RAG（检索增强生成）技术的兴起有着充分的理由：
RAG 通过有选择地引入相关信息，解决了上述问题，并允许在过时的模型基础上引入最新的数据。
假设你有一套公司文档，有人问：“我们 API 的认证功能怎么设置？”
传统的 RAG 方法：
在上下文受限且成本高昂的时代，这种方法效果非常好，但是……
虽然理论上 RAG 听起来优雅简单，但实际落地和维护的复杂程度远超大多数教程所描述的。
我们来总结一下实际中面临的一些挑战：
文档切分（Chunking）) 是指将文档拆分成更小的语义单元，以便存储到向量数据库中。切分策略众多，每种策略都有优缺点，没有完美的方案，主要问题包括：
向量数据库使用的相似性搜索在概念上类似于 Elasticsearch，但专门针对 LLM 的 embedding 进行了优化。尽管 Elasticsearch 经过多年实战验证，但向量数据库相对较新，仍在不断发展完善中。正如我们稍后将讨论的，RAG 的一个替代方案是利用现有的 Elasticsearch 基础设施，通过构建一个工具来查询 Elasticsearch 进行上下文检索。这种方法对那些已经拥有 Elasticsearch 集群的组织尤其有价值，因为它避免了额外引入向量数据库等新基础设施的麻烦。
现实情况总结：
正是由于上述这些复杂性，我们正在构建的这种以简单搜索为先的方案才逐渐受到青睐。核心思想是直接访问数据源，而不是构建复杂的 RAG（检索增强生成）流水线。
与其管理庞大的基础设施，我们更倾向于使用成熟的搜索 API 和大上下文窗口（context window），以更低的复杂度获得更好的效果。
过去 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 可以根据输入内容动态决定使用哪些工具，从而智能地选择工具并配置参数，创造真正的非确定性工作流。
下面我们来看看一些用于从现有系统中提取上下文的工具示例：
公共搜索工具
企业搜索工具
数据库集成工具
然而，在以下场景中，RAG 仍然是合理的选择：
关键结论：RAG 不应作为你的首选方案，仅在特定场景下才值得考虑，比如对隐私有严格要求，或数据量巨大且难以实时访问的情况。
好了，理论讲得够多了。接下来我们动手构建一个实际案例，证明搜索优先方案比传统 RAG 更好用。
我们将创建一个 ** 特定领域的问答 Agent **，它具备以下特点：
你可以将其理解为“去掉 R 的 RAG”——我们不再使用预处理的文档片段，而是直接使用实时、全面的搜索结果来增强生成效果。
传统上，当我们受限于仅 4K 的 token 窗口时，开发者通常会将所有文档加载到向量数据库中，并使用 RAG来提取相关上下文。这种方案复杂且难以维护，如下图所示：
而我们的方案则完全不同：我们直接从源头获取数据，使用工具，充分利用已经成熟且经过验证的搜索引擎进行数据提取。
说明： 此方案仅适用于搜索引擎已索引的公开网站。如果你需要处理内部文档或私有数据，我也可以演示如何使用 ElasticSearch 进行内部上下文检索，甚至使用 SQL 实时检索结构化数据。
1. 初始化阶段： Agent 启动时，会读取一个 CSV 文件，其中包含：
2. 查询处理：当用户提交查询时，LLM 会对查询内容进行分析，智能地选择最适合回答该问题的网站。所有查询均采用异步方式处理。
3. 搜索执行：选定的网站会被传递给我们的搜索工具，该工具使用 Tavily 在这些预先批准的域名范围内快速、精准地执行搜索。
4. 动态决策：根据搜索结果，LLM 会判断当前信息是否足以给出完整答案：
这一方案充分体现了真正 Agent 设计的精妙之处——结构简单，却功能强大。我们并未采用死板的预设路径，而是允许 Agent 根据每次查询的具体上下文，自主决定下一步行动和工具选择。为实现这一目标，我们使用了一个简单的非思考型模型，遵循著名的 ReACT 框架，让 Agent 在循环中自主思考并决定下一步行动。这是一种让非思考型模型实现“思考”的简单方法。
我们采用了双层策略：
这里有一个颇具争议的观点：我们刻意选择了 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 是决定做什么菜的主厨，而工具模块则是专门的厨房设备。
search_tool.py）我们先从搜索功能开始。该工具连接到 Tavily 搜索 API，用于查找相关信息。
"""
使用Tavily进行特定域名的网页搜索工具
"""
import logging
from typing import List, Optional, Type, Any
from pydantic import BaseModel, Field, ConfigDict
from langchain.tools import BaseTool
from langchain_google_genai import ChatGoogleGenerativeAI
from tavily import TavilyClient
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class TavilySearchInput(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'（高级）"
)
class TavilyDomainSearchTool(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,
)
ifnot api_key:
raise ValueError("必须提供TAVILY_API_KEY")
object.__setattr__(self, "tavily_client", TavilyClient(api_key=api_key))
if enable_summarization and google_api_key:
summarizer = create_summarizer_llm(google_api_key)
object.__setattr__(self, "summarizer_llm", summarizer)
logger.info("已启用Gemini Flash-Lite进行搜索结果摘要")
elif enable_summarization:
logger.warning("未提供google_api_key，摘要功能已禁用")
object.__setattr__(self, "enable_summarization", False)
logger.info(
f"Tavily搜索工具初始化完成（摘要功能：{'已启用' if self.enable_summarization else '未启用'}）"
)
asyncdef _search_async(self, query: str, sites: List[str], max_results: int = None, depth: str = None) -> str:
"""异步执行搜索请求"""
try:
final_max_results = max_results or self.default_max_results
final_depth = depth or self.default_depth
logger.info(f"🔍 正在搜索：'{query}'，目标网站：{sites}")
logger.info(
f"搜索参数：max_results={final_max_results}, depth={final_depth}"
)
# 注意：TavilyClient目前没有异步方法，因此使用线程执行
search_results = await asyncio.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', []))}条搜索结果")
if not search_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)} 个字符"
)
if self.enable_summarization and self.summarizer_llm:
try:
logger.info("正在对结果进行智能摘要...")
summarized_result = await self._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}%）"
)
return summarized_result
except Exception as e:
logger.error(
f"摘要生成失败：{e}，返回原始搜索结果。"
)
return final_result
except Exception as e:
error_msg = f"搜索过程中出现错误：{str(e)}"
logger.error(error_msg)
return error_msg
asyncdef _summarize_results_async(self, search_results: str, original_query: str) -> str:
"""使用LLM异步生成搜索结果摘要"""
try:
prompt = create_summary_prompt(search_results, original_query)
response = await asyncio.to_thread(self.summarizer_llm.invoke, prompt)
return response.content
except Exception as e:
logger.error(f"LLM摘要生成失败：{e}")
return search_results
def _run(
self, query: str, sites: List[str], max_results: int = None, depth: str = None
) -> str:
"""执行搜索（同步方式）"""
return asyncio.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:
"""执行搜索（异步方式）"""
returnawait self._search_async(query, sites, max_results, depth)
代码解析与技术要点：
include_domains 参数实现安全防护，确保仅搜索经过批准的网站。_arun 方法支持异步执行。scraping_tool.py）：可选当搜索结果不足以满足需求时，该工具可抓取整个网页内容，以获取更全面的信息。
"""
使用Chromium浏览器进行动态网页内容抓取的工具
"""
import logging
import asyncio
from typing import List, Type
from pydantic import BaseModel, Field, ConfigDict
from langchain.tools import BaseTool
from langchain_community.document_loaders import AsyncChromiumLoader
from langchain_community.document_transformers import BeautifulSoupTransformer
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def get_default_tags() -> List[str]:
"""获取默认用于网页抓取的HTML标签"""
return ["p", "li", "div", "a", "span", "h1", "h2", "h3", "h4", "h5", "h6"]
class WebScrapingInput(BaseModel):
url: str = Field(description="待抓取的网页URL")
tags_to_extract: List[str] = Field(
default_factory=get_default_tags, description="需要提取的HTML标签"
)
class WebScrapingTool(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:
if tags_to_extract isNone:
tags_to_extract = get_default_tags()
loader = AsyncChromiumLoader([url])
if is_async:
html_docs = await asyncio.to_thread(loader.load)
else:
html_docs = loader.load()
ifnot html_docs:
returnf"无法加载来自 {url} 的内容"
bs_transformer = BeautifulSoupTransformer()
if is_async:
docs_transformed = await asyncio.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,
)
ifnot docs_transformed:
returnf"未能从 {url} 提取任何内容"
content = docs_transformed[0].page_content
if len(content) > self.max_content_length:
content = (
content[: self.max_content_length] + "\n\n...（内容已截断）"
)
returnf"""
**抓取的网站：** {url}
**提取的内容：**
{content}
**注意：**以上为完整网页内容，便于全面分析。
"""
except Exception as e:
returnf"网页抓取出错，网址：{url}，错误信息：{str(e)}"
def _run(self, url: str, tags_to_extract: List[str] = None) -> str:
"""同步网页抓取"""
return asyncio.run(
self._process_scraping(url, tags_to_extract, is_async=False)
)
asyncdef _arun(self, url: str, tags_to_extract: List[str] = None) -> str:
"""异步网页抓取"""
returnawait self._process_scraping(url, tags_to_extract, is_async=True)
实现原理解析：
qa_agent.py）——智能大脑这是整个系统的核心所在。 Agent 负责决定使用哪些工具，并协调整个对话流程。
"""
具备领域特定网页搜索能力的问答 Agent
"""
```python
import logging
import pandas as pd
from typing import List, Dict, Any, Optional
from langchain.agents import AgentExecutor, create_structured_chat_agent
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.schema import BaseMessage, HumanMessage, AIMessage
from search_tool import TavilyDomainSearchTool
from scraping_tool import WebScrapingTool
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def create_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
)
return AgentExecutor(
agent=agent,
tools=[self.search_tool, self.scraping_tool],
verbose=True,
max_iterations=10, # 限制迭代次数，防止无限循环
return_intermediate_steps=True,
handle_parsing_errors=True, # 优雅处理解析错误
)
asyncdef achat(self, user_input: str) -> str:
"""异步处理用户输入"""
try:
logger.info(f"正在处理输入：{user_input}")
agent_input = {
"input": user_input,
"chat_history": (
self.chat_history[-5:] if self.chat_history else []
), # 限制上下文历史为最近5条
}
# 异步调用Agent执行器
response = await self.agent_executor.ainvoke(agent_input)
answer = response.get("output", "抱歉，我无法处理您的请求。")
# 更新对话历史
self.chat_history.extend(
[HumanMessage(content=user_input), AIMessage(content=answer)]
)
return answer
except Exception as e:
error_msg = f"发生错误：{str(e)}"
logger.error(error_msg)
return error_msg
def reset_memory(self):
"""重置对话记忆"""
self.chat_history.clear()
logger.info("对话记忆已重置")
LangChain 使得 Agent 代码非常简洁，大部分代码都是 Prompt 定义，未来可将 Prompt 单独抽取到文件中。
这里发生了什么？
main.py）- Web 应用的前端入口我们使用 FastAPI 创建了一个生产级的 Web API，用户可通过 HTTP 请求与 Agent 交互。
"""
领域特定问答Agent的FastAPI应用程序
从.env文件中读取环境变量，并使用它们初始化问答Agent。
提供一个chat端点，允许用户与Agent进行对话。
"""
import logging
import os
import uuid
from contextlib import asynccontextmanager
from typing import Dict, Any
from fastapi import FastAPI, HTTPException, Cookie, Response
from pydantic import BaseModel
import uvicorn
from dotenv import load_dotenv
from qa_agent import DomainQAAgent
load_dotenv()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def get_int_env(key: str, default: int) -> int:
"""从环境变量中解析整数，若失败则使用默认值"""
try:
return int(os.getenv(key, default))
except ValueError:
logger.warning(f"{key}无效，使用默认值：{default}")
return default
def get_float_env(key: str, default: float) -> float:
"""从环境变量中解析浮点数，若失败则使用默认值"""
try:
return float(os.getenv(key, default))
except ValueError:
logger.warning(f"{key}无效，使用默认值：{default}")
return default
def validate_api_keys():
"""验证必要的API密钥是否已配置"""
google_api_key = os.getenv("GOOGLE_API_KEY")
tavily_api_key = os.getenv("TAVILY_API_KEY")
ifnot google_api_key or google_api_key == "your_google_api_key_here":
raise ValueError("必须设置环境变量 GOOGLE_API_KEY")
ifnot tavily_api_key or tavily_api_key == "your_tavily_api_key_here":
raise ValueError("必须设置环境变量 TAVILY_API_KEY")
return google_api_key, tavily_api_key
def build_config() -> Dict[str, Any]:
"""从环境变量构建配置字典"""
google_api_key, tavily_api_key = validate_api_keys()
search_depth = os.getenv("SEARCH_DEPTH", "basic")
if search_depth notin ["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),
}
def log_config(config: Dict[str, Any]):
"""以易读格式记录配置信息（不包含API密钥）"""
safe_config = {k: v for k, v in config.items() ifnot k.endswith("_api_key")}
logger.info("已加载配置信息：")
for key, value in safe_config.items():
logger.info(f" {key}: {value}")
def create_config() -> Dict[str, Any]:
"""创建并验证完整的配置信息"""
try:
config = build_config()
log_config(config)
logger.info("环境变量验证完成")
return config
except Exception as e:
logger.error(f"环境变量验证失败：{str(e)}")
raise
@asynccontextmanager
asyncdef lifespan(app: FastAPI):
"""初始化和清理问答代理的生命周期管理"""
try:
logger.info("正在初始化问答代理...")
config = create_config()
# 初始化空的会话存储，而非单一代理实例
app.state.user_sessions = {}
app.state.config = config
logger.info("会话存储初始化成功")
except Exception as e:
logger.error(f"会话存储初始化失败：{str(e)}")
raise
yield
logger.info("正在关闭会话存储...")
# 清理所有代理实例
if hasattr(app.state, 'user_sessions'):
for session_id in list(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,
)
def get_or_create_agent(session_id: str) -> DomainQAAgent:
"""获取已有的代理实例，或为会话创建新的代理实例"""
ifnot hasattr(app.state, "user_sessions"):
raise HTTPException(status_code=500, detail="会话存储未初始化")
if session_id notin app.state.user_sessions:
logger.info(f"为会话 {session_id} 创建新的代理实例")
app.state.user_sessions[session_id] = DomainQAAgent(config=app.state.config)
return app.state.user_sessions[session_id]
class ChatRequest(BaseModel):
message: str
reset_memory: bool = False
class ChatResponse(BaseModel):
response: str
status: str = "success"
session_id: str
@app.get("/health")
asyncdef health_check():
"""健康检查接口，包含会话存储状态"""
return {
"message": "领域问答代理API运行正常",
"status": "healthy",
"version": "1.0.0",
"active_sessions": len(app.state.user_sessions) if hasattr(app.state, "user_sessions") else0,
}
@app.post("/chat", response_model=ChatResponse, summary="与问答代理进行对话")
asyncdef chat(
request: ChatRequest,
response: Response,
session_id: str = Cookie(None)
):
"""通过问答代理处理用户提问"""
# 如果没有会话ID，则生成新的会话ID
ifnot session_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)
if request.reset_memory:
agent.reset_memory()
logger.info(f"会话 {session_id} 请求重置记忆")
response_text = await agent.achat(request.message)
logger.info(f"成功处理会话 {session_id} 的聊天请求")
return ChatResponse(
response=response_text,
status="success",
session_id=session_id
)
@app.post("/reset", summary="重置会话记忆")
asyncdef reset_memory(session_id: str = Cookie(None)):
"""重置当前会话的记忆"""
ifnot session_id:
raise HTTPException(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")
这里到底发生了什么？
当用户提出问题时，系统的运作流程如下：
/chat端点发送 POST 请求，携带消息内容。qa_agent.py接收问题及对话历史。search_tool.py快速获取结果。scraping_tool.py抓取最相关网页内容。这种架构展示了如何利用 ReAct 框架，以系统化的推理方式，让像 Gemini Flash 这样成本低廉的“非推理型”模型，在结构化任务上胜过昂贵的“推理型”模型。基于工具的架构确保了结果的可靠性和可追溯性，非常适合基于公开网站的生产级问答系统！
如何扩展？
你可以持续添加更多的工具：如 SQL 生成与执行、ElasticSearch、S3 访问等等，可能性无限。只需构建更多工具并通过 MCP 暴露即可！但在此之前，建议先查看 MCP 的仓库列表，很可能其他人已经实现了你需要的集成。
这种设计的核心思想是直接访问数据源，而非构建复杂的数据管道，因为大模型对噪声或未清理的数据有很好的容错能力。
接下来，让我们看看 Agent 的实际表现！以下是运行方法：
配置你的知识源
要配置你的 Agent 能够搜索哪些网站，你需要编辑sites_data.csv文件。这个 CSV 文件包含三个关键列，用于告诉代理可用的资源：
CSV 文件结构：
示例条目：
python,python.langchain.com,LangChain官方Python文档，包含构建LLM应用程序的指南、教程和API参考
web-development,developer.mozilla.org,全面的Web开发文档，涵盖HTML、CSS、JavaScript和Web API
注意：
描述信息非常重要，因为 Agent 正是根据描述来判断某个网站是否有助于回答用户的问题。务必具体说明每个网站涵盖的主题和信息类型。
你可以根据需要添加任意数量的网站。Agent 会根据用户的查询和你提供的描述，智能地选择合适的网站进行搜索。
你需要前往 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密钥
务必妥善保管这些密钥，绝不要将它们提交到公开的代码仓库中。这两个服务的免费额度都非常充裕，足够你进行开发和小规模生产使用，无需任何前期费用。
# 克隆并安装依赖
git clone https://github.com/javiramos1/qagent.git
cd qagent
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 仍然具有明显优势。但对大多数组织而言，以搜索为主的简单方案才是更好的起点。
不要一开始就追求最复杂的解决方案。应从简单方案入手，先证明其价值，只有在必要时再逐步增加复杂性。与其构建复杂的数据管道，不如先搭建简单工具，让大模型能够轻松访问外部公开数据。
53AI，企业落地大模型首选服务商
产品：场景落地咨询+大模型应用平台+行业解决方案
承诺：免费场景POC验证，效果验证后签署服务协议。零风险落地应用大模型，已交付160+中大型企业
2025-06-21
一文搞懂什么是RAG
2025-06-21
Google | 溯源分析RAG系统错误，提出选择性生成框架，让RAG问答准确率提升10%
2025-06-20
鹅厂实习生血泪贴：Agent/RAG黑科技，真相竟是这样！
2025-06-20
拒绝AI“一本正经地胡说八道”：我用三版Prompt驯服RAG模型的实战复盘
2025-06-20
从0到1落地一个RAG智能客服系统
2025-06-20
RAG 知识库核心模块全解（产品视角 + 技术细节）
2025-06-20
不依赖于复杂框架，用简单易懂的实现教你二十三种RAG技巧！
2025-06-20
RAGFlow实战：如何根据文档类型选择最佳切片策略？
2025-03-24
2025-03-24
2025-03-24
2025-03-28
2025-04-01
2025-04-13
2025-04-19
2025-04-09
2025-04-16
2025-05-08
2025-06-20
2025-06-19
2025-06-13
2025-06-09
2025-06-06
2025-05-30
2025-05-29
2025-05-29
回到顶部
粤ICP备17114055号