如何在OpenAI Assistant流式传输消息时统计消耗的Token数量

一、获取最终准确的Token消耗（官方计算值） #

当Run流式更新完成（状态变为Completed）时，OpenAI会返回包含完整token使用数据的Run对象，直接提取其中的Usage字段即可，这是最准确的统计结果。

.NET 代码示例 #

using OpenAI.Assistants;

await foreach (var update in assistantClient.CreateRunStreaming(threadId, assistantId))
{
    // 处理其他流式更新并通过SignalR发送给客户端...
    
    if (update is RunUpdate runUpdate && runUpdate.Run.Status == RunStatus.Completed)
    {
        var tokenUsage = runUpdate.Run.Usage;
        int promptTokens = tokenUsage.PromptTokens.Value;
        int completionTokens = tokenUsage.CompletionTokens.Value;
        int totalTokens = tokenUsage.TotalTokens.Value;
        
        // 将最终统计结果通过SignalR发送给客户端
        await hubContext.Clients.User(userId).SendAsync("FinalTokenStats", new 
        {
            PromptTokens = promptTokens,
            CompletionTokens = completionTokens,
            TotalTokens = totalTokens
        });
    }
}

Python 代码示例 #

from openai import OpenAI

client = OpenAI()

stream = client.beta.threads.runs.create_stream(
    thread_id=thread_id,
    assistant_id=assistant_id
)

for event in stream:
    # 处理其他流式事件并发送给客户端...
    
    if event.event == "thread.run.completed":
        usage = event.data.usage
        # 提取官方统计的token数
        prompt_tokens = usage.prompt_tokens
        completion_tokens = usage.completion_tokens
        total_tokens = usage.total_tokens
        
        # 通过SignalR发送最终统计
        send_to_client({
            "type": "final_token_stats",
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "total_tokens": total_tokens
        })

二、实时近似统计Completion Token #

如果需要在流式响应过程中实时统计生成的Completion Token，可以通过提取每个增量文本，用OpenAI的tiktoken工具计算token数并累加。注意这种方法是近似值，和官方最终统计可能有微小差异。

.NET 实现步骤 #

1. 安装tiktoken.NET库：Install-Package Tiktoken.NET
2. 代码示例：

using Tiktoken;
using OpenAI.Assistants;

// 根据使用的模型选择对应编码，比如gpt-4o
var encoding = Encoding.GetEncodingForModel("gpt-4o");
int totalCompletionTokens = 0;

await foreach (var update in assistantClient.CreateRunStreaming(threadId, assistantId))
{
    if (update is MessageDeltaUpdate messageDeltaUpdate)
    {
        foreach (var delta in messageDeltaUpdate.Message.Delta.Content)
        {
            if (delta is TextContentPartDelta textDelta)
            {
                // 计算当前增量文本的token数
                int newTokens = encoding.Encode(textDelta.Text.Value).Count;
                totalCompletionTokens += newTokens;
                
                // 通过SignalR实时发送累计的Completion Token数
                await hubContext.Clients.User(userId).SendAsync("UpdateCompletionTokens", totalCompletionTokens);
            }
        }
    }
    
    // 处理其他更新并发送给客户端...
}

Python 实现步骤 #

1. 安装tiktoken库：pip install tiktoken
2. 代码示例：

import tiktoken
from openai import OpenAI

client = OpenAI()
# 根据模型选择编码
encoding = tiktoken.encoding_for_model("gpt-4o")
total_completion_tokens = 0

stream = client.beta.threads.runs.create_stream(
    thread_id=thread_id,
    assistant_id=assistant_id
)

for event in stream:
    if event.event == "thread.message.delta":
        delta_text = event.data.delta.content[0].text.value
        # 计算增量文本的token数并累加
        new_tokens = len(encoding.encode(delta_text))
        total_completion_tokens += new_tokens
        
        # 实时发送累计数给客户端
        send_to_client({
            "type": "update_completion_tokens",
            "count": total_completion_tokens
        })
    
    # 处理其他事件并发送给客户端...

注意事项 #

• 必须使用对应模型的编码规则，不同模型的token计算逻辑不同（比如gpt-3.5-turbo和gpt-4系列）。
• 实时统计仅能计算Completion Token，Prompt Token只能在Run完成后从官方Usage字段获取。
• 实时统计的近似值和官方结果的差异主要来自OpenAI对系统提示、工具调用格式等元数据的token计算，日常使用误差可忽略。

内容的提问来源于stack exchange，提问作者Sankaranarayanan Rajkumar