随着时间的推移,许多代理设置都会积累搜索工具:用于网络搜索的 Brave、用于学术论文的 Exa、用于 Google 的 Serper 以及用于特定网站的自定义抓取工具。每个都添加自己的 API 密钥、速率限制、响应格式和计费。整合到单个搜索 API 可以简化代理架构、减少故障模式,并且通常可以降低成本。 Scavio API 通过一个端点和一个 API 密钥涵盖 Google 网络、地图、图像、购物、YouTube、亚马逊和 TikTok。本教程展示如何将多 API 代理迁移到单个后端。
前置条件
- 已安装 Python 3.9+
- 具有多个搜索工具的现有代理
- 来自 scavio.dev 的 Scavio API 密钥
- 了解您当前的搜索工具使用情况
操作指南
步骤 1: 审核您当前的搜索工具堆栈
列出您的代理使用的每个搜索 API、其用途、每月费用和呼叫量。这揭示了可以整合哪些工具。
# Common multi-API agent stack:
current_stack = [
{'tool': 'Brave Search', 'purpose': 'Web search', 'cost': '$5/1K queries',
'monthly_calls': 2000, 'monthly_cost': 10.0},
{'tool': 'Serper', 'purpose': 'Google SERP', 'cost': '$50/mo Pro',
'monthly_calls': 5000, 'monthly_cost': 50.0},
{'tool': 'Exa', 'purpose': 'Research search', 'cost': '$5/1K',
'monthly_calls': 500, 'monthly_cost': 2.5},
{'tool': 'Custom scraper', 'purpose': 'Amazon prices', 'cost': 'Server + proxy',
'monthly_calls': 1000, 'monthly_cost': 30.0},
]
total_calls = sum(t['monthly_calls'] for t in current_stack)
total_cost = sum(t['monthly_cost'] for t in current_stack)
print(f'Current: {len(current_stack)} APIs, {total_calls:,} calls/mo, ${total_cost:.2f}/mo')
consolidated_cost = total_calls * 0.005
print(f'Consolidated: 1 API, {total_calls:,} calls/mo, ${consolidated_cost:.2f}/mo')
print(f'Savings: ${total_cost - consolidated_cost:.2f}/mo ({((total_cost - consolidated_cost) / total_cost * 100):.0f}%)')步骤 2: 构建统一的搜索界面
创建一个函数,通过一个 API 路由不同的搜索类型。网络搜索、地图、购物和 YouTube 都使用具有不同参数的相同端点。
import requests, os
API_KEY = os.environ['SCAVIO_API_KEY']
SEARCH_URL = 'https://api.scavio.dev/api/v1/search'
TIKTOK_URL = 'https://api.scavio.dev/api/v1/tiktok'
def unified_search(query: str, search_type: str = 'web', **kwargs) -> dict:
"""Single search function replacing multiple APIs."""
if search_type == 'tiktok':
endpoint = kwargs.get('endpoint', 'search/videos')
resp = requests.post(f'{TIKTOK_URL}/{endpoint}',
headers={'Authorization': f'Bearer {API_KEY}',
'Content-Type': 'application/json'},
json={'keyword': query, 'count': 10, 'cursor': 0})
else:
body = {'query': query, 'country_code': kwargs.get('country', 'us')}
if search_type != 'web':
body['platform'] = search_type
resp = requests.post(SEARCH_URL,
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
json=body)
resp.raise_for_status()
return resp.json()步骤 3: 为每个旧工具创建直接替代品
编写与旧工具的签名相匹配的包装函数,以便迁移需要最少的代码更改。
# Replace Brave Search:
def brave_search_replacement(query: str) -> list:
data = unified_search(query, 'web')
return [{'title': r['title'], 'url': r['link'], 'description': r.get('snippet', '')}
for r in data.get('organic_results', [])]
# Replace Serper:
def serper_replacement(query: str) -> dict:
data = unified_search(query, 'web')
return {'organic': data.get('organic_results', []),
'peopleAlsoAsk': data.get('people_also_ask', [])}
# Replace custom Amazon scraper:
def amazon_search_replacement(query: str) -> list:
data = unified_search(query, 'amazon')
return data.get('products', [])
# Test all replacements:
for name, fn, q in [
('Web', brave_search_replacement, 'best crm 2026'),
('SERP', serper_replacement, 'crm pricing'),
('Amazon', amazon_search_replacement, 'crm software book'),
]:
result = fn(q)
count = len(result) if isinstance(result, list) else len(result.get('organic', []))
print(f'{name}: {count} results')步骤 4: 更新您的代理工具定义
将旧的工具定义替换为新的统一工具定义。代理的工具调用行为保持不变——只有后端发生了变化。
# Before: multiple tool definitions with different auth
# tools = [brave_tool, serper_tool, exa_tool, amazon_scraper]
# After: one tool with type parameter
from langchain.tools import tool
@tool
def search(query: str, search_type: str = 'web') -> str:
"""Search the web, Amazon, YouTube, Maps, or TikTok.
Use search_type: web, amazon, youtube, maps, or tiktok."""
data = unified_search(query, search_type)
if search_type == 'web':
results = data.get('organic_results', [])[:5]
return '\n'.join(f'{r["title"]}: {r["link"]}' for r in results)
elif search_type == 'amazon':
products = data.get('products', [])[:5]
return '\n'.join(f'{p.get("title", "")}: {p.get("price", "")}' for p in products)
return str(data)
# Register single tool instead of four:
# agent = create_agent(llm, tools=[search])Python 示例
import os, requests
API_KEY = os.environ['SCAVIO_API_KEY']
def unified_search(query: str, search_type: str = 'web') -> dict:
if search_type == 'tiktok':
resp = requests.post('https://api.scavio.dev/api/v1/tiktok/search/videos',
headers={'Authorization': f'Bearer {API_KEY}', 'Content-Type': 'application/json'},
json={'keyword': query, 'count': 10, 'cursor': 0})
else:
body = {'query': query, 'country_code': 'us'}
if search_type != 'web': body['platform'] = search_type
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'}, json=body)
return resp.json()
def main():
for stype, query in [('web', 'best crm 2026'), ('amazon', 'crm book'), ('tiktok', 'crm tips')]:
data = unified_search(query, stype)
print(f'{stype}: {query} -> got response')
if __name__ == '__main__':
main()JavaScript 示例
const API_KEY = process.env.SCAVIO_API_KEY;
async function unifiedSearch(query, type = 'web') {
if (type === 'tiktok') {
const resp = await fetch('https://api.scavio.dev/api/v1/tiktok/search/videos', {
method: 'POST',
headers: { 'Authorization': `Bearer ${API_KEY}`, 'Content-Type': 'application/json' },
body: JSON.stringify({ keyword: query, count: 10, cursor: 0 })
});
return resp.json();
}
const body = { query, country_code: 'us' };
if (type !== 'web') body.platform = type;
const resp = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': API_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify(body)
});
return resp.json();
}
async function main() {
for (const [type, q] of [['web', 'best crm'], ['amazon', 'crm book']]) {
const data = await unifiedSearch(q, type);
console.log(`${type}: got response`);
}
}
main().catch(console.error);预期输出
Current: 4 APIs, 8,500 calls/mo, $92.50/mo
Consolidated: 1 API, 8,500 calls/mo, $42.50/mo
Savings: $50.00/mo (54%)
Web: 10 results
SERP: 10 results
Amazon: 8 results