ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Haystack Token Counters 完全指南:在调用模型前精确估算对话与工具 Schema 的 Token 占用

Haystack Token Counters 完全指南:在调用模型前精确估算对话与工具 Schema 的 Token 占用 Haystack Token Counters 完全指南在调用模型前精确估算对话与工具 Schema 的 Token 占用【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack本文是一份围绕 Haystacktoken_counters模块的深度技术指南。该模块提供了TokenCounter协议与三种内置实现ApproximateTokenCounter、TiktokenCounter、OpenAITokenCounter用于在把消息发送给大模型之前估算一段对话和工具 Schema 会占用多少 token从而支撑上下文窗口预算校验、Agent 上下文压缩、token 计费等场景。读完本文你将掌握每个计数器的定位、全部构造参数与调用方法、内部渲染与计数的实现原理以及如何自定义属于自己的计数器。为什么需要 Token 计数器在生产级 LLM 应用中很多功能需要在发送请求前知道对话的体积判断当前对话是否超出模型的上下文窗口超限时触发上下文压缩Haystack 的 agent 上下文压缩机制即可借助计数器决定要丢弃多少历史消息在Agent 多轮工具调用场景中估算不断累积的系统提示、工具结果和工具 Schema 对上下文的消耗为计费或配额管理预留估算。ChatMessage列表和可选工具 Schema 正是模型输入的主体token_counters 模块 围绕list[ChatMessage]与tools两个输入统一暴露计数能力。所有计数器都会把消息中的角色、文本、工具调用、工具结果以及可选的工具 Schema 纳入统计并覆盖消息中的图片、文件等非文本内容。TokenCounter 协议所有计数器的统一接口所有计数器都实现TokenCounter协议定义见 haystack/token_counters/types/protocol.py它要求实现三个成员count(messages, toolsNone) - int返回给定消息及可选工具 Schema估算占用的 token 数。传入tools会将其 Schema 一并计入置为None则只测消息本身to_dict() - dict[str, Any]把计数器序列化为字典保证其设置能够在序列化后保留下来from_dict(cls, data) - TokenCounter类方法反序列化。默认实现直接调用default_from_dict把参数字典原样传回构造函数只有当to_dict()输出需要先重建再传入构造函数的值例如Secret或嵌套组件时才需要重写它。协议默认的from_dict对普通标量参数chars_per_token、encoding、tokens_per_image等已经足够三个内置计数器因此都没有重写from_dict。三种内置计数器对比计数器计数方式额外依赖适合场景ApproximateTokenCounter将渲染后的文本长度除以可配置的每 token 字符数无快速、零依赖的估算TiktokenCounter使用 OpenAI 的tiktoken字节对编码器在本地计数tiktoken针对 OpenAI 模型更准确的估算OpenAITokenCounter调用 OpenAI 输入 token 计数 APIPOST /v1/responses/input_tokensOpenAI API Key精确的、模型特定的计数涵盖图片、文件与工具此外在anthropic-haystack、google-genai-haystack等官方集成包中还提供了AnthropicTokenCounter、GoogleGenAITokenCounter等 provider 专属实现见 token counters 指南本仓库内置的三种实现则无需安装任何集成包即可使用。ApproximateTokenCounter零依赖的字符比估算ApproximateTokenCounter源码见 haystack/token_counters/approximate_counter.py是所有计数器中唯一无需安装任何额外依赖、也无需 warm_up 加载模型的实现适合在初始化阶段或资源受限环境中做快速预算。构造参数__init__( chars_per_token: float 4.0, tokens_per_image: int 85, tokens_per_file: int 1000, ) - Nonechars_per_tokenfloat默认4.0多少个字符按一个 token 计。这个值决定了估算的松紧度——对于英文内容 4 字符/token 是常用经验值中文等字符密度较高的内容可适当调低tokens_per_imageint默认85每张图片收取的 token 数。默认值对应 OpenAI 对小尺寸图片的计费若发送大图应调高tokens_per_fileint默认1000每个文件收取的 token 数是对短文档的粗略替代值实际成本取决于页数发送长文档时应调高异常当chars_per_token 0时抛出ValueError源码第 44-45 行。count一次简单的除法count的实现非常直观源码第 50-64 行text _rendered_conversation(messages) _rendered_tools(tools) text_tokens int(len(text) / self.chars_per_token) return text_tokens _non_text_tokens(...)当messages与tools均为空时直接返回0。文本 token 数 渲染后文本总长度字符数除以chars_per_token后取整最后再加上图片与文件的固定估算值。使用示例from haystack.dataclasses import ChatMessage from haystack.token_counters import ApproximateTokenCounter counter ApproximateTokenCounter(chars_per_token4.0) messages [ ChatMessage.from_user(Hello, how are you?), ChatMessage.from_assistant(Im good, thank you! How can I assist you today?) ] token_count counter.count(messages) print(fEstimated token count: {token_count})TiktokenCounter本地字节对编码估算TiktokenCounter源码见 haystack/token_counters/tiktoken_counter.py使用 OpenAI 开源的tiktoken字节对编码器在本地完成计数比字符比估算更贴近 OpenAI 模型的实际分词结果。构造参数__init__( encoding: str o200k_base, tokens_per_image: int 85, tokens_per_file: int 1000, ) - Noneencodingstr默认o200k_base使用的tiktoken编码。o200k_base是当前 OpenAI 模型使用的编码旧模型可改用cl100k_base等tokens_per_image/tokens_per_file与ApproximateTokenCounter含义一致用于给分词器无法度量的图片、文件记固定开销异常未安装tiktoken时在构造阶段抛出ImportError提示运行pip install tiktoken。测试 test_tiktoken_counter.py 明确验证了依赖缺失在构造时报告而非在首次 count 时才失败这避免了错误在运行中途才暴露。warm_up按需加载编码器def warm_up(self) - None: if self._encoder is not None: return self._encoder tiktoken.get_encoding(self.encoding)warm_up会加载编码器首次使用时若本地无缓存会自动下载词表重复调用是幂等的编码器已加载则直接返回。count()内部会自动调用warm_up()所以日常使用无需手动预热。两个必须知道的限制仅文本图片和文件只能按tokens_per_image/tokens_per_file的固定值估算无法真实分词它是 OpenAI 的编码器其他提供商的模型分词方式不同计数会有偏差跨 provider 使用时只能作为参考。使用示例from haystack.dataclasses import ChatMessage from haystack.token_counters import TiktokenCounter counter TiktokenCounter(encodingo200k_base) messages [ ChatMessage.from_user(Hello, how are you?), ChatMessage.from_assistant(Im good, thank you! How can I assist you today?) ] token_count counter.count(messages) print(fToken count: {token_count})OpenAITokenCounter调用官方 API 获取精确计数OpenAITokenCounter源码见 haystack/token_counters/openai_counter.py与前两者完全不同——它把输入发送到 OpenAI 的POST /v1/responses/input_tokens计数端点返回的结果包含模型特定的消息与工具 Schema 格式化开销以及图片、文件等受支持的非文本内容是三种实现中唯一能给出精确计数的方案。构造参数__init__( model: str, *, api_key: Secret Secret.from_env_var(OPENAI_API_KEY), api_base_url: str | None None, organization: str | None None, timeout: float | None None, max_retries: int | None None, http_client_kwargs: dict[str, Any] | None None, ) - Nonemodelstr必填按哪个模型的分词规则计数例如gpt-5-miniapi_keySecret默认读取OPENAI_API_KEY环境变量也可显式传入Secret.from_token(...)api_base_urlstr | NoneOpenAI API 的可选基础 URL用于代理或兼容端点organizationstr | NoneOpenAI 组织 IDtimeoutfloat | None客户端调用超时。未设置时使用OPENAI_TIMEOUT环境变量默认 30 秒源码第 77 行max_retriesint | None最大重试次数。未设置时使用OPENAI_MAX_RETRIES环境变量默认 5源码第 78-80 行http_client_kwargsdict[str, Any] | None用于配置底层 HTTPX 客户端的关键字参数经由 init_http_client 构造。生命周期warm_up / count / closewarm_up()初始化 OpenAI 客户端重复调用幂等。客户端通过init_http_client构建并传入OpenAI(...)count(messages, toolsNone)把每条ChatMessage通过_convert_chat_message_to_responses_api_format来自 haystack/components/generators/chat/openai_responses.py转换为 Responses API 输入格式若传入了tools则通过flatten_tools_or_toolsets展平后包装为{type: function, ...}并入请求最后调用client.responses.input_tokens.count(**request)并返回response.input_tokens。空输入同样返回0close()关闭 OpenAI 客户端及其底层 HTTP 资源并把self.client置空。这在长生命周期服务中释放连接时很有用。使用示例from haystack.dataclasses import ChatMessage from haystack.token_counters import OpenAITokenCounter counter OpenAITokenCounter(gpt-5-mini) messages [ChatMessage.from_user(Hello, how are you?)] token_count counter.count(messages) print(fToken count: {token_count})注意该计数器需要网络请求与 API Key计数会消耗配额与时间适合对精确度要求高例如超限判定、计费的场景。统一渲染机制计数器内部是如何看见对话的三种计数器对文本的测量都建立在同一个渲染层上见 haystack/token_counters/utils.py理解它就能理解计数结果到底是什么_render_message把单条ChatMessage渲染为若干行纯文本规则包括普通消息带角色前缀如[user] Hello、[system] rules工具调用渲染为[assistant - tool_call] search({q: x})参数 JSON 按键排序保证渲染结果稳定工具结果渲染为[tool:search] found it出错时带(error)标记图片与文件用占位符代替image、file: report.pdf因为它们没有文本形态但同样消耗 token推理内容ReasoningContent被刻意排除——provider 在轮次间会丢弃推理内容它不属于被测上下文源码第 38-39 行注释。_rendered_conversation把整段对话拼接为一个纯文本块这正是计数器测量的对象_rendered_tools把工具 Schema 序列化为一个 JSON 块模拟 provider 随消息一起发送的格式_non_text_tokens统计图片与文件数量乘以固定费率。注意它会遍历工具结果——嵌套在工具结果中的ImageContent/FileContent例如工具返回的截图同样会被计入因为工具结果本身就在上下文里。test/token_counters/test_utils.py 中完整覆盖了各类消息系统、用户、带工具调用的助手消息、工具结果、错误工具结果、带图片和文件的用户消息、空消息的渲染输出可作为理解该机制的精确参考。工具 Schema 的计数工具 Schema 会与消息一起发送给模型并消耗上下文 token因此所有计数器都支持传入toolsfrom typing import Annotated from haystack.dataclasses import ChatMessage from haystack.token_counters import ApproximateTokenCounter from haystack.tools import tool tool def search(query: Annotated[str, The search query]) - str: Search for documents that match the query. return Search results messages [ChatMessage.from_user(Find information about Haystack.)] counter ApproximateTokenCounter() token_count counter.count(messages, tools[search])三种计数器的行为一致传入tools后其 Schema 计入估算测试test_tool_schemas_add_to_the_count验证了counter.count(messages, tools[search]) counter.count(messages)可以只数工具不数消息counter.count([], tools[search])同样返回正数messages与tools都为空时才返回0。序列化与反序列化让计数器配置可持久化三个内置计数器都通过default_to_dict实现to_dict序列化结果遵循 Haystack 统一格式——包含type类的完整限定名与init_parameters。例如 test_approximate_counter.py 验证的 round-tripdata ApproximateTokenCounter(chars_per_token3.5, tokens_per_image200, tokens_per_file3000).to_dict() # data { # type: haystack.token_counters.approximate_counter.ApproximateTokenCounter, # init_parameters: {chars_per_token: 3.5, tokens_per_image: 200, tokens_per_file: 3000}, # } restored ApproximateTokenCounter.from_dict(data)TiktokenCounter的to_dict序列化encoding、tokens_per_image、tokens_per_fileOpenAITokenCounter额外序列化api_keySecret对象、model、api_base_url、organization、timeout、max_retries、http_client_kwargs。由于OpenAITokenCounter的to_dict输出了Secret理论上需要重写from_dict来重建Secret——不过其默认from_dict路径依赖default_from_dict对Secret的兼容处理实际使用时以测试与序列化工具的实际行为为准。自定义 TokenCounter接入自己的计数逻辑当内置实现不满足需求时例如对接某个提供商的专有计数端点实现TokenCounter协议即可参考 token counters 指南from typing import Any from haystack.core.serialization import default_to_dict from haystack.dataclasses import ChatMessage from haystack.token_counters import TokenCounter from haystack.tools import ToolsType class ProviderTokenCounter(TokenCounter): def count( self, messages: list[ChatMessage], tools: ToolsType | None None, ) - int: # Call the providers token-counting endpoint here. ... def to_dict(self) - dict[str, Any]: return default_to_dict(self)必须实现count()与to_dict()两个方法默认from_dict()会直接还原普通构造参数当to_dict()序列化了需要先重建的值如Secret或嵌套组件时再重写from_dict()。选型建议与适用前提零依赖快速估算CI 检查、粗略预算选ApproximateTokenCounter无需安装任何包、无需网络OpenAI 模型的本地高精度估算选TiktokenCounter需pip install tiktoken首次warm_up会下载词表且只对 OpenAI 编码准确精确计数含图片、文件与工具格式开销选OpenAITokenCounter需要OPENAI_API_KEY与网络请求注意其存在超时与重试配置默认 30 秒 / 5 次重试若使用 Claude、Gemini 等模型可查阅官方集成包中的AnthropicTokenCounter、GoogleGenAITokenCounter实现。延伸阅读Token Counters 指南文档各计数器细分指南docs-website/docs/token-counters/目录下的approximatetokencounter.mdx、tiktokencounter.mdx、openaitokencounter.mdx等Token Counters API 参考本 API 参考原始页面源码haystack/token_counters/计数器实现与渲染工具、haystack/token_counters/types/protocol.py协议定义测试test/token_counters/test_approximate_counter.py、test_tiktoken_counter.py、test_openai_counter.py、test_utils.py覆盖计数、异常、序列化 round-trip 与工具计数行为。【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表