feat(context-manager): add tiktoken dependency and improve token counting accuracy

- Add tiktoken as a required dependency for accurate token counting
- Update documentation with installation instructions and token counting strategy
- Improve logging to include tool definition token estimates
- Enhance error handling to fail fast when tiktoken is not available
- Update context manager to use tiktoken's cl100k_base encoding for GPT-4 compatibility
This commit is contained in:
YueGuobin committed 2026-03-05 01:01:34 +08:00
1 parent 7368ac098a
commit f8aa653ef9
4 files changed
+392 -43

No files matched your search

@@ -16,6 +16,7 @@ including:
- System message preservation
"""
import json
import logging
from typing import Any, Literal
@@ -30,6 +31,151 @@ from langchain_core.messages.utils import trim_messages
logger = logging.getLogger(__name__)
# ============================================================================
# Token Counting - Using tiktoken for accuracy
# ============================================================================
# Global tiktoken encoding cache (lazy loading)
_tiktoken_encoding = None
def _get_tiktoken_encoding():
"""
Get tiktoken encoding instance (cached).
Uses cl100k_base encoding (GPT-4) which is a good approximation
for most modern LLMs including OpenAI, Anthropic, and DeepSeek.
Returns:
Encoding object
Raises:
ImportError: If tiktoken is not installed
"""
global _tiktoken_encoding
if _tiktoken_encoding is None:
try:
import tiktoken
_tiktoken_encoding = tiktoken.get_encoding("cl100k_base")
logger.debug("Using tiktoken (cl100k_base) for accurate token counting")
except ImportError:
raise ImportError(
"tiktoken is required for accurate token counting. "
"Please install it with: pip install tiktoken>=0.8.0"
)
return _tiktoken_encoding
def count_tokens_accurately(text: str) -> int:
"""
Count tokens in text accurately using tiktoken.
Args:
text: The text to count tokens for
Returns:
Number of tokens
Raises:
ImportError: If tiktoken is not installed
"""
if not text:
return 0
encoding = _get_tiktoken_encoding()
try:
return len(encoding.encode(text))
except Exception as e:
logger.error("tiktoken encoding failed: %s", e)
raise
def count_messages_tokens(messages: list[Any]) -> int:
"""
Count total tokens in a list of messages accurately.
Args:
messages: List of LangChain messages
Returns:
Total token count
Raises:
ImportError: If tiktoken is not installed
"""
total = 0
for msg in messages:
if hasattr(msg, 'content') and msg.content:
# Handle both string and complex content
content = str(msg.content)
total += count_tokens_accurately(content)
return total
# ============================================================================
# Tool Definition Token Estimation
# ============================================================================
def estimate_tool_tokens(tools: list[Any]) -> int:
"""
Estimate the token count of tool definitions.
When tools are bound to an LLM, their schemas (name, description, parameters)
are converted to JSON and sent to the LLM. This function estimates how many
tokens those definitions will consume.
Args:
tools: List of LangChain BaseTool instances
Returns:
Estimated token count for all tool definitions
Raises:
ImportError: If tiktoken is not installed
"""
if not tools:
return 0
total_tokens = 0
encoding = _get_tiktoken_encoding()
for tool in tools:
try:
# Build tool schema as it would be sent to LLM
tool_schema = {
"type": "function",
"function": {
"name": tool.name,
"description": tool.description if hasattr(tool, 'description') else "",
}
}
# Add parameters schema if available
if hasattr(tool, 'args_schema') and tool.args_schema:
try:
tool_schema["function"]["parameters"] = tool.args_schema.schema()
except Exception:
# If schema generation fails, use empty object
tool_schema["function"]["parameters"] = {"type": "object"}
# Count tokens in the schema
schema_str = json.dumps(tool_schema, ensure_ascii=False)
tool_tokens = len(encoding.encode(schema_str))
total_tokens += tool_tokens
logger.debug(
"Tool '%s': ~%d tokens (schema size: %d chars)",
tool.name, tool_tokens, len(schema_str)
)
except Exception as e:
logger.error("Failed to estimate tokens for tool '%s': %s", tool.name, e)
raise
logger.info("Tool definitions estimated at ~%d total tokens (%d tools)", total_tokens, len(tools))
return total_tokens
# NOTE: Built-in model context limits have been removed.
# Model providers frequently update context limits, and maintaining this list is not sustainable.
# Users MUST configure context_limit explicitly in their LLM model configurations.
@@ -139,8 +285,9 @@ def trim_messages_for_context(
"""
Trim messages to fit within model's context window.
This function uses LangChain's trim_messages utility to intelligently
reduce message history while preserving conversation flow.
This function uses tiktoken for accurate token counting and LangChain's
trim_messages utility to intelligently reduce message history while
preserving conversation flow.
Args:
messages: List of LangChain messages (HumanMessage, AIMessage, etc.)
@@ -168,34 +315,29 @@ def trim_messages_for_context(
max_tokens = calculate_max_tokens(model_limit, strategy)
# Check if trimming is needed
try:
from langchain_core.messages.utils import count_tokens_approximately
current_tokens = count_messages_tokens(messages)
# Estimate current token count
current_tokens = count_tokens_approximately(messages)
if current_tokens <= max_tokens:
logger.debug(
"Messages fit in context: %d / %d tokens",
current_tokens, max_tokens
)
return messages
logger.info(
"Trimming messages: %d → %d tokens (model: %s)",
current_tokens, max_tokens, model_name
if current_tokens <= max_tokens:
logger.debug(
"Messages fit in context: %d / %d tokens",
current_tokens, max_tokens
)
return messages
except Exception as e:
logger.warning("Failed to count tokens: %s, proceeding with trim", e)
logger.info(
"Trimming messages: %d → %d tokens (model: %s)",
current_tokens, max_tokens, model_name
)
# Trim messages using LangChain's utility
try:
from langchain_core.messages.utils import count_tokens_approximately
trimmed = trim_messages(
messages,
strategy="last", # Keep most recent messages
max_tokens=max_tokens,
preserve_system=preserve_system,
token_counter=count_tokens_approximately, # Required: token counting function
include_system=preserve_system, # Keep system messages if requested
start_on="human", # Ensure we start with a human message
end_on=("human", "tool", "ai"), # End on human/tool/ai messages
)
@@ -229,7 +371,8 @@ def trim_messages_for_context(
def get_token_usage_summary(
messages: list[Any],
model_name: str,
llm_config: dict[str, Any] | None = None
llm_config: dict[str, Any] | None = None,
tool_tokens: int = 0,
) -> dict[str, Any]:
"""
Get a summary of token usage for the given messages.
@@ -238,20 +381,22 @@ def get_token_usage_summary(
messages: List of LangChain messages
model_name: Name of the LLM model
llm_config: Optional LLM config dict from database (may contain context_limit in K tokens)
tool_tokens: Optional token count for tool definitions
Returns:
dict: Token usage summary including:
- estimated_tokens: Estimated total tokens
- estimated_tokens: Estimated total tokens (messages only)
- tool_tokens: Token count for tool definitions
- total_tokens: Sum of messages and tools
- model_limit_k: Model's context window limit in K tokens
- model_limit_tokens: Model's context window limit in actual tokens
- usage_percentage: Percentage of context used
- usage_percentage: Percentage of context used (excluding tools)
- total_usage_percentage: Percentage including tools
- message_count: Number of messages
- needs_trimming: Whether messages exceed 80% of limit
"""
try:
from langchain_core.messages.utils import count_tokens_approximately
estimated_tokens = count_tokens_approximately(messages)
estimated_tokens = count_messages_tokens(messages)
except Exception as e:
logger.warning("Failed to count tokens: %s", e)
estimated_tokens = 0
@@ -259,12 +404,17 @@ def get_token_usage_summary(
model_limit_tokens = get_model_context_limit(model_name, llm_config)
model_limit_k = model_limit_tokens // 1000
usage_percentage = (estimated_tokens / model_limit_tokens * 100) if model_limit_tokens > 0 else 0
total_tokens = estimated_tokens + tool_tokens
total_usage_percentage = (total_tokens / model_limit_tokens * 100) if model_limit_tokens > 0 else 0
return {
"estimated_tokens": estimated_tokens,
"tool_tokens": tool_tokens,
"total_tokens": total_tokens,
"model_limit_k": model_limit_k,
"model_limit_tokens": model_limit_tokens,
"usage_percentage": round(usage_percentage, 2),
"total_usage_percentage": round(total_usage_percentage, 2),
"message_count": len(messages),
"needs_trimming": usage_percentage > 80,
}
@@ -277,6 +427,7 @@ def prepare_context_messages(
topology_context: str | None,
model_name: str,
llm_config: dict[str, Any] | None = None,
tools: list[Any] | None = None,
) -> list[Any]:
"""
Prepare full context messages for LLM call with automatic trimming.
@@ -290,6 +441,7 @@ def prepare_context_messages(
topology_context: Optional topology information string
model_name: Name of the LLM model
llm_config: Optional LLM config dict from database (may contain context_limit and context_strategy)
tools: Optional list of LangChain tools (for token estimation)
Returns:
list: Prepared messages ready for LLM invocation
@@ -314,6 +466,9 @@ def prepare_context_messages(
else:
logger.warning("Invalid context_strategy '%s', using 'balanced'", strategy)
# Estimate tool tokens (tools are sent with each LLM call)
tool_tokens = estimate_tool_tokens(tools) if tools else 0
# Build base context (system + topology)
context_messages = [SystemMessage(content=system_prompt)]
@@ -334,14 +489,16 @@ def prepare_context_messages(
preserve_system=True, # Always keep system prompts
)
# Log summary
summary = get_token_usage_summary(trimmed_messages, model_name, llm_config)
# Log summary (including tool tokens)
summary = get_token_usage_summary(trimmed_messages, model_name, llm_config, tool_tokens)
logger.info(
"Context prepared: %d msgs, ~%d tokens / %dK limit (%.1f%%), strategy=%s",
"Context prepared: %d msgs, ~%d tokens (messages) + %d tokens (tools) = %d total / %dK limit (%.1f%%), strategy=%s",
summary["message_count"],
summary["estimated_tokens"],
summary["tool_tokens"],
summary["total_tokens"],
summary["model_limit_k"],
summary["usage_percentage"],
summary["total_usage_percentage"],
trim_strategy
)
@@ -204,6 +204,7 @@ def llm_call(state: dict, config: RunnableConfig | None = None):
topology_context=topology_context,
model_name=llm_config.get("model", "default"),
llm_config=llm_config,
tools=tools, # Pass tools for accurate token estimation
)
# Create fresh model with tools for each LLM call