如何解决Claude API的128000 Token限制：分块处理与流式响应实战

1次阅读

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

在使用 Claude API 进行长文本生成时，开发者经常会遇到 api error: claude's response exceeded the 128000 output token maximum 错误。这个限制意味着单个 API 响应不能超过 128,000 个 token（大约相当于 96,000 个英文单词或 64,000 个中文字）。

这种限制对以下场景影响尤为显著：

生成长篇文档或报告
处理大型代码库的分析
执行复杂的研究论文摘要
进行大规模数据集的文本处理

超过这个限制会导致 API 请求直接失败，无法获取完整的响应内容。

分块处理的核心思想是将输入文本拆分为多个符合 token 限制的较小片段，然后分别发送请求，最后合并所有响应。

优势：

实现相对简单
适用于所有 API 版本
可以精确控制每个请求的大小

劣势：

需要额外的拼接逻辑
可能增加总体延迟
上下文连续性可能受影响

流式响应利用 API 的 streaming 功能逐步获取输出，避免一次性接收所有内容。

优势：

可以更快地获取部分结果
减少内存压力
更自然的用户体验

劣势：

需要处理更复杂的响应逻辑
不适用于所有使用场景
实现难度较高

import tiktoken

def count_tokens(text: str, model: str = "gpt-4") -> int:
    """计算文本的 token 数量"""
    encoder = tiktoken.encoding_for_model(model)
    return len(encoder.encode(text))

def smart_chunking(text: str, max_tokens: int = 120000) -> list:
    """
    将长文本智能分块，确保不在句子中间拆分
    :param text: 输入文本
    :param max_tokens: 每个分块的最大 token 数
    :return: 分块后的文本列表
    """paragraphs = text.split('\n\n')
    chunks = []
    current_chunk = []
    current_count = 0

    for para in paragraphs:
        para_count = count_tokens(para)

        # 如果当前段落就超过限制（极罕见情况）if para_count > max_tokens:
            # 进一步按句子分割
            sentences = para.split('.')
            temp_para = ""
            for sent in sentences:
                sent_count = count_tokens(sent)
                if current_count + sent_count > max_tokens:
                    chunks.append('\n\n'.join(current_chunk))
                    current_chunk = [sent]
                    current_count = sent_count
                else:
                    current_chunk.append(sent)
                    current_count += sent_count
        elif current_count + para_count > max_tokens:
            chunks.append('\n\n'.join(current_chunk))
            current_chunk = [para]
            current_count = para_count
        else:
            current_chunk.append(para)
            current_count += para_count

    if current_chunk:
        chunks.append('\n\n'.join(current_chunk))

    return chunks

import requests

def make_chunked_request(api_key: str, prompt: str, model: str = "claude-v1") -> str:
    """
    分块处理长文本请求
    :param api_key: API 密钥
    :param prompt: 输入的提示文本
    :param model: 使用的模型版本
    :return: 拼接后的完整响应
    """
    chunks = smart_chunking(prompt)
    full_response = ""

    for chunk in chunks:
        headers = {"Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }

        data = {
            "prompt": chunk,
            "model": model,
            "max_tokens": 4000  # 控制每个分块响应的最大长度
        }

        try:
            response = requests.post(
                "https://api.anthropic.com/v1/complete",
                headers=headers,
                json=data
            )
            response.raise_for_status()
            full_response += response.json()["completion"] + "\n\n"
        except Exception as e:
            print(f"处理分块时出错: {str(e)}")
            # 这里可以添加重试逻辑

    return full_response.strip()

import json

def stream_response(api_key: str, prompt: str, model: str = "claude-v1") -> str:
    """
    使用流式 API 获取长响应
    :param api_key: API 密钥
    :param prompt: 输入的提示文本
    :param model: 使用的模型版本
    :return: 完整的响应内容
    """headers = {"Authorization": f"Bearer {api_key}","Content-Type":"application/json","Accept":"text/event-stream"
    }

    data = {
        "prompt": prompt,
        "model": model,
        "stream": True
    }

    full_response = ""

    try:
        with requests.post(
            "https://api.anthropic.com/v1/complete",
            headers=headers,
            json=data,
            stream=True
        ) as response:
            response.raise_for_status()

            for line in response.iter_lines():
                if line:
                    decoded_line = line.decode('utf-8')
                    if decoded_line.startswith('data:'):
                        event_data = json.loads(decoded_line[6:])
                        if "completion" in event_data:
                            full_response += event_data["completion"]
                            # 可以在这里添加实时处理逻辑
                            print(event_data["completion"], end="", flush=True)
    except Exception as e:
        print(f"流式请求出错: {str(e)}")

    return full_response

分块处理：每个分块 1 次 API 调用，n 个分块需要 n 次调用
流式响应：始终只需要 1 次 API 调用

分块处理：总延迟 = 各分块处理时间之和 + 网络延迟×分块数
流式响应：延迟与单次请求相当，但可以边接收边处理

分块处理：各分块独立处理，可能丢失跨分块上下文
流式响应：保持完整上下文，连贯性更好

在分块边界处添加重叠内容（如前一个块的最后几句话复制到下一个块开头）
为每个分块添加上下文摘要
使用较小的分块大小（如 80,000 token）留出上下文空间

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def make_api_request(headers, data):
    response = requests.post(
        "https://api.anthropic.com/v1/complete",
        headers=headers,
        json=data
    )
    response.raise_for_status()
    return response