使用 Brave Search API 与 uAgents 框架构建智能搜索代理
引言:当智能代理遇见强大搜索
在信息爆炸的时代,如何快速准确地获取所需信息成为关键挑战。本文将深入解析如何结合 Brave Search API 的强大搜索能力和 uAgents 框架的智能代理功能,创建一个能实时搜索网络和本地商业信息的 AI 代理系统。这个解决方案无需复杂基础设施,通过 Python 即可实现,特别适合需要动态信息获取的应用场景。
核心价值:本文提供的技术方案能帮助开发者构建具备实时网络搜索和本地商业搜索能力的智能代理,适用于聊天机器人、研究工具和本地服务发现等多种应用场景。
一、技术生态系统解析
1.1 Fetch.ai 平台概述
Fetch.ai 提供了一个完整的去中心化人工智能生态系统,核心组件包括:
-
uAgents 框架:轻量级代理开发框架,内置身份管理、消息传递和协议支持 -
Agentverse 平台:开放的代理市场,支持代理注册、发现和使用 -
ASI:One LLM:专为代理工作流设计的 Web3 原生语言模型 -
Chat Protocol:标准化的代理间对话通信协议
这些组件共同构成了构建去中心化 AI 应用的坚实基础。
1.2 Brave Search API 优势
Brave Search 提供不同于传统搜索引擎的独特价值:
-
独立索引:不依赖其他搜索引擎的独立结果 -
隐私保护:默认不跟踪用户搜索行为 -
丰富 API:提供网络搜索和本地商业搜索双重能力 -
过滤选项:支持按时间、安全级别等条件筛选结果
二、项目架构与设计目标
2.1 系统架构图解

2.2 核心功能设计
本项目构建的 AI 代理具备双重搜索能力:
智能回退机制:当本地搜索无结果时,系统自动切换至网络搜索,确保总能返回相关信息。
三、代码实现详解
3.1 环境准备与依赖安装
在开始前,请确保准备好以下资源:
-
Brave Search API 密钥(从 Brave 开发者平台获取) -
ASI:One API 密钥(登录 ASI:One 后在”API Keys”选项卡获取) -
Python 3.8+ 环境
安装所需依赖包:
pip install uagents uagents-adapter requests python-dotenv
3.2 代理设置与 MCP 连接
创建 agent_setup.py
文件:
from uagents_adapter import MCPServerAdapter
from server import mcp # 从 server.py 导入 MCP 实例
from uagents import Agent
# 初始化 MCP 适配器
mcp_adapter = MCPServerAdapter(
mcp_server=mcp,
asi1_api_key="your-asi1-api-key", # 替换为实际密钥
model="asi1-mini" # 使用的语言模型
)
# 创建并配置代理
agent = Agent()
for protocol in mcp_adapter.protocols:
agent.include(protocol, publish_manifest=True)
if __name__ == "__main__":
mcp_adapter.run(agent)
这段代码实现了:
-
创建代理实例 -
连接到 MCP(多上下文协议)服务器 -
集成 ASI:One 模型进行智能工具选择 -
发布代理清单到 Agentverse
3.3 MCP 服务器与搜索工具实现
创建 server.py
文件:
import os
import time
import requests
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
BRAVE_API_KEY = os.getenv("BRAVE_API_KEY")
if not BRAVE_API_KEY:
raise ValueError("BRAVE_API_KEY 环境变量必须设置")
# 速率限制配置
RATE_LIMIT = {"per_second": 1, "per_month": 15000}
request_count = {"second": 0, "month": 0, "last_reset": time.time()}
def check_rate_limit():
"""确保请求不超出 Brave API 限制"""
now = time.time()
if now - request_count["last_reset"] > 1:
request_count["second"] = 0
request_count["last_reset"] = now
if (request_count["second"] >= RATE_LIMIT["per_second"] or
request_count["month"] >= RATE_LIMIT["per_month"]):
raise ValueError("超出速率限制")
request_count["second"] += 1
request_count["month"] += 1
# 创建 MCP 实例
mcp = FastMCP("BraveSearch")
3.3.1 网络搜索工具实现
@mcp.tool()
def brave_web_search(
query: str,
count: int = 10,
offset: int = 0,
result_type: str = "all",
safety_level: str = "moderate",
freshness: str = "all"
) -> str:
"""使用 Brave Search API 搜索网页内容"""
check_rate_limit()
# 参数验证
if len(query) > 400 or count not in range(1, 21) or offset not in range(10):
raise ValueError("无效的查询、数量或偏移值")
# API 请求配置
url = "https://api.search.brave.com/res/v1/web/search"
params = {"q": query, "count": count, "offset": offset, "safesearch": safety_level}
if result_type != "all":
params["result_filter"] = result_type
if freshness != "all":
params["freshness"] = freshness
headers = {"Accept": "application/json", "X-Subscription-Token": BRAVE_API_KEY}
# 发送请求
response = requests.get(url, params=params, headers=headers)
if not response.ok:
raise ValueError(f"API 错误: {response.status_code}")
# 处理结果
data = response.json()
results = []
for rtype in ["web", "news", "videos"]:
if result_type in ["all", rtype] and data.get(rtype, {}).get("results"):
for r in data[rtype]["results"]:
result_str = f"标题: {r['title']}\n描述: {r['description']}\nURL: {r['url']}"
if r.get('published'):
result_str += f"\n发布时间: {r['published']}"
results.append(result_str)
return "\n\n".join(results) or "未找到结果"
关键参数说明:
-
result_type
:指定结果类型(web/news/videos/all) -
safety_level
:安全级别(strict/moderate/off) -
freshness
:结果新鲜度(day/week/month/year/all) -
count
:返回结果数量(1-20) -
offset
:分页偏移量(0-9)
3.3.2 本地搜索工具实现
@mcp.tool()
def brave_local_search(query: str, count: int = 5, safety_level: str = "moderate") -> str:
"""使用 Brave Local Search API 查找本地商家"""
check_rate_limit()
if len(query) > 400 or count not in range(1, 21):
raise ValueError("无效的查询或数量")
# 第一步:获取位置ID
url = "https://api.search.brave.com/res/v1/web/search"
params = {
"q": query,
"search_lang": "en",
"result_filter": "locations",
"count": count,
"safesearch": safety_level
}
headers = {"Accept": "application/json", "X-Subscription-Token": BRAVE_API_KEY}
response = requests.get(url, params=params, headers=headers)
if not response.ok:
raise ValueError(f"API 错误: {response.status_code}")
data = response.json()
location_ids = [r["id"] for r in data.get("locations", {}).get("results", [])]
# 无结果时回退到网页搜索
if not location_ids:
return brave_web_search(query, count, 0, "all", safety_level, "all")
# 第二步:获取POI详细信息
poi_url = "https://api.search.brave.com/res/v1/local/pois"
poi_response = requests.get(poi_url, params={"ids": location_ids}, headers=headers)
if not poi_response.ok:
raise ValueError(f"API 错误: {poi_response.status_code}")
# 第三步:获取描述信息
desc_url = "https://api.search.brave.com/res/v1/local/descriptions"
desc_response = requests.get(desc_url, params={"ids": location_ids}, headers=headers)
desc_data = desc_response.json().get("descriptions", {}) if desc_response.ok else {}
# 结果格式化
results = []
for loc in poi_response.json().get("results", []):
address_components = [
loc.get("address", {}).get("streetAddress", ""),
loc.get("address", {}).get("addressLocality", ""),
loc.get("address", {}).get("addressRegion", ""),
loc.get("address", {}).get("postalCode", "")
]
address = ", ".join(filter(None, address_components)) or "无地址信息"
rating_value = loc.get('rating', {}).get('ratingValue', '无评分')
rating_count = loc.get('rating', {}).get('ratingCount', 0)
result_entry = (
f"名称: {loc.get('name', '未知')}\n"
f"地址: {address}\n"
f"电话: {loc.get('phone', '无电话')}\n"
f"评分: {rating_value} ({rating_count}条评价)\n"
f"描述: {desc_data.get(loc.get('id'), '无描述')}"
)
results.append(result_entry)
return "\n---\n".join(results) or "未找到本地结果"
三阶段搜索流程:
-
通过位置搜索API获取位置ID -
使用POI API获取详细商业信息 -
通过描述API获取补充描述文本
3.3.3 启动 MCP 服务器
if __name__ == "__main__":
mcp.run(transport="stdio")
此代码启动 MCP 服务器,通过标准输入/输出处理查询请求。
四、系统部署与交互实践
4.1 代理部署与访问
-
在 Agentverse 注册代理:
https://agentverse.ai/agents/details/agent1qgfnnx5nwxspd55e3zjwtra2gegdt77edf254k47tkcl0nc9dv2zvj6jjhj/profile -
通过 Agentverse 聊天界面与代理交互:
4.2 实际使用示例
网络搜索示例:
brave_web_search(
query="2025年AI突破",
count=5,
result_type="news",
freshness="pm", # 过去一个月
safety_level="strict"
)
返回结果示例:
标题: 2025年人工智能预测:改变行业的5大突破
描述: 专家预测到2025年,AI将在医疗、交通等领域带来革命性变化...
URL: https://example.com/ai-predictions-2025
发布时间: 2025-03-15T08:30:00Z
本地搜索示例:
brave_local_search("旧金山寿司店", count=3)
返回结果示例:
名称: 寿司天堂
地址: 123 Main St, 旧金山, CA 94105
电话: (555) 123-4567
评分: 4.5 (128条评价)
描述: 提供正宗江户前寿司,使用每日新鲜渔获...
---
名称: 东京寿司吧
地址: 456 Elm St, 旧金山, CA 94110
电话: (555) 987-6543
评分: 4.2 (95条评价)
描述: 现代与传统结合的寿司体验...
4.3 组件协作原理

-
用户请求:通过聊天界面或API发送查询 -
uAgent 处理:代理接收请求并路由到 MCP 服务器 -
MCP 服务器:调用相应的 Brave 搜索工具 -
Brave API:执行实际搜索操作 -
结果返回:格式化结果并通过代理返回用户
五、应用场景与价值分析
5.1 对话式AI增强

将搜索代理集成到聊天机器人中:
-
动态回答时事问题 -
提供本地商业推荐 -
获取最新研究报告 -
替代传统知识库的静态回复
5.2 本地化服务应用
创建基于位置的智能服务:
-
旅行助手:实时查找附近酒店、餐厅 -
商业发现:按特定需求(如”无障碍设施餐厅”)筛选场所 -
紧急服务:快速定位最近的医院、药房
5.3 研究分析工具
构建专业研究助手:
-
学术趋势追踪:监控特定领域最新论文 -
竞争情报:收集行业动态和对手信息 -
内容聚合:自动整理多来源信息
六、最佳实践与注意事项
6.1 速率限制策略
Brave API 有严格限制:
-
每秒最多1次请求 -
每月最多15,000次请求 -
实现 check_rate_limit()
函数进行控制
优化建议:
-
缓存频繁查询结果 -
合理安排请求时间分布 -
监控使用量接近上限时发出警报
6.2 错误处理增强
在生产环境中应增加:
-
API 响应异常处理 -
网络超时重试机制 -
备用搜索源切换 -
详细错误日志记录
6.3 结果优化技巧
提升搜索结果质量:
# 在 brave_web_search 中添加结果过滤
if "付费推广" in r.get('extra', {}).get('sponsored', False):
continue # 跳过赞助结果
七、总结与拓展方向
本文详细介绍了如何使用 Brave Search API 和 uAgents 框架构建智能搜索代理。核心要点包括:
-
模块化架构:通过 MCP 服务器解耦搜索功能与代理逻辑 -
双模式搜索:无缝切换网络搜索和本地商业搜索 -
智能回退:本地无结果时自动切换至网络搜索 -
轻量级部署:基于 Python 的简洁实现
未来拓展方向:
-
增加多语言搜索支持 -
集成个性化推荐算法 -
添加用户偏好记忆功能 -
开发浏览器扩展插件
资源与社区
-
ASI:One 模型:https://asi1.ai/ -
Agentverse 平台:https://agentverse.ai/ -
uAgents 文档:https://uagents.fetch.ai/docs -
Fetch.ai 开发者中心:https://fetch.ai/ -
Brave Search API:https://brave.com/search/api/
实践是掌握技术的最佳途径。建议从修改搜索参数开始,逐步尝试添加新功能,如结果排序或自定义过滤。遇到挑战时,欢迎加入 Fetch.ai 开发者社区 交流探讨,共同推进智能代理技术的发展。