news 2026/5/10 15:04:51

为AI Agent打造多源聚合搜索技能:并行抓取与AI提炼的工程实践

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
为AI Agent打造多源聚合搜索技能:并行抓取与AI提炼的工程实践

1. 项目概述:为AI Agent打造的多源聚合搜索技能

如果你正在使用Claude Code、OpenCode这类AI编程助手,或者任何需要联网搜索能力的AI Agent,你肯定遇到过内置搜索工具的局限性:结果单一、信息陈旧、格式杂乱,Agent拿到手后还得花大量“脑力”去整理和判断。这正是我开发SearchOnline这个技能的初衷——它不是一个简单的搜索接口封装,而是一个专为AI Agent设计的“信息副驾驶”。

简单来说,SearchOnline的核心工作流程是:并行抓取 + AI提炼。它同时向多个高质量的付费搜索源(如Tavily、WebSearchAPI)发起请求,然后将所有返回的原始信息(可能包含重复、无关或格式不一的片段)一股脑儿扔给Google的Gemini大模型,由Gemini来扮演“信息整理师”的角色,最终输出一份结构清晰、重点突出、且格式高度适配AI Agent“阅读习惯”的Markdown报告。这相当于给你的Agent配备了一个专业的“研究助理”,它不仅能帮你找到信息,还能先帮你把信息消化、整理好。

这个工具特别适合处理几种典型场景:一是需要获取最新动态的,比如“Python 3.13有什么新特性?”或“今天OpenAI发布了什么公告?”;二是需要综合多方信息的复杂研究,比如“对比一下React和Vue在大型项目中的性能表现”;三是需要实时数据的查询,比如“特斯拉当前的股价是多少?”。在这些场景下,内置搜索工具往往力不从心,而SearchOnline通过聚合与提炼,能显著提升Agent回答的时效性、准确性和信息密度。

2. 核心设计思路与架构解析

2.1 为什么选择“多源聚合”而非单一搜索?

很多开发者第一个疑问是:市面上已经有Tavily、SerpAPI等优秀的搜索API,为什么还要费劲做聚合?直接调用一个最好的不就行了吗?这里涉及到信息获取的“覆盖率”和“可靠性”问题。

首先,没有哪个单一的搜索源是完美的。Tavily在学术和技术文档搜索上很强,但对某些区域性新闻或小众论坛的覆盖可能不如WebSearchAPI。Google Search Grounding(通过Gemini API)能直接访问谷歌的实时索引,权威性高,但在结果格式的规整性上可能不如专门的API。通过并行请求多个源,我们实际上是在做一个“冗余备份”和“交叉验证”。即使某一个源暂时故障或返回空结果,其他源也能顶上,确保了服务的鲁棒性。更重要的是,不同源的排序算法和侧重点不同,聚合后的结果集往往比单一源更全面,能减少因算法偏见导致的信息遗漏。

2.2 AI提炼:从“信息堆”到“知识块”的关键一跃

仅仅把多个来源的结果拼凑在一起,会产生大量噪音:重复的链接、矛盾的片段、无关的广告信息。如果直接把这样的原始数据丢给Agent,反而会增加其认知负担。因此,引入AI进行后处理是质变的关键

SearchOnline选择Gemini模型作为“大脑”来进行这项提炼工作,主要基于几点考量:

  1. 强大的上下文理解和摘要能力:Gemini Pro/Flash模型在长文本理解和信息浓缩方面表现优异,能准确抓住多个搜索结果中的共同主题和核心事实。
  2. 可控的结构化输出:我们可以通过精心设计的Prompt,指挥Gemini按照我们想要的格式(比如分点列表、对比表格、时间线等)来组织信息,这比解析不同网站千变万化的HTML要可靠得多。
  3. 与Google生态的无缝集成:使用Gemini API可以直接调用其“Google Search Grounding”功能,这相当于在一个请求里同时完成了搜索和初步理解,简化了架构。

这个设计思路的本质是将计算负载转移:将原本需要AI Agent在收到杂乱信息后进行的“理解-筛选-重组”工作,前置到了一个专门的、优化的流程中。让SearchOnline这个“预处理管道”承担脏活累活,最终交付给Agent的是可以直接吸收利用的“营养膏”。

2.3 面向Agent的优化设计

这个技能从诞生之初目标就很明确:服务于AI Agent。因此,它的所有输出设计都围绕着一个核心——降低Agent的解析难度,提升其行动效率

  • 格式标准化:强制输出为Markdown。Markdown是大多数AI模型训练数据中的常见格式,模型对其结构(标题、列表、代码块、表格)的理解和生成能力很强。一份结构良好的Markdown文档,Agent可以轻松地提取其中的关键信息点,或直接将其嵌入到自己的回答中。
  • 信息密度自适应:对于“今天天气如何”这类简单查询,提炼结果会是高度浓缩的一句话或一个数据点;对于“请分析Web3游戏的发展现状”这类复杂研究,输出则会是一个包含背景、现状、挑战、案例的详细报告。这种自适应能力,使得Agent在不同场景下都能获得恰到好处的信息量。
  • 错误处理与降级方案:考虑到AI服务(尤其是Gemini API)可能存在速率限制或暂时不可用,技能内设计了自动重试和模型降级逻辑(例如,从Gemini Pro回退到Gemini Flash)。最坏情况下,如果AI提炼完全失败,它也不会“罢工”,而是将格式化后的原始搜索结果返回,确保Agent至少能获得一些信息,实现了“优雅降级”。

3. 环境配置与核心代码实现详解

3.1 依赖管理与环境变量配置

项目本身依赖极简,核心就是requests库用于HTTP通信。更关键的是外部服务的API密钥管理。我强烈建议使用.env文件来管理,并与.gitignore配合,绝对避免密钥泄露。

.env文件配置示例与深度解析:

GEMINI_API_KEY=your_google_ai_studio_key_here TAVILY_API_KEY=your_tavily_key_here WEBSEARCHAPI_KEY=your_websearchapi_key_here
  • GEMINI_API_KEY(必需):这是整个技能的“引擎”。没有它,AI提炼功能就无法工作。获取地址是Google AI Studio。这里有个实战经验:注册后,新手通常有免费的调用额度,足够进行大量的开发和测试。但在生产环境,务必关注其定价策略,尤其是对于gemini-pro这类更高阶的模型。
  • TAVILY_API_KEY(强烈推荐):Tavily是我目前体验下来,对开发者最友好的搜索API之一。它针对AI应用做了大量优化,返回的结果已经过初步清洗和来源标注,质量很高。它的免费套餐通常也足够个人项目使用。
  • WEBSEARCHAPI_KEY(可选):作为Tavily的一个补充。在某些测试中,我发现它对中文互联网内容或一些非常垂直的论坛有更好的覆盖。如果你的应用场景涉及多语言或特定社区,可以考虑配置它。

注意:成本控制:并行调用多个付费API意味着成本也是并行的。在开发阶段,可以通过在代码中设置较低的max_results(比如1或2)来限制每次查询返回的结果数量,从而有效控制成本。上线前,务必对每个API的调用费用和预计查询量进行估算。

3.2 核心执行流程代码拆解

让我们深入到SearchOnline.py的核心函数中,看看“并行搜索”和“AI提炼”是如何具体实现的。以下是我简化后的逻辑框架,并加入了关键注释:

import os import concurrent.futures import requests from typing import List, Dict import google.generativeai as genai # 1. 配置加载与客户端初始化 def setup_clients(): """初始化各搜索API的客户端和Gemini客户端""" genai.configure(api_key=os.getenv('GEMINI_API_KEY')) # 初始化Tavily、WebSearchAPI等客户端(此处省略具体SDK初始化代码) clients = {} if tavily_key := os.getenv('TAVILY_API_KEY'): clients['tavily'] = TavilyClient(tavily_key) # ... 其他客户端初始化 return clients # 2. 并行搜索执行器 def fetch_parallel(query: str, max_results: int, clients: Dict) -> Dict[str, List]: """并发地向所有配置好的搜索源发起请求""" results = {} def search_with_client(source_name, client): try: # 调用不同客户端的搜索方法 if source_name == 'tavily': return client.search(query, max_results=max_results) # ... 其他来源的处理 except Exception as e: print(f"Error from {source_name}: {e}") return [] with concurrent.futures.ThreadPoolExecutor(max_workers=len(clients)) as executor: # 提交所有搜索任务 future_to_source = { executor.submit(search_with_client, name, client): name for name, client in clients.items() } # 收集结果 for future in concurrent.futures.as_completed(future_to_source): source_name = future_to_source[future] try: results[source_name] = future.result(timeout=10) # 设置超时 except concurrent.futures.TimeoutError: print(f"Timeout from {source_name}") results[source_name] = [] return results # 3. AI提炼与格式化核心 def synthesize_with_gemini(query: str, aggregated_raw_results: Dict[str, List]) -> str: """将聚合的原始结果交给Gemini进行智能摘要和格式化""" # 第一步:构建给Gemini的Prompt。这是效果好坏的关键! prompt = f""" 你是一个专业的研究助手。请根据以下来自多个搜索源的、关于“{query}”的原始信息,进行综合、去重、提炼,并生成一份结构清晰的Markdown格式报告。 报告要求: 1. 开头给出一个简洁的概述。 2. 将核心信息分点或分章节组织。 3. 如果信息涉及对比、步骤、列表,请使用Markdown表格或列表。 4. 保留重要的事实、数据、日期和来源指向。 5. 确保信息准确,如果不同来源有冲突,请注明。 6. 语言简洁、客观。 原始搜索数据如下: {str(aggregated_raw_results)} """ # 第二步:选择模型并生成。这里实现了简单的降级逻辑。 model_priority = ['gemini-2.0-flash-exp', 'gemini-1.5-flash', 'gemini-1.5-pro'] response_text = "AI summarization failed. Falling back to raw results." for model_name in model_priority: try: model = genai.GenerativeModel(model_name) # 设置生成配置,控制输出长度和随机性 response = model.generate_content( prompt, generation_config=genai.GenerationConfig( max_output_tokens=2000, temperature=0.2, # 低温度保证输出稳定、事实性强 ) ) response_text = response.text break # 成功则跳出循环 except Exception as e: print(f"Model {model_name} failed: {e}. Trying next...") continue return response_text # 4. 主函数 def main(query: str, max_results: int = 5): """技能的主入口函数""" clients = setup_clients() if not clients.get('gemini'): # 至少需要Gemini return "Error: GEMINI_API_KEY is required." print(f"Searching for: {query}") # 并行获取原始结果 raw_results = fetch_parallel(query, max_results, clients) # AI提炼 final_report = synthesize_with_gemini(query, raw_results) return final_report

关键实现细节解析:

  1. 并发控制:使用ThreadPoolExecutor实现真正的并行I/O操作。网络请求是典型的I/O密集型任务,并发可以大幅缩短总等待时间。max_workers设置为客户端数量,确保所有搜索同时发起。
  2. 超时与容错:每个future.result(timeout=10)设置了10秒超时。防止某个缓慢的API拖垮整个流程。任何单个源的失败都会被捕获,不影响其他源的结果收集。
  3. Prompt工程synthesize_with_gemini函数中的prompt是灵魂。它明确指示了Gemini的角色、任务、格式要求和处理原则(如冲突处理)。在实际使用中,你可能需要根据不同的查询类型(事实查询 vs. 分析报告)微调这个Prompt,以达到最佳效果。
  4. 模型降级策略model_priority列表定义了模型尝试顺序。通常将更快、更便宜的模型(如Flash)放在前面,只有在它失败(如达到速率限制)时才尝试更强大、可能更慢更贵的Pro模型。这是一种兼顾成本、速度和成功率的实用策略。

4. 集成到AI Agent的实战指南

4.1 作为Claude Code Skill集成

对于Claude Code,集成过程非常直观。项目中的SKILL.md文件就是为它准备的说明书。核心步骤是让Claude Code“学会”在特定场景下调用这个本地技能。

你需要做的是:

  1. 在Claude Code的界面中,找到添加本地技能的选项。
  2. 将技能路径指向包含SearchOnline.pySKILL.md的目录。
  3. SKILL.md中会定义技能的触发条件(例如,当用户问题涉及“最新”、“查询”、“搜索”等关键词,或明确要求联网搜索时)、调用方式(命令行)以及输入输出格式。

集成后,当你在Claude Code中提问“帮我查一下最近关于Rust内存安全的新文章”,Claude Code会识别出这是一个需要联网搜索的请求,自动触发SearchOnline技能,并将最终整理好的Markdown报告纳入其上下文中,用于生成最终回答。

4.2 作为通用Python模块调用

如果你是在自己的Python脚本或AI Agent框架(如LangChain、AutoGen)中使用,你可以将SearchOnline模块化。

示例:封装成一个可调用的函数库

# 在你的Agent主程序中 import sys sys.path.append('/path/to/searchonline') # 添加SearchOnline项目路径 from SearchOnline import search_online def agent_think_and_act(user_query): # ... 你的Agent逻辑判断是否需要搜索 if need_web_search(user_query): search_results = search_online(user_query, max_results=3) # 将search_results作为上下文的一部分,送入你的LLM生成最终回答 final_answer = generate_answer_with_context(user_query, search_results) return final_answer else: # ... 其他处理逻辑

与LangChain集成示例: 你可以将SearchOnline包装成一个LangChain Tool,这样就能无缝接入LangChain的Agent执行链。

from langchain.tools import BaseTool from SearchOnline import search_online class SearchOnlineTool(BaseTool): name = "Enhanced_Web_Search" description = "A powerful tool to search the web for current information. Use for queries about recent events, real-time data, or topics beyond general knowledge." def _run(self, query: str) -> str: """执行搜索并返回格式化结果""" return search_online(query, max_results=5) async def _arun(self, query: str) -> str: # 异步版本(如果需要) raise NotImplementedError("Async not implemented yet") # 然后在你的LangChain Agent中加载这个Tool

4.3 性能调优与配置建议

在实际部署中,有几个参数可以根据你的需求进行调整:

  • max_results(每源结果数):这是平衡速度、成本和质量的主要杠杆。对于事实性查询(如“谁赢得了2023年诺贝尔文学奖?”),设置为1或2足矣。对于深度研究,可以提高到5-10。不建议超过10,因为信息冗余度会急剧增加,加重AI提炼的负担和API成本。
  • 并发线程数:在ThreadPoolExecutor中,通常将max_workers设置为配置的搜索源数量即可。如果源很多(比如超过5个),可以考虑适当调低,避免对本地网络或目标API造成过大压力。
  • Gemini模型选择:代码中默认的降级链是合理的。对于绝大多数信息提炼任务,gemini-1.5-flash在速度、成本和效果上取得了最佳平衡,可以作为首选。gemini-1.5-pro更适合需要深度推理、处理极其复杂或矛盾信息的场景。
  • 超时时间:根据你的网络环境和目标API的响应速度,调整future.result(timeout)的值。对于国内用户,如果使用了某些访问速度较慢的海外API,可能需要适当延长超时时间。

5. 常见问题排查与实战心得

5.1 错误排查速查表

问题现象可能原因解决方案
运行后立即报错ModuleNotFoundError缺少requests库或其它依赖。在项目目录下执行pip install -r requirements.txt(如果存在) 或pip install requests
报错Invalid API KeyAuthentication Error1. API密钥未设置。
2. 密钥错误或已失效。
3..env文件未正确加载。
1. 检查.env文件是否存在且格式正确(无空格,无引号)。
2. 在终端执行echo $GEMINI_API_KEY(Linux/Mac) 或echo %GEMINI_API_KEY%(Windows) 验证环境变量是否生效。
3. 前往对应API提供商后台,确认密钥状态并重置。
程序卡住很长时间无输出1. 某个搜索API请求超时。
2. 网络连接问题(如代理设置)。
1. 检查代码中的超时设置,尝试增加超时时间。
2. 临时关闭系统代理或VPN试试。
3. 在代码中添加更详细的日志,打印出每个搜索任务的开始和结束时间,定位卡顿源。
返回结果中出现AI summarization failed...Gemini API调用失败(配额用尽、速率限制、服务暂时不可用)。1. 检查Google AI Studio的用量面板,确认配额和速率限制。
2. 查看控制台错误信息,如果是429 Too Many Requests,需要降低调用频率或升级配额。
3. 确保代码中的模型降级逻辑正常工作。
搜索结果质量差或不相关1. 查询语句不清晰。
2. 某个搜索源质量不佳。
3. AI提炼的Prompt不够精准。
1. 尝试优化你的查询词,更具体、明确。
2. 在代码中临时注释掉某个搜索源,测试单个源的质量,考虑更换或调整权重。
3. 优化synthesize_with_gemini函数中的Prompt,加入更明确的指令,例如“请优先关注来自权威科技媒体(如TechCrunch, The Verge)的信息”。
集成到Claude Code后不触发SKILL.md中的触发规则(triggers)定义得太窄或与你的问题不匹配。编辑SKILL.md,扩展triggers部分的关键词列表,或调整匹配逻辑(如使用正则表达式)。Claude Code的文档有详细说明。

5.2 从实战中积累的经验与技巧

技巧一:Prompt是效果的放大器不要满足于默认的提炼Prompt。针对不同类型的查询,设计专门的Prompt模板,效果会立竿见影。例如:

  • 事实查询Prompt:强调“提取精确数字、名称、日期”,“用一句话回答”。
  • 对比分析Prompt:要求“以Markdown表格形式呈现双方优缺点”,“列出支持各自观点的来源”。
  • 教程/步骤类Prompt:指示“按时间或逻辑顺序编号步骤”,“注明每个步骤所需的工具或前提条件”。

你可以建立一个小的Prompt模板库,让SearchOnline根据查询的初步分类(可以用一个简单的关键词匹配或另一个轻量级LLM来判断)自动选择最合适的模板。

技巧二:实施结果缓存对于相对静态或重复率高的问题(例如,“Python的创始人是谁?”),频繁搜索是一种浪费。可以引入一个简单的缓存机制,例如使用diskcachesqlite,将(query, max_results)作为键,将AI提炼后的最终结果缓存起来,并设置一个合理的过期时间(TTL,例如1小时或1天)。这能极大降低API调用成本并提升响应速度。

技巧三:善用“优雅降级”SearchOnline的降级逻辑(AI失败则返回原始结果)是一个安全网。但你可以做得更好。例如,当AI提炼失败时,可以尝试一个更简单、成本更低的模型进行二次提炼,或者仅对原始结果做一个简单的去重和排序后再返回。始终确保你的Agent能拿到“可用”的信息,哪怕不是“最优”的。

技巧四:监控与日志在生产环境使用,一定要加入监控。记录每次搜索的耗时、各API的调用成功率、Gemini模型的使用情况以及最终返回结果的长度。这些数据能帮你发现瓶颈(哪个API最慢?哪个最容易失败?),并优化你的配置和代码,比如调整超时时间、更换不稳定的搜索源。

开发SearchOnline的过程,让我深刻体会到,为AI Agent构建工具,核心思想是“预处理”和“标准化”。把杂乱、异构的互联网信息,通过工程化和智能化的手段,处理成Agent易于消化吸收的“标准餐”,能极大地解放Agent的“思考”能力,让它专注于更高层次的推理和创造。这个技能本身也在不断迭代,比如我正在考虑加入对学术数据库(如Google Scholar)或特定垂直领域(如GitHub、Stack Overflow)搜索的支持,让它的信息获取能力更加立体。如果你有好的想法或遇到了有趣的问题,非常欢迎一起探讨。

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/5/10 15:03:45

如何用Video2X实现免费AI视频画质提升:新手终极指南

如何用Video2X实现免费AI视频画质提升:新手终极指南 【免费下载链接】video2x A machine learning-based video super resolution and frame interpolation framework. Est. Hack the Valley II, 2018. 项目地址: https://gitcode.com/GitHub_Trending/vi/video2x…

作者头像 李华
网站建设 2026/5/10 15:03:34

FFmpeg GUI终极指南:3分钟掌握图形化音视频处理技巧

FFmpeg GUI终极指南:3分钟掌握图形化音视频处理技巧 【免费下载链接】ffmpegGUI ffmpeg GUI 项目地址: https://gitcode.com/gh_mirrors/ff/ffmpegGUI 还在为复杂的FFmpeg命令行参数而烦恼吗?FFmpeg GUI是一款免费的图形化音视频处理工具&#xf…

作者头像 李华
网站建设 2026/5/10 14:59:30

RHEL8/CentOS 7用户看过来:保姆级教程,搞定exFAT格式U盘挂载难题

RHEL8/CentOS 7系统挂载exFAT格式U盘实战指南 你是否遇到过这样的场景:同事递来一个存有4GB虚拟机镜像的U盘,你信心满满地插入Linux服务器,却发现系统根本不识别这个exFAT格式的存储设备?作为长期奋战在运维一线的工程师&#xff…

作者头像 李华
网站建设 2026/5/10 14:52:43

3分钟掌握DeepL翻译插件:让浏览器变身多语言阅读神器

3分钟掌握DeepL翻译插件:让浏览器变身多语言阅读神器 【免费下载链接】deepl-chrome-extension A DeepL Translator Chrome extension 项目地址: https://gitcode.com/gh_mirrors/de/deepl-chrome-extension 还在为看不懂的外文网页而烦恼吗?Deep…

作者头像 李华