AI应用开发：那些让人头疼的Token限制和成本优化

做AI应用开发，最头疼的就是Token限制和成本控制了。刚开始我没在意，结果第一个月账单出来，直接傻眼。今天就把我踩过的坑和优化经验都分享出来。

Token到底是个啥？

简单说，Token就是大模型处理文本的基本单位。英文的话，1个Token大约等于0.75个单词。中文比较复杂，1个中文字可能是1-2个Token。

Token的限制主要有两个：

1. 输入Token限制：比如GPT-4是8K，GPT-4 Turbo是128K
2. 输出Token限制：一般默认是几百到几千

超过限制就会报错，而且超出部分还要收费。所以必须得控制好。

成本到底有多高？

我算了一下，用GPT-4的话：

• 输入：$0.03 / 1K tokens
• 输出：$0.06 / 1K tokens

看起来不多，但用起来是真烧钱。我第一个项目，一天就花了100多美金，直接崩溃。

实际案例：

• 一个RAG应用，每次检索5个文档，每个文档500字，问题100字
• 输入大约：5 * 500 + 100 = 2600 tokens
• 输出大约：500 tokens
• 一次请求成本：2600 * 0.03 / 1000 + 500 * 0.06 / 1000 = 0.108 USD
• 1000次请求就是108美金！

这还是保守估计，如果文档更长，成本更高。

Token计算：怎么知道用了多少？

不同模型的Token计算方式不一样，但基本思路是一样的。

Java实现Token计算

public class TokenCalculator {
    
    // 粗略估算（实际应该用tiktoken库，但Java没有官方实现）
    public int estimateTokens(String text) {
        // 英文：1 token ≈ 4字符
        // 中文：1 token ≈ 1.5字符
        int chineseChars = countChineseChars(text);
        int englishChars = text.length() - chineseChars;
        
        int chineseTokens = (int) Math.ceil(chineseChars * 1.5);
        int englishTokens = (int) Math.ceil(englishChars / 4.0);
        
        return chineseTokens + englishTokens;
    }
    
    private int countChineseChars(String text) {
        return (int) text.chars()
            .filter(c -> c >= 0x4E00 && c <= 0x9FFF)
            .count();
    }
    
    // 更准确的方式：调用API计算
    public int calculateTokens(String text, String model) {
        // OpenAI提供了计算Token的API
        // 但Java没有官方SDK，可以用这个库：https://github.com/knuddelsgmbh/jtokkit
        // ...
    }
}

实际项目里，我用了jtokkit库：

<dependency>
    <groupId>com.knuddels</groupId>
    <artifactId>jtokkit</artifactId>
    <version>1.0.0</version>
</dependency>

import com.knuddels.jtokkit.Encodings;
import com.knuddels.jtokkit.api.Encoding;
import com.knuddels.jtokkit.api.EncodingType;

public int calculateTokens(String text, String model) {
    Encoding encoding = Encodings.newDefaultEncodingRegistry()
        .getEncodingForModel(model);
    return encoding.countTokens(text);
}

成本优化策略

1. 压缩输入内容

这是最直接的方法。很多场景下，输入的内容其实可以精简。

文档摘要：

public String summarize(String text, int maxTokens) {
    // 如果文档太长，先摘要
    int currentTokens = calculateTokens(text);
    if (currentTokens <= maxTokens) {
        return text;
    }
    
    // 提取关键句（简单实现）
    String[] sentences = text.split("[。！？]");
    List<String> importantSentences = extractImportant(sentences);
    
    StringBuilder summary = new StringBuilder();
    for (String sentence : importantSentences) {
        if (calculateTokens(summary.toString() + sentence) > maxTokens) {
            break;
        }
        summary.append(sentence).append("。");
    }
    
    return summary.toString();
}

去除冗余：

public String removeRedundancy(String text) {
    // 去除重复段落
    String[] paragraphs = text.split("\\n\\s*\\n");
    Set<String> seen = new LinkedHashSet<>();
    
    for (String para : paragraphs) {
        if (!seen.contains(para)) {
            seen.add(para);
        }
    }
    
    return String.join("\n\n", seen);
}

2. 智能截断

当输入超过限制时，不是简单截断，而是保留最重要的部分。

public String smartTruncate(String text, int maxTokens, String query) {
    // 计算每个段落和查询的相关性
    String[] paragraphs = text.split("\\n\\s*\\n");
    
    List<ScoredParagraph> scored = new ArrayList<>();
    for (String para : paragraphs) {
        double score = calculateRelevance(para, query);
        scored.add(new ScoredParagraph(para, score));
    }
    
    // 按相关性排序
    scored.sort((a, b) -> Double.compare(b.score, a.score));
    
    // 选择最相关的段落，直到达到Token限制
    StringBuilder result = new StringBuilder();
    for (ScoredParagraph sp : scored) {
        if (calculateTokens(result.toString() + sp.text) > maxTokens) {
            break;
        }
        result.append(sp.text).append("\n\n");
    }
    
    return result.toString();
}

3. 缓存机制

相同的问题没必要每次都调用API，缓存起来。

@Service
public class CachedChatService {
    
    @Autowired
    private ChatClient chatClient;
    
    @Autowired
    private RedisTemplate<String, String> redis;
    
    public String chat(String message) {
        // 生成缓存key（可以用MD5）
        String cacheKey = "chat:" + DigestUtils.md5Hex(message);
        
        // 先查缓存
        String cached = redis.opsForValue().get(cacheKey);
        if (cached != null) {
            return cached;
        }
        
        // 缓存未命中，调用API
        String response = chatClient.call(message);
        
        // 存入缓存（设置过期时间，比如1天）
        redis.opsForValue().set(cacheKey, response, 1, TimeUnit.DAYS);
        
        return response;
    }
}

4. 使用更便宜的模型

对于不需要太强能力的场景，可以用便宜点的模型。

@Service
public class ModelSelector {
    
    @Autowired
    private ChatClient gpt4Client;
    
    @Autowired
    private ChatClient gpt35Client; // 便宜10倍
    
    public String chat(String message, boolean needsStrongModel) {
        if (needsStrongModel) {
            return gpt4Client.call(message);
        } else {
            // 简单问题用便宜的模型
            return gpt35Client.call(message);
        }
    }
    
    // 根据问题复杂度自动选择
    public String smartChat(String message) {
        double complexity = estimateComplexity(message);
        return chat(message, complexity > 0.7);
    }
    
    private double estimateComplexity(String message) {
        // 简单规则：长度、关键词等
        int length = message.length();
        boolean hasComplexKeywords = message.contains("分析") || 
                                     message.contains("比较") ||
                                     message.contains("解释");
        
        if (hasComplexKeywords && length > 100) {
            return 0.8;
        }
        return 0.3;
    }
}

5. 批量处理

如果有多个相似请求，可以合并处理。

public List<String> batchChat(List<String> messages) {
    // 如果问题相似，合并成一个prompt
    if (areSimilar(messages)) {
        String combinedPrompt = combineMessages(messages);
        String response = chatClient.call(combinedPrompt);
        return splitResponse(response, messages.size());
    }
    
    // 不相似，单独处理
    return messages.parallelStream()
        .map(chatClient::call)
        .collect(Collectors.toList());
}

6. 流式输出 + 提前终止

如果用户已经满意了，可以提前终止，节省Token。

public void streamChat(String message, Consumer<String> onChunk, 
                       Predicate<String> shouldStop) {
    chatClient.stream(message)
        .takeWhile(chunk -> {
            String content = chunk.getResult().getOutput().getContent();
            onChunk.accept(content);
            return !shouldStop.test(content);
        })
        .subscribe();
}

7. 限制输出长度

设置max_tokens参数，避免生成过长的回答。

public String chatWithLimit(String message, int maxTokens) {
    // Spring-AI中可以这样配置
    ChatOptions options = ChatOptions.builder()
        .withMaxTokens(maxTokens)
        .build();
    
    return chatClient.call(
        Prompt.builder()
            .withMessage(message)
            .withOptions(options)
            .build()
    );
}

监控和告警

成本控制必须要有监控，不然什么时候超预算了都不知道。

@Service
public class TokenMonitor {
    
    @Autowired
    private RedisTemplate<String, Long> redis;
    
    public void recordTokens(String apiKey, int inputTokens, int outputTokens) {
        String today = LocalDate.now().toString();
        String key = "tokens:" + apiKey + ":" + today;
        
        // 累计Token使用量
        redis.opsForValue().increment(key + ":input", inputTokens);
        redis.opsForValue().increment(key + ":output", outputTokens);
        
        // 计算成本
        double cost = inputTokens * 0.03 / 1000.0 + outputTokens * 0.06 / 1000.0;
        redis.opsForValue().increment(key + ":cost", (long)(cost * 100)); // 用分存储
        
        // 检查是否超限
        checkLimit(apiKey, today);
    }
    
    private void checkLimit(String apiKey, String date) {
        String costKey = "tokens:" + apiKey + ":" + date + ":cost";
        Long costInCents = redis.opsForValue().get(costKey);
        
        if (costInCents != null && costInCents > 10000) { // 超过100美金
            // 发送告警
            sendAlert(apiKey, "每日成本超限：" + costInCents / 100.0 + " USD");
        }
    }
}

实际案例：RAG系统的成本优化

我的RAG系统优化前后对比：

优化前：

• 每次检索：5个文档 * 1000 tokens = 5000 tokens
• 问题：200 tokens
• 回答：800 tokens
• 单次成本：0.174 USD
• 日均1000次请求 = 174 USD/天

优化后：

• 文档摘要：5个文档 * 300 tokens = 1500 tokens（减少70%）
• 问题：200 tokens
• 回答：500 tokens（限制长度）
• 单次成本：0.063 USD（减少64%）
• 加上缓存命中率30%，实际成本更低

优化措施：

1. 文档存入前先摘要
2. 智能选择最相关的3个文档（而不是5个）
3. 限制回答长度
4. 加入缓存（30%命中率）
5. 简单问题用GPT-3.5

最终日均成本降到了50 USD左右，节省了70%以上。

踩坑总结

1. Token计算不准：刚开始用简单的字符数估算，结果差了很多。后来用了专业的库才准。

2. 缓存策略不对：刚开始缓存时间太长，结果用户反馈回答不够实时。后来改成1天过期，加上手动刷新机制。

3. 没有监控：第一个月账单出来才发现超预算了。现在每天都看成本报表。

4. 模型选择不当：什么任务都用GPT-4，浪费钱。现在会根据任务复杂度选模型。

总结

Token限制和成本控制是AI应用开发必须面对的挑战。但通过合理的优化策略，完全可以控制成本。关键是：

1. 监控使用量
2. 优化输入输出
3. 合理使用缓存
4. 选择合适的模型
5. 设置预算告警

好了，今天就聊到这里。如果你也在做AI应用，欢迎分享成本优化的经验。完整代码我放在GitHub上了，需要的同学可以看看。