docs: Update hooks lesson with improved context tracker example
- Replace simple Stop-only context-usage hook with hook pair pattern - Add UserPromptSubmit + Stop hook combination for tracking delta - Include both char-estimation and tiktoken versions as separate files - Show how to use session_id for isolated state tracking
This commit is contained in:
149
06-hooks/context-tracker-tiktoken.py
Normal file
149
06-hooks/context-tracker-tiktoken.py
Normal file
@@ -0,0 +1,149 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Context Usage Tracker (tiktoken version) - Tracks token consumption per request.
|
||||
|
||||
Uses UserPromptSubmit as "pre-message" hook and Stop as "post-response" hook
|
||||
to calculate the delta in token usage for each request.
|
||||
|
||||
This version uses tiktoken with p50k_base encoding for ~90-95% accuracy.
|
||||
Requires: pip install tiktoken
|
||||
|
||||
For a zero-dependency version, see context-tracker.py.
|
||||
|
||||
Usage:
|
||||
Configure both hooks to use the same script:
|
||||
- UserPromptSubmit: saves current token count
|
||||
- Stop: calculates delta and reports usage
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
|
||||
try:
|
||||
import tiktoken
|
||||
|
||||
TIKTOKEN_AVAILABLE = True
|
||||
except ImportError:
|
||||
TIKTOKEN_AVAILABLE = False
|
||||
print(
|
||||
"Warning: tiktoken not installed. Install with: pip install tiktoken",
|
||||
file=sys.stderr,
|
||||
)
|
||||
|
||||
# Configuration
|
||||
CONTEXT_LIMIT = 128000 # Claude's context window (adjust for your model)
|
||||
|
||||
|
||||
def get_state_file(session_id: str) -> str:
|
||||
"""Get temp file path for storing pre-message token count, isolated by session."""
|
||||
return os.path.join(tempfile.gettempdir(), f"claude-context-{session_id}.json")
|
||||
|
||||
|
||||
def count_tokens(text: str) -> int:
|
||||
"""
|
||||
Count tokens using tiktoken with p50k_base encoding.
|
||||
|
||||
This provides ~90-95% accuracy compared to Claude's actual tokenizer.
|
||||
Falls back to character estimation if tiktoken is not available.
|
||||
|
||||
Note: Anthropic hasn't released an official offline tokenizer.
|
||||
tiktoken with p50k_base is a reasonable approximation since both
|
||||
Claude and GPT models use BPE (byte-pair encoding).
|
||||
"""
|
||||
if TIKTOKEN_AVAILABLE:
|
||||
enc = tiktoken.get_encoding("p50k_base")
|
||||
return len(enc.encode(text))
|
||||
else:
|
||||
# Fallback to character estimation (~4 chars per token)
|
||||
return len(text) // 4
|
||||
|
||||
|
||||
def read_transcript(transcript_path: str) -> str:
|
||||
"""Read and concatenate all content from transcript file."""
|
||||
if not transcript_path or not os.path.exists(transcript_path):
|
||||
return ""
|
||||
|
||||
content = []
|
||||
with open(transcript_path, "r") as f:
|
||||
for line in f:
|
||||
try:
|
||||
entry = json.loads(line.strip())
|
||||
# Extract text content from various message formats
|
||||
if "message" in entry:
|
||||
msg = entry["message"]
|
||||
if isinstance(msg.get("content"), str):
|
||||
content.append(msg["content"])
|
||||
elif isinstance(msg.get("content"), list):
|
||||
for block in msg["content"]:
|
||||
if isinstance(block, dict) and block.get("type") == "text":
|
||||
content.append(block.get("text", ""))
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
|
||||
return "\n".join(content)
|
||||
|
||||
|
||||
def handle_user_prompt_submit(data: dict) -> None:
|
||||
"""Pre-message hook: Save current token count before request."""
|
||||
session_id = data.get("session_id", "unknown")
|
||||
transcript_path = data.get("transcript_path", "")
|
||||
|
||||
transcript_content = read_transcript(transcript_path)
|
||||
current_tokens = count_tokens(transcript_content)
|
||||
|
||||
# Save to temp file for later comparison
|
||||
state_file = get_state_file(session_id)
|
||||
with open(state_file, "w") as f:
|
||||
json.dump({"pre_tokens": current_tokens}, f)
|
||||
|
||||
|
||||
def handle_stop(data: dict) -> None:
|
||||
"""Post-response hook: Calculate and report token delta."""
|
||||
session_id = data.get("session_id", "unknown")
|
||||
transcript_path = data.get("transcript_path", "")
|
||||
|
||||
transcript_content = read_transcript(transcript_path)
|
||||
current_tokens = count_tokens(transcript_content)
|
||||
|
||||
# Load pre-message count
|
||||
state_file = get_state_file(session_id)
|
||||
pre_tokens = 0
|
||||
if os.path.exists(state_file):
|
||||
try:
|
||||
with open(state_file, "r") as f:
|
||||
state = json.load(f)
|
||||
pre_tokens = state.get("pre_tokens", 0)
|
||||
except (json.JSONDecodeError, IOError):
|
||||
pass
|
||||
|
||||
# Calculate delta
|
||||
delta_tokens = current_tokens - pre_tokens
|
||||
remaining = CONTEXT_LIMIT - current_tokens
|
||||
percentage = (current_tokens / CONTEXT_LIMIT) * 100
|
||||
|
||||
# Report usage (stderr so it doesn't interfere with hook output)
|
||||
method = "tiktoken" if TIKTOKEN_AVAILABLE else "estimated"
|
||||
print(
|
||||
f"Context ({method}): ~{current_tokens:,} tokens "
|
||||
f"({percentage:.1f}% used, ~{remaining:,} remaining)",
|
||||
file=sys.stderr,
|
||||
)
|
||||
if delta_tokens > 0:
|
||||
print(f"This request: ~{delta_tokens:,} tokens", file=sys.stderr)
|
||||
|
||||
|
||||
def main():
|
||||
data = json.load(sys.stdin)
|
||||
event = data.get("hook_event_name", "")
|
||||
|
||||
if event == "UserPromptSubmit":
|
||||
handle_user_prompt_submit(data)
|
||||
elif event == "Stop":
|
||||
handle_stop(data)
|
||||
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user