代理搜索失败是人工智能代理产生错误或不完整答案的最常见原因。故障模式是可预测的:API 密钥配置错误、格式错误的查询、空结果解析、慢速网络超时和速率限制。本教程提供了系统的调试清单和可重复使用的诊断功能,您可以将其放入任何代理中。每个测试都使用 Scavio API 端点,但模式适用于任何搜索后端。
前置条件
- 已安装 Python 3.9+
- 请求已安装库
- 来自 scavio.dev 的 Scavio API 密钥
- 具有搜索工具集成以进行调试的代理
操作指南
步骤 1: 测试 API 连接和身份验证
首先要检查的是您的 API 密钥是否有效以及端点是否可访问。此函数运行最低程度的运行状况检查。
import requests, os
def test_api_health(api_key: str = None) -> dict:
key = api_key or os.environ.get('SCAVIO_API_KEY', '')
if not key:
return {'status': 'FAIL', 'error': 'No API key found'}
try:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': key, 'Content-Type': 'application/json'},
json={'query': 'test', 'country_code': 'us'},
timeout=10)
return {'status': 'OK' if resp.status_code == 200 else 'FAIL',
'http_code': resp.status_code,
'has_results': bool(resp.json().get('organic_results'))}
except requests.exceptions.Timeout:
return {'status': 'FAIL', 'error': 'Timeout after 10s'}
except Exception as e:
return {'status': 'FAIL', 'error': str(e)}
print(test_api_health())步骤 2: 诊断空结果
如果 API 返回 200 但没有结果,则查询可能过于具体、格式错误或所在区域设置没有覆盖。使用逐渐简单的查询进行测试。
def diagnose_empty_results(original_query: str) -> dict:
api_key = os.environ['SCAVIO_API_KEY']
test_queries = [
original_query,
' '.join(original_query.split()[:3]), # first 3 words
'python programming', # known good query
]
results = {}
for q in test_queries:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': api_key, 'Content-Type': 'application/json'},
json={'query': q, 'country_code': 'us'})
count = len(resp.json().get('organic_results', []))
results[q] = count
return results
diag = diagnose_empty_results('xylophone quantum blockchain synergy 2026')
for query, count in diag.items():
status = 'OK' if count > 0 else 'EMPTY'
print(f'[{status}] "{query}" -> {count} results')步骤 3: 将结构化日志记录添加到搜索工具
使用日志记录封装您的搜索功能,以便您可以跟踪每个调用、其延迟、结果计数和任何错误。
import time, logging
logging.basicConfig(level=logging.INFO)
log = logging.getLogger('agent_search')
def logged_search(query: str, country: str = 'us') -> dict:
start = time.time()
log.info(f'Searching: "{query}" country={country}')
try:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': os.environ['SCAVIO_API_KEY'],
'Content-Type': 'application/json'},
json={'query': query, 'country_code': country},
timeout=15)
elapsed = time.time() - start
data = resp.json()
count = len(data.get('organic_results', []))
log.info(f'Result: {resp.status_code}, {count} results, {elapsed:.2f}s')
return data
except Exception as e:
elapsed = time.time() - start
log.error(f'Failed after {elapsed:.2f}s: {e}')
return {'organic_results': [], 'error': str(e)}步骤 4: 构建具有指数退避的重试包装器
暂时性故障(网络故障、速率限制)最好通过重试来处理。使用指数退避以避免对 API 造成影响。
def search_with_retry(query: str, max_retries: int = 3) -> dict:
for attempt in range(max_retries):
try:
data = logged_search(query)
if data.get('organic_results'):
return data
if attempt < max_retries - 1:
wait = 2 ** attempt
log.warning(f'Empty results, retrying in {wait}s...')
time.sleep(wait)
except Exception as e:
if attempt < max_retries - 1:
wait = 2 ** attempt
log.warning(f'Error: {e}, retrying in {wait}s...')
time.sleep(wait)
else:
raise
return {'organic_results': [], 'error': 'All retries exhausted'}步骤 5: 运行完整的诊断套件
将所有检查合并到一个诊断功能中,报告代理搜索集成的运行状况。
def full_diagnostic() -> None:
print('=== Agent Search Diagnostic ===')
# 1. API Health
health = test_api_health()
print(f'1. API Health: {health["status"]}')
if health['status'] != 'OK':
print(f' Error: {health.get("error", health.get("http_code"))}')
return
# 2. Result quality
data = logged_search('best crm software 2026')
results = data.get('organic_results', [])
print(f'2. Result quality: {len(results)} results')
if results:
print(f' Top result: {results[0]["title"]}')
# 3. Retry behavior
retry_data = search_with_retry('obscure test query 12345')
print(f'3. Retry test: {len(retry_data.get("organic_results", []))} results')
print('=== Diagnostic complete ===')
full_diagnostic()Python 示例
import os, time, requests, logging
logging.basicConfig(level=logging.INFO)
log = logging.getLogger('debug')
API_KEY = os.environ['SCAVIO_API_KEY']
def search(query: str) -> dict:
start = time.time()
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': 'us'}, timeout=15)
elapsed = time.time() - start
data = resp.json()
count = len(data.get('organic_results', []))
log.info(f'"{query}" -> {count} results in {elapsed:.2f}s')
return data
def main():
# Health check
health = search('test')
print(f'Health: {"OK" if health.get("organic_results") else "FAIL"}')
# Real query
data = search('best crm 2026')
for r in data.get('organic_results', [])[:3]:
print(f' #{r["position"]} {r["title"]}')
if __name__ == '__main__':
main()JavaScript 示例
const API_KEY = process.env.SCAVIO_API_KEY;
async function search(query) {
const start = Date.now();
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({ query, country_code: 'us' })
});
const data = await resp.json();
const ms = Date.now() - start;
console.log(`"${query}" -> ${(data.organic_results || []).length} results in ${ms}ms`);
return data;
}
async function main() {
const health = await search('test');
console.log(`Health: ${health.organic_results?.length ? 'OK' : 'FAIL'}`);
const data = await search('best crm 2026');
(data.organic_results || []).slice(0, 3).forEach(r => {
console.log(` #${r.position} ${r.title}`);
});
}
main().catch(console.error);预期输出
=== Agent Search Diagnostic ===
1. API Health: OK
INFO:agent_search:Searching: "best crm software 2026" country=us
INFO:agent_search:Result: 200, 10 results, 0.87s
2. Result quality: 10 results
Top result: Best CRM Software for Small Business (2026)
3. Retry test: 0 results
=== Diagnostic complete ===