OpenAI Assistant流式传输Token计数详解与代码示例

2025-03-18 08:22:39

OpenAI Assistant 流式传输消息时的 Token 计数方法

使用 OpenAI Assistant API 构建应用时，我们经常会用到流式传输（streaming）模式来接收消息更新。这样做的好处是可以实现类似“打字机”的效果，提升用户体验。但一个问题来了，如何在流式传输的过程中计算已消耗的 Token 数量呢？这篇文章就来详细说说这个问题，分析产生原因，并给出解决方案。

一、为什么需要计算 Token？

OpenAI API 的计费是基于 Token 数量的。了解 Token 消耗情况，能帮我们：

控制成本： 准确掌握 API 使用费用，避免超支。
优化性能： 分析不同请求的 Token 消耗，找出潜在的优化点，减少不必要的 Token 使用。
资源限制： OpenAI API 有速率限制和 Token 限制，知道 Token 消耗，就能更好地管理这些限制。

二、流式传输时 Token 计数难点

与普通的一次性请求不同，流式传输是分段接收数据的。streamingUpdate 对象包含一系列增量更新。直接在每一次的增量中试图统计 token 消耗，并不可靠。主要原因：

增量更新的粒度： OpenAI 返回的增量更新可能只包含几个字符、一个单词，甚至只是标点符号，或者没有任何更新。每一次统计都将造成麻烦。
Token 化过程的复杂性： OpenAI 使用特定的 Tokenizer 将文本转换为 Token。Token 并不是简单地对应于单词或字符，它还受到上下文的影响。我们很难在客户端准确地模拟这个 Tokenizer。
输出没有直接可用的token 信息。 CreateRunStreaming 方式获取的信息流里并不直接包括每一次的token信息。

三、解决方案

由于上述难点，直接对 streamingUpdate 做实时的 token 统计是不推荐的。更可行的办法是采用其他思路：

1. 利用 Run 对象获取最终 Token 计数 (推荐)

最简单可靠的方法是在 Run 完成后，通过 Run 对象获取总的 Token 使用量。OpenAI 会在 Run 结束后提供 usage 字段，其中包含了 prompt tokens 和 completion tokens 的计数。

原理：

OpenAI 服务端会在 Run 结束后，统计整个过程的 Token 消耗。我们只需要在确认Run 结束后，读取 usage 字段就行了。

代码示例 (C#):

using Azure.AI.OpenAI.Assistants;

// 假设你已经有了 assistantClient、threadId 和 runId
// 等待 Run 完成
while (true)
{
	Run run = await assistantClient.GetRunAsync(threadId, runId);
    if (run.Status == RunStatus.Completed)
    {
	   // 获取 token 使用量统计结果
        Console.WriteLine($"Prompt tokens: {run.Usage.PromptTokens}");
        Console.WriteLine($"Completion tokens: {run.Usage.CompletionTokens}");
        Console.WriteLine($"Total tokens: {run.Usage.TotalTokens}");
        break;
    }
    else if (run.Status == RunStatus.Failed || run.Status == RunStatus.Cancelled || run.Status == RunStatus.Expired)
    {
       Console.WriteLine($"运行状态：{run.Status}");
        Console.WriteLine($"错误信息:{run.LastError}");

	   //处理运行异常情况
        break;
    }
    // 稍作等待，避免过于频繁的轮询
    await Task.Delay(TimeSpan.FromSeconds(1));
}

说明：

run.Usage.PromptTokens：表示提示（输入）部分消耗的 Token 数。
run.Usage.CompletionTokens：表示补全（输出）部分消耗的 Token 数。
run.Usage.TotalTokens：总的 Token 消耗。

安全建议：

对 run.Status 进行判断，处理 Failed、Cancelled、Expired 等异常情况。
设置合适的等待间隔 (Task.Delay)，避免轮询过于频繁，消耗过多请求次数。
检查错误信息：在生产环境，要认真处理所有可能的异常。

2. 结合非流式方法预估 (适用于预估消耗)

如果需要在流式传输开始之前就预估 Token 消耗（例如，用于限制输入长度），可以使用非流式方法获取现有内容的token，再加上新内容的Token统计，作为整体的 token 计数，并判断是否满足最大允许值。

原理：

虽然流式传输过程中很难精确计算，但我们可以“曲线救国”，先用非流式方法算出已有会话的 Token。增加内容的时候, 也用同样方法预估增量内容的token 数。

步骤：

获取历史消息的 Token 计数: 将整个对话历史组织成非流式的输入。调用如: tiktoken-go 的方法获取当前会话内容的token 计数。
计算新消息的Token： 使用 tiktoken-go 一样的工具。获取新增输入内容的Token数量。
预估流的总体token 数。

代码示例 (结合 C# 和 go 服务, tiktoken-go示例)：

假设你有一个 go 服务，可以根据字符串来计算 token数. 可以构建以下HTTP 调用。 (你需要有可用的Go环境来构建和提供服务，这里仅简略示意，具体go 实现请搜索tiktoken-go)

//go
func CountToken(w http.ResponseWriter, r *http.Request){

   reqBody,err := io.ReadAll(r.Body)
	//处理错误
   //使用tiktoken 来获得编码器
   enc, err := tiktoken.GetEncoding("cl100k_base") // 或者 p50k_base 等

   //处理错误
    //将文本转换为token
	token := enc.Encode(string(reqBody), nil, nil)

	strResult := strconv.Itoa(len(token))

   io.WriteString(w,strResult )
}

//C#

// 假定你的Go 服务运行在 localhost:8081
using System.Net.Http;

public static  async Task<int> EstimateMessageTokenByGoServer(string msg)
{
  HttpClient client = new HttpClient();
  HttpRequestMessage reqMsg = new HttpRequestMessage(HttpMethod.Post, "http://localhost:8081/api/count_token");
    reqMsg.Content = new StringContent(msg, Encoding.UTF8);
    HttpResponseMessage response = await client.SendAsync(reqMsg);
   response.EnsureSuccessStatusCode();
   var rspStr  =  await response.Content.ReadAsStringAsync();
  if(int.TryParse(rspStr, out int n)) {

     return n;

  }else
    {
   //log the error.
     return 0;
    }

}

进阶技巧 (更精确地估算):

为了提高估算的精度，可以使用针对 Assistant API 优化的 Tokenizer。不同模型使用不同的 Tokenizer. 必须匹配一致。比如 "cl100k_base" 用于 gpt-4、gpt-3.5-turbo、text-embedding-ada-002 等. "p50k_base" 用于 text-davinci-002、text-davinci-003 等。
注意这只是一个估算。特别注意系统消息，Assistant 可能对最终用户的输出的消息会调整系统提示，并多次执行流传输, 所以会有额外的计算成本。尤其是在流式处理中可能会出现估算偏差。最终统计还是建议通过第一种方式实现。

注意事项：

请先安装 tiktoken-go，并搭建对应 go 服务, 以暴露给 C# 服务调用.
务必正确匹配所选模型对应的 encoding.

3.本地模拟估算(适用少量文本)

如果消息内容较为简短(小于几百字)，而且希望在不联网的情况下实现粗略估算，可以使用类似 SharpToken 这类 C# 的开源库，但这种方式不能实现很精确的估算. 误差会比第2种方法更大一些。

原理类似方案2, 区别在于可以在 C# 本地简单完成token统计工作。并不需要远程调用.

安装:

Install-Package SharpToken -Version 1.0.35

代码示例：

using SharpToken;

var encoding = GptEncoding.GetEncodingForModel("gpt-3.5-turbo"); // 假设使用 gpt-3.5-turbo 模型
var prompt = "你好，世界！";
var tokenCount = encoding.Encode(prompt).Count;
Console.WriteLine(tokenCount);

进阶技巧：
和方案二的进阶相同。需要注意 Encoding 必须选择与你的 Assistant 模型对应。
重要说明：
无论是使用何种客户端 token 计数库。由于可能存在版本不同步等问题, 请务必不要在生产环境依赖这样的方式进行任何形式的token 消耗审计和成本结算。这会导致明显的误差.
这种方式适合非实时的粗略token数量统计, 尤其在测试的时候适用。

四、总结

给使用 OpenAI Assistant API 流式传输的朋友们提个醒，记得一定要在Run 完成后，检查一下 Run 对象的 Usage 信息，拿到实际消耗的Token数量。要是想提前预估Token数量，试试第二和第三种方法。但这些方法只能估个大概，最后还得看 Run.Usage 的结果。再次强调, 使用任何第三方的 token 计数工具都有偏差. 不要做为成本依据。