Agents调用工具时如何高效处理Stream数据：原理与实战避坑指南

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

同步调用引发的血案

某天气查询 Agent 在暴雨季节频繁崩溃，追踪发现当返回超长预报数据时，内存占用从 200MB 飙升至 2GB。根本原因是同步获取了包含十年历史数据的 JSON 响应（约 800MB），而开发者未对响应体大小做校验。这种全量加载（bulk loading）模式在服务端渲染场景尤为致命。

流式与非流式模式性能对比

核心实现方案

异步流式处理器

async def stream_fetch(url: str, chunk_size: int = 1024):
    """
    :param url: 目标 API 地址
    :param chunk_size: 单次读取块大小 (字节)
    :yield: 按 chunk 返回二进制数据
    """
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            resp.raise_for_status()
            async for chunk in resp.content.iter_chunked(chunk_size):
                yield chunk  # 关键点：通过异步迭代器逐步消费 

超时重试装饰器

def retry(max_attempts=3, timeout=5.0):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts+1):
                try:
                    return await asyncio.wait_for(func(*args, **kwargs), 
                        timeout=timeout
                    )
                except Exception as e:
                    if attempt == max_attempts:
                        raise
                    await asyncio.sleep(attempt*0.5)
        return wrapper
    return decorator

响应体校验 Schema

class WeatherSchema(BaseModel):
    timestamp: datetime
    temperature: confloat(ge=-50, le=60)  # 温度范围校验
    precipitation: Optional[confloat(ge=0)]

    @validator('timestamp')
    def check_future(cls, v):
        if v > datetime.now() + timedelta(days=10):
            raise ValueError('预报不能超过 10 天')
        return v

性能优化实战

令牌桶限流实现

class TokenBucket:
    def __init__(self, rate: int, capacity: int):
        self._rate = rate  # 令牌生成速率 (个 / 秒)
        self._capacity = capacity  # 桶容量
        self._tokens = capacity
        self._last_refill = time.monotonic()

    async def consume(self, tokens=1):
        while self._tokens < tokens:
            await asyncio.sleep(0.1)
            self._refill()
        self._tokens -= tokens

    def _refill(self):
        now = time.monotonic()
        elapsed = now - self._last_refill
        self._tokens = min(
            self._capacity,
            self._tokens + elapsed * self._rate
        )
        self._last_refill = now

内存占用对比测试

使用 memory_profiler 工具观察两种模式差异：

# 流式模式内存曲线
Line #    Mem usage    Increment
     1     10.2 MiB     10.2 MiB
     2     10.5 MiB      0.3 MiB
     3     10.7 MiB      0.2 MiB

# 非流式模式内存曲线
Line #    Mem usage    Increment
     1     10.2 MiB     10.2 MiB
     2    512.4 MiB    502.2 MiB
     3   1024.1 MiB    511.7 MiB

安全防护要点

1. JSON 注入防护
2. 使用 ijson 库进行流式解析而非直接 eval

3. 设置最大嵌套深度限制

   ijson.parse(urlopen(url), max_depth=10)

4. TLS 最佳实践

5. 复用 ClientSession 降低握手开销
6. 启用 OCSP 装订减少验证延迟

   conn = aiohttp.TCPConnector(
       limit_per_host=5,
       ssl=ssl.create_default_context())

诊断练习

1. 用 Wireshark 抓包分析 HTTP/ 2 的 DATA 帧传输间隔
2. 模拟网络抖动环境测试流式恢复能力
3. 使用 py-spy 生成火焰图定位阻塞点

通过正确处理流式数据，某电商平台的推荐 API 延迟从 3.2 秒降至 800 毫秒，同时内存消耗降低 92%。建议在满足业务需求的前提下，优先考虑渐进式处理方案。