Claude中断处理实战指南：如何优雅处理interrupted请求

1次阅读

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

在基于 Claude 构建对话应用时，网络波动或服务端问题可能导致请求意外中断(interrupted)。这类问题若不妥善处理，会造成用户重复提交、对话上下文丢失等体验问题，严重时甚至引发数据不一致。尤其在移动端弱网环境下，中断率可能高达 5%-10%，必须建立完善的容错机制。

HTTP 状态码检测：重点关注 503（服务不可用）、502（网关错误）和 504（网关超时）三类状态码
错误类型识别 ：捕获ConnectionError、TimeoutError 等网络层异常
业务标记检测 ：部分 API 会在响应体中包含"interrupted":true 等自定义标记

方案类型	优点	缺点	适用场景
本地存储	零网络延迟	多设备间状态不同步	纯客户端应用
分布式缓存	支持水平扩展	需要处理缓存一致性	高并发服务端应用
数据库存储	数据持久性强	读写性能较低	需要审计日志的场景

import aiohttp
from datetime import datetime, timedelta
import json

class ClaudeClient:
    def __init__(self, api_key):
        self.session = aiohttp.ClientSession()
        self.retry_config = {
            'max_retries': 3,
            'initial_delay': 1.0,
            'backoff_factor': 2.0
        }

    async def _exponential_backoff(self, attempt):
        delay = min(self.retry_config['initial_delay'] * (self.retry_config['backoff_factor'] ** attempt),
            60  # 最大延迟 60 秒
        )
        await asyncio.sleep(delay)

    async def send_request(self, prompt, context=None):
        last_error = None

        for attempt in range(self.retry_config['max_retries'] + 1):
            try:
                payload = {
                    "prompt": prompt,
                    "context": context or {},
                    "request_id": str(uuid.uuid4())  # 幂等键
                }

                async with self.session.post(
                    "https://api.claude.ai/v1/complete",
                    headers={"Authorization": f"Bearer {self.api_key}"},
                    json=payload,
                    timeout=30
                ) as resp:
                    if resp.status in (502, 503, 504):
                        raise ConnectionError(f"Bad status: {resp.status}")

                    data = await resp.json()
                    if data.get("interrupted"):
                        return await self._handle_interruption(payload)

                    return data

            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                last_error = e
                if attempt == self.retry_config['max_retries']:
                    break

                await self._exponential_backoff(attempt)
                continue

        raise last_error or RuntimeError("Unknown error")