引言

在构建基于 Anthropic Claude 模型的应用时，提示（Prompt）的长度管理是确保系统稳定、经济且高效的关键环节。不同 Claude 模型拥有各异的上下文窗口、处理速度和调用成本。若提交的提示长度超出模型的最大 token 限制，API 调用将直接失败；若不加区分地使用同一模型，则可能造成成本浪费或响应延迟。通过预先统计 token 数量并动态选择模型（即模型分流），开发者可以在避免错误的同时，优化成本与延迟。
通过 Anthropic SDK 的 count_tokens 方法，开发者可以在实际调用 API 之前准确获知提示的 token 长度，并基于此动态选择合适的模型。这种方法能够有效避免因超出上下文窗口而导致的请求失败，同时实现成本与延迟的优化。结合业务需求，还可以进一步细化分流策略，例如根据任务类型、用户等级等因素调整模型选择。
本文将详细阐述利用 Anthropic Python SDK 的 count_tokens 方法实现 token 统计和模型分流的原理、方法，并提供完整的代码示例，帮助读者构建智能的模型路由机制。

1. 为什么需要 Token 统计与模型分流？

1.1 模型上下文窗口限制

每个 Claude 模型都有其最大 token 处理能力（上下文窗口）。例如（以 Claude 3 系列为例）：

  • Claude 3 Haiku：约 200k token
  • Claude 3 Sonnet：约 200k token
  • Claude 3 Opus：约 200k token
  • Claude 2.1 等旧模型也支持 200k token。

如果提示（包括系统消息、历史对话、用户输入等）的总 token 数超过模型的上限，API 将返回错误，导致请求失败。因此，在调用前准确统计 token 并选择容量足够的模型是必要的。

1.2 成本差异

不同模型的价格差异显著：

  • Haiku：最便宜，适合高频、简单任务。
  • Sonnet：中等价位，平衡性能与成本。
  • Opus：最昂贵，用于复杂推理。

为短小、简单的提示选用低成本模型，可以大幅降低总体支出。

1.3 速度差异

模型推理速度与其参数量相关：Haiku 最快，Opus 最慢。对于实时性要求高的场景（如聊天机器人），应优先选择能够快速响应的模型；而长提示或复杂任务则可交给速度较慢但能力更强的模型。

1.4 分流的好处

  • 避免超限错误：提前确保提示长度在所选模型窗口内。
  • 成本优化：短提示走低成本模型，长提示走大窗口模型（即使成本较高），保证请求成功。
  • 延迟控制：根据提示长度和任务复杂度分流，提升用户体验。
  • 资源合理分配：将不同任务分流至最合适的模型，实现资源利用最大化。

2. 方法：使用 count_tokens 评估提示长度

2.1 Anthropic SDK 中的 token 统计

Anthropic Python SDK 提供了 Anthropic().count_tokens() 方法，用于本地计算提示的 token 数量。该方法直接基于 Claude 的 tokenizer 计算，不发起网络请求，因此快速且免费。它接受字符串（纯文本）或消息列表（符合 Messages API 格式）作为输入，并且支持传入 system 参数。

from anthropic import Anthropic

client = Anthropic(api_key="your-api-key")

# 统计字符串的 token 数
token_count = client.count_tokens("Hello, world!")
print(token_count)

# 统计消息列表的 token 数
messages = [
    {"role": "user", "content": "What is the capital of France?"},
    {"role": "assistant", "content": "The capital of France is Paris."},
]
total_tokens = client.count_tokens(messages=messages)
print(total_tokens)

# 包含系统提示的统计
system_prompt = "You are a helpful assistant."
total_with_system = client.count_tokens(messages=messages, system=system_prompt)

注意：count_tokens 计算的是消息内容和系统提示本身的 token 数量，不包括元数据或 API 调用额外添加的格式 token。但根据官方文档，该计数与 API 实际计费的 token 数高度一致，足以用于预估。

2.2 模型分流策略

根据 token 计数结果，选择适合的模型。可以定义基于 token 数量的阈值，例如：

Token 范围 推荐模型 最大上下文窗口 成本 速度
0 - 4000 claude-3-haiku 200k 低 最快
4001 - 20000 claude-3-sonnet 200k 中 中等
20001 - 200000 claude-3-opus 200k 高 最慢

也可以动态计算：对于任何长度的提示，直接选择能容纳它的最便宜、最快的模型。即：

  • 如果 token ≤ Haiku 的最大窗口 → 使用 Haiku
  • 否则如果 token ≤ Sonnet 的最大窗口 → 使用 Sonnet
  • 否则如果 token ≤ Opus 的最大窗口 → 使用 Opus
  • 否则 → 拒绝请求或提示用户缩短输入。

由于 Claude 3 系列模型最大窗口均为 200k（假设一致），上述阈值实际上只依赖于提示长度是否超出 200k。但实践中，开发者可能希望根据提示的“复杂性”或任务类型来分流，而不仅仅是 token 数。例如，简单问答即使很长也可以使用 Haiku，而复杂推理即使很短也可以使用 Opus。因此，分流逻辑应结合业务需求灵活设计。为简化，本文示例将基于 token 数量选择模型，同时确保不超过模型窗口。

3. 具体代码实现

3.1 环境准备

首先安装 anthropic 库：

pip install anthropic

3.2 初始化客户端

import os
from anthropic import Anthropic

# 从环境变量获取 API 密钥（推荐安全实践）
api_key = os.environ.get("ANTHROPIC_API_KEY")
client = Anthropic(api_key=api_key)

3.3 定义分流函数

def select_model_by_token_count(token_count: int) ->str:
    """
    根据 token 数量选择最合适且能容纳该长度的模型。
    如果超出所有模型窗口，抛出异常。
    """
    # 模型最大上下文（以 Claude 3 为例，可根据官方文档更新）
    MODEL_MAX_TOKENS = {
        "claude-3-haiku-20240307": 200000,
        "claude-3-sonnet-20240229": 200000,
        "claude-3-opus-20240229": 200000,
    }

    # 按成本和速度排序的模型列表（从低到高）
    models_in_order = ["claude-3-haiku-20240307", "claude-3-sonnet-20240229", "claude-3-opus-20240229"]

    for model in models_in_order:
        if token_count <= MODEL_MAX_TOKENS[model]:
            return model

    # 如果 token 数超过所有模型窗口，抛出异常
    raise ValueError(f"Prompt too long: {token_count} tokens exceeds all model limits.")

3.4 组合：统计 token + 调用模型

以下函数展示了完整的流程：统计 token、选择模型、发起 API 调用。

def ask_claude_with_token_routing(user_prompt: str, system_prompt: str = ""):
    """
    统计 token，根据 token 数量选择模型，然后发送请求并返回响应。
    """
    # 构建消息列表（当前仅包含单条用户消息，可扩展多轮对话）
    messages = [{"role": "user", "content": user_prompt}]

    # 统计 token（包括系统提示和消息）
    total_tokens = client.count_tokens(messages=messages, system=system_prompt)
    print(f"Total tokens: {total_tokens}")

    # 选择模型
    model = select_model_by_token_count(total_tokens)
    print(f"Selected model: {model}")

    # 调用 API
    response = client.messages.create(
        model=model,
        system=system_prompt,   # 系统提示单独传入
        messages=messages,
        max_tokens=1000,        # 可根据需要调整
        temperature=0.7,
    )

    # 提取回复内容
    answer = response.content[0].text
    return answer

3.5 使用示例

if __name__ == "__main__":
    user_input = "Explain quantum computing in simple terms."
    system = "You are a helpful assistant."

    try:
        result = ask_claude_with_token_routing(user_input, system)
        print(f"Assistant: {result}")
    except Exception as e:
        print(f"Error: {e}")

3.6 注意事项

  • Token 统计准确性：count_tokens 需要传入与后续调用完全相同的参数结构（messages, system, tools 等），以确保计数准确。示例中将 system 单独传入，符合 API 规范。
  • 错误处理：当提示超过所有模型窗口时，应优雅地处理（如提示用户缩短输入或记录错误）。
  • 成本控制：可在 select_model 中加入更多业务逻辑，例如对特定任务固定使用 Haiku，即使 token 较多也允许（只要不超限），以控制成本。
  • 模型版本：模型名称和最大上下文窗口可能随官方更新而变化，建议从配置或环境变量中读取，而非硬编码。
  • 异步支持：如果需要高并发，可使用 AsyncAnthropic 客户端。

4. 完整例子：交互式问答机器人

下面是一个更完整的命令行交互式问答机器人，每次用户输入后自动统计 token、分流并回复。

import os
from anthropic import Anthropic

class ClaudeRouter:
    def __init__(self, api_key=None):
        self.client = Anthropic(api_key=api_key or os.environ.get("ANTHROPIC_API_KEY"))
        self.model_max_tokens = {
            "claude-3-haiku-20240307": 200000,
            "claude-3-sonnet-20240229": 200000,
            "claude-3-opus-20240229": 200000,
        }
        self.models_by_cost = ["claude-3-haiku-20240307", "claude-3-sonnet-20240229", "claude-3-opus-20240229"]

    def count_tokens(self, messages, system=""):
        """统计 token，支持 system 和 messages"""
        return self.client.count_tokens(messages=messages, system=system)

    def select_model(self, token_count):
        for model in self.models_by_cost:
            if token_count <= self.model_max_tokens[model]:
                return model
        raise ValueError(f"Token count {token_count} exceeds all available models.")

    def ask(self, user_message, system_prompt="You are a helpful assistant.", max_tokens=1000, temperature=0.7):
        # 构建消息列表（可以扩展支持多轮对话）
        messages = [{"role": "user", "content": user_message}]

        # 统计 token
        token_count = self.count_tokens(messages=messages, system=system_prompt)
        print(f"[Info] Token count: {token_count}")

        # 选择模型
        model = self.select_model(token_count)
        print(f"[Info] Using model: {model}")

        # 调用 API
        response = self.client.messages.create(
            model=model,
            system=system_prompt,
            messages=messages,
            max_tokens=max_tokens,
            temperature=temperature,
        )
        return response.content[0].text

def main():
    router = ClaudeRouter()
    print("Claude Router Bot (type 'quit' to exit)")
    while True:
        user_input = input("\nYou: ").strip()
        if user_input.lower() in ("quit", "exit"):
            break
        if not user_input:
            continue
        try:
            reply = router.ask(user_input)
            print(f"Assistant: {reply}")
        except Exception as e:
            print(f"Error: {e}")

if __name__ == "__main__":
    main()

运行效果示例：

Claude Router Bot (type 'quit' to exit)

You: Hello
[Info] Token count: 5
[Info] Using model: claude-3-haiku-20240307
Assistant: Hello! How can I assist you today?

You: Explain the theory of relativity in detail.
[Info] Token count: 9
[Info] Using model: claude-3-haiku-20240307
Assistant: (详细回答...)
# 算法 # 开发语言 # 人工智能 # 软件工程 # 性能优化
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

通过自主 IT 平台和 Elastic 迈出可观测性的下一步

avatar AtomGit开源社区
cover

Python基础 - for循环的基本语法与遍历思想

avatar AtomGit开源社区
cover

锂离子电池二阶RC模型；simulink模型 可用来仿真电池端电压变化 用来验证辨识参数的正确性 或进行卡尔曼滤波估计SOC。 (1)

avatar AtomGit开源社区