Baidu-Search Skill 实战：如何构建高可用的搜索引擎集成方案

1次阅读

没有评论

共计 1540 个字符，预计需要花费 4 分钟才能阅读完成。

最近在开发一个聚合新闻平台时，我们需要集成百度搜索 API 来补充内容源。但在实际使用中遇到了几个头疼的问题：

API 偶尔会返回 5xx 错误，导致整个推荐流程中断
高峰期响应时间从 200ms 飙升到 2s 以上
解析 HTML 格式的搜索结果时，XPath 规则稍改动就失效

这些问题直接影响了用户体验，于是我们决定重构整个搜索集成层。

首先封装一个带连接池管理的异步客户端，核心特性包括：

自动添加鉴权头
连接超时与读取超时分离配置
支持 GZIP 压缩

import aiohttp
from typing import Dict, Any

class BaiduSearchClient:
    def __init__(self, api_key: str):
        self._session = aiohttp.ClientSession(headers={'Authorization': f'Bearer {api_key}'},
            timeout=aiohttp.ClientTimeout(total=3, connect=1)
        )

    async def search(self, query: str) -> Dict[str, Any]:
        params = {'q': query, 'format': 'json'}
        async with self._session.get(
            'https://api.baidu.com/search', 
            params=params,
            raise_for_status=True
        ) as resp:
            return await resp.json()

针对瞬态故障实现带指数退避的重试：

import asyncio
import random

async def retry_with_backoff(
    func, 
    max_retries: int = 3,
    initial_delay: float = 0.1
):
    retry = 0
    while retry < max_retries:
        try:
            return await func()
        except Exception as e:
            retry += 1
            if retry == max_retries:
                raise

            delay = initial_delay * (2 ** retry) + random.uniform(0, 0.1)
            await asyncio.sleep(delay)

内存缓存：使用 lru_cache 缓存热点查询
Redis 缓存：存储历史搜索结果，设置 TTL
本地磁盘缓存：作为最后防线

使用 locust 压测不同实现方式：

实现方式	QPS	平均延迟	错误率
同步请求	42	230ms	12%
异步基础版	118	85ms	3%
带缓存优化	210	45ms	<1%

# 最优线程池配置
connector = aiohttp.TCPConnector(
    limit=100,  # 最大连接数
    limit_per_host=20,  # 单域名限制
    enable_cleanup_closed=True  # 自动回收连接
)

使用密钥管理系统动态获取
双密钥交替更新
监控调用量异常波动

令牌桶限流（推荐pyrate_limiter）
基于 CPU 使用率的自适应限流
熔断降级开关

def sanitize_results(results):
    blacklist = ['身份证号', '手机号', '银行卡']
    return [
        r for r in results 
        if not any(b in str(r) for b in blacklist)
    ]

在实现语义搜索时，我们发现：