kv-cache¶
第2章 · 上下文工程 · 配套项目
chapter2/kv-cache
项目说明¶
KV Cache Demonstration with ReAct Agent¶
A comprehensive demonstration of KV (Key-Value) cache importance in LLMs using a ReAct pattern agent with local file system tools. This project shows how different implementation patterns can significantly impact performance through their effect on KV cache utilization.
🎯 Overview¶
This project implements a ReAct (Reasoning and Acting) agent that uses a current Moonshot Kimi model (default kimi-k2.6) to analyze code projects. The agent uses the standard OpenAI tool calling format and demonstrates six different implementation patterns - one correct and five incorrect - showing how seemingly minor changes can invalidate KV cache and dramatically impact performance.
Model note. On the live Moonshot endpoint the whole current Kimi family (
kimi-k2.5/kimi-k2.6/kimi-k2.7*/kimi-k3) are reasoning models: they emitreasoning_contentand only accepttemperature=1(the code handles this automatically). They do reportcached_tokens, which is exactly what this experiment measures. The legacy non-reasoningmoonshot-v1-*chat models, by contrast, do not reportcached_tokens, so they cannot demonstrate the cache effect. We default tokimi-k2.6because it reports cache hits like the rest of the family but has the lightest reasoning footprint, which keeps TTFT the least noisy. Because any reasoning model adds variable thinking latency, the cache hit rate / cache ratio columns (which are deterministic) are the robust signal here; treat TTFT as secondary.
What is KV Cache?¶
KV cache stores the key-value pairs from the attention mechanism of transformer models. When the conversation context remains stable, these cached values can be reused, significantly reducing computation time and improving response latency (especially Time to First Token - TTFT).
🚀 Features¶
- ReAct Pattern Agent: Implements reasoning and acting pattern for systematic task execution
- Standard Tool Calling: Uses OpenAI's standard tool calling format (no manual parsing)
- Local File System Tools: Safe implementations of
read_file,find, andgrepcommands - Robust Error Handling: Continues execution when tools fail, passing errors as results
- Six Implementation Modes: Demonstrates correct and incorrect KV cache patterns
- Comprehensive Metrics: Tracks TTFT, total time, cache hits/misses, and token usage
- Offline Comparison Report (
--report): Renders the cross-strategy table from saved result files without an API key - Cost Estimate: Comparison table adds an illustrative billable-token / savings column (configurable via
--cache-price-ratio) - Detailed Logging: Provides insights into cache behavior and performance impact
- Smart Completion: Automatically considers responses without tool calls as final answers
- Argument Filtering: Safely handles unexpected tool arguments without breaking
📦 Implementation Modes¶
1. ✅ Correct Implementation (correct)¶
Maintains stable context throughout the conversation: - Fixed system prompt - Consistent tool ordering - Stable message history format - No unnecessary context changes
2. ❌ Dynamic System Prompt (dynamic_system)¶
Adds a timestamp to the system prompt on each request: - Recreates entire message context each iteration - Invalidates entire cache on every call - Significantly increases TTFT - Wastes computational resources
3. ❌ Shuffled Tools (shuffled_tools)¶
Randomly reorders the tool list for each request: - Recreates context with shuffled tools each iteration - Breaks cache even though functionality is identical - Demonstrates importance of consistent ordering - Shows how minor changes affect caching
4. ❌ Dynamic User Profile (dynamic_profile)¶
Includes changing user credits in the context: - Recreates context with updated credits each iteration - Adds unnecessary dynamic information - Forces cache invalidation for irrelevant changes - Simulates common anti-patterns in production
5. ❌ Sliding Window (sliding_window)¶
Keeps only the most recent 5 messages: - Recreates context with different message window each iteration - Appears to reduce context length - Actually breaks cache continuity - Demonstrates why truncation strategies can backfire
6. ❌ Text Format (text_format)¶
Formats conversation history as plain text instead of structured messages: - Recreates context in text format each iteration - Breaks the expected message format - Prevents proper cache utilization - Shows importance of following API conventions
🔑 Critical Implementation Details¶
For KV cache to be properly invalidated in the incorrect modes, the entire message context must be recreated from scratch at the START of EACH iteration.
The implementation ensures this by:
- CORRECT mode:
- Builds messages list once on first iteration
- Keeps using the same list throughout execution
- Appends new messages (assistant responses, tool results) to this persistent list
-
Result: Stable context → KV cache works efficiently
-
Incorrect modes:
- Recreates the entire messages list from conversation history at the start of each iteration
- Within an iteration, still appends tool results to ensure proper API flow
- But next iteration starts fresh with recreated context
- Result: Context changes → KV cache invalidated
Important: Both modes append tool results within an iteration to ensure the API sees the complete conversation flow. The key difference is that incorrect modes throw away the messages list and rebuild it at the start of each new iteration, which forces cache invalidation.
🛠️ Installation¶
# Navigate to the project directory
cd chapter2/kv-cache
# Install dependencies
pip install -r requirements.txt
# Set your Kimi API key
export MOONSHOT_API_KEY="your-api-key-here"
通用回退（OpenRouter）：未设置
MOONSHOT_API_KEY/KIMI_API_KEY时，只要 配置了OPENROUTER_API_KEY，实验会自动改走 OpenRouter（kimi-*会映射为moonshotai/kimi-k2）。设置了 Moonshot key 时行为完全不变。
📖 Usage¶
Interactive Mode (Default)¶
# Run interactive mode selection menu
python main.py
# You'll see a menu like:
# ============================================================
# KV CACHE DEMONSTRATION - MODE SELECTION
# ============================================================
#
# Select a mode to run:
#
# 1. ✅ Correct Implementation - Optimal KV cache usage
# 2. ❌ Dynamic System Prompt - Adds timestamps
# 3. ❌ Shuffled Tools - Randomizes tool order
# 4. ❌ Dynamic Profile - Updates user credits
# 5. ❌ Sliding Window - Keeps only recent messages
# 6. ❌ Text Format - Plain text instead of structured
# 7. 📊 Compare All - Run all modes and compare
#
# 0. Exit
Command Line Options¶
The CLI ships Chinese --help; run python main.py --help for the full list. Key flags:
| Flag | 说明 |
|---|---|
--mode MODE |
运行单个策略（correct / dynamic_system / shuffled_tools / dynamic_profile / sliding_window / text_format） |
--compare |
依次运行全部策略并打印横向对比表（需要 API Key） |
--report |
离线：从已保存的 result_*.json / comparison_*.json 生成对比表，无需 API Key |
--input ... |
配合 --report 指定结果文件 / 通配符 / 目录（默认扫描当前目录） |
--model MODEL |
选择模型（默认 kimi-k2.6；同族 kimi-k2.5 / kimi-k3 亦可，均会上报 cached_tokens） |
--output PATH |
指定结果 JSON 的输出路径（默认按模式 + 时间戳自动命名） |
--cache-price-ratio R |
成本估算中缓存 token 相对正常 token 的计费比例（默认 0.1，即一折），仅作示意 |
--task, --root-dir |
自定义任务 / 文件工具根目录 |
# Run specific mode directly (bypasses menu)
python main.py --mode correct
# Pick a model and write to a named file
python main.py --mode sliding_window --model kimi-k2.6 --output run.json
# Run comparison across all modes (needs API key)
python main.py --compare
# Disable interactive mode
python main.py --no-interactive --mode correct
Offline Comparison Report (no API key)¶
Live runs need a Moonshot/Kimi API key. To read the already-saved result files and print the cross-strategy comparison table in one command:
# Uses the result_*.json files already in this directory
python main.py --report
# Or point at specific files / a directory, and change the assumed cache discount
python main.py --report --input result_correct_*.json result_text_format_*.json
python main.py --report --cache-price-ratio 0.5
The table compares cache hit rate, cache ratio, TTFT latency, total time, and an
illustrative billable-token / savings estimate across strategies. The report parses
both the legacy single-mode files (metrics stored as an AgentMetrics(...) string) and
the newer dict-format files, so pre-existing results remain usable.
The
Bill.Tok/Save%columns are a transparent function of the measured token counts and the--cache-price-ratioyou supply — they are an illustration of the cost impact, not a specific provider's price quote.
Custom Tasks¶
# Provide custom task via command line
python main.py --mode correct --task "Read all README files and summarize their contents"
# Use different root directory
python main.py --mode correct --root-dir ../.. --task "Analyze the project structure"
📊 Metrics Explained¶
Performance Metrics¶
- TTFT (Time to First Token): Time until the first token is generated - critical for user experience
- Tracked per iteration to show cache benefits
- First iteration: Cold start (no cache)
- Subsequent iterations: Should show improvement with cache
- TTFT Analysis:
- Per-iteration tracking shows cache effectiveness
- Average TTFT demonstrates overall performance
- Improvement percentage quantifies cache benefits
- Total Execution Time: Complete time for task execution
- Iterations: Number of ReAct cycles performed
- Tool Calls: Number of tool invocations made
Cache Statistics¶
- Cached Tokens: Number of tokens retrieved from cache
- Cache Hits: Successful cache retrievals
- Cache Misses: Failed cache retrievals requiring recomputation
- Cache Hit Rate: Percentage of successful cache usage
Token Usage¶
- Prompt Tokens: Tokens in the input prompt
- Completion Tokens: Tokens generated by the model
- Cache Ratio: Percentage of prompt tokens served from cache
📈 Expected Results¶
When comparing implementations, you should observe:
- Correct Implementation:
- Every iteration after the first reports cached tokens (the stable prefix is reused)
- Low, steady first-token latency
-
The prefix (system prompt + tools + prior turns) is served from cache
-
Incorrect Implementations:
- A collapsing cache ratio — e.g. shuffling the tool list drops the share of prompt tokens served from cache to roughly a third of the correct run
- Consistently higher TTFT (reformatting history as text or shuffling tools more than doubles first-token latency in the measured run)
- Longer total time (up to ~2.4x the correct run for
text_format)
On a reasoning model (the whole current Kimi family reasons) TTFT carries extra variance from hidden thinking tokens, so the cache ratio column is the cleanest evidence of the effect. Note also that appending dynamic data at the end of an otherwise-stable prefix (as
dynamic_system/dynamic_profiledo) only invalidates the cache from that point on — the base prefix before it still caches — so their headline cache ratio can look close tocorrectwhile their total time still regresses. The lesson: keep dynamic data out of the prefix entirely.
🔍 Example Output¶
📊 Performance Metrics:
• Time to First Token (TTFT): 0.823 seconds
• TTFT per iteration:
Iteration 1: 0.823s
Iteration 2: 0.234s (with cache)
Iteration 3: 0.198s (with cache)
Iteration 4: 0.187s (with cache)
Iteration 5: 0.192s (with cache)
• TTFT Analysis:
First iteration: 0.823s
Last iteration: 0.192s
Average (after first): 0.203s
Improvement: 76.7%
Comparison Table (--report on the saved result files in this directory)¶
These are the actual measured numbers from a single --compare run — all six
strategies under one task (kimi-k2.6, root dir = this folder, task = "find all
Python files, read main.py + agent.py, summarize in 3 sentences"), then read
back with python main.py --report:
Mode Iters 1st TTFT Avg TTFT Total(s) Prompt Cached Hit% Cache% Bill.Tok Save%
----------------------------------------------------------------------------------------------------------------
correct 3 2.328 6.054 18.163 7,567 768 100.0 10.1 6,876 9.1
dynamic_profile 3 2.206 5.986 17.962 7,652 768 100.0 10.0 6,961 9.0
dynamic_system 3 2.497 8.085 24.260 7,639 768 100.0 10.1 6,948 9.0
shuffled_tools 3 7.818 11.122 33.369 7,568 256 100.0 3.4 7,338 3.0
sliding_window 5 2.234 3.649 14.704 2,224 1,510 100.0 67.9 865 61.1
text_format 3 6.189 14.432 43.297 7,430 674 100.0 9.1 6,823 8.2
Reading the table: shuffled_tools reorders the tool definitions that sit near the
front of the prefix, so its cache ratio collapses from 10.1% to 3.4% and its
first-token latency jumps from ~2.3s to ~7.8s. text_format rebuilds history as one
plain-text blob every turn, roughly 2.4x the correct total time. sliding_window
shows a high cache ratio only because truncation shrinks the prompt itself (fewer,
mostly-cached tokens) — a reminder that a high ratio on a tiny prompt is not the same
as an efficient run.
Cache% is the share of prompt tokens served from cache; Hit% is the share of
iterations that saw any cache. Bill.Tok/Save% assume cached tokens bill at
--cache-price-ratio (default 0.1) of a normal token — an illustration, not a
provider quote. All six rows come from one --compare run, so they are directly
comparable; regenerate with python main.py --compare to reproduce.
💡 Key Insights¶
- Stable Context is Critical: Any change to the message context invalidates KV cache
- Order Matters: Even reordering identical content breaks caching
- Avoid Dynamic Metadata: Timestamps, counters, and other changing data harm performance
- Proper Formatting: Use the API's expected message format for optimal caching
- Context Continuity: Maintaining full history often performs better than truncation
🏗️ Architecture¶
kv-cache/
├── agent.py # ReAct agent implementation with different modes
├── main.py # Main script for running experiments
├── demo_quick.py # Quick demonstration script
├── test_tools.py # Test local file system tools
├── test_error_handling.py # Test error recovery capabilities
├── test_completion.py # Test final answer detection
├── requirements.txt # Project dependencies
├── README.md # This file
└── *.log # Generated log files
Components¶
- KVCacheAgent: Main agent class implementing ReAct pattern
- LocalFileTools: Safe file system operations (read, find, grep)
- KVCacheMode: Enum defining different implementation patterns
- AgentMetrics: Dataclass for tracking performance metrics
Error Handling¶
The agent implements robust error handling: - Tool Errors: When a tool fails (e.g., file not found), the error is returned as a tool result - Argument Errors: Unexpected tool arguments are filtered out automatically - Continuation: The agent continues execution even when tools fail - Error Reporting: Errors are passed to the model, which can acknowledge and work around them - Security: Access outside the root directory is denied with clear error messages
🔧 Advanced Configuration¶
Environment Variables¶
# API Configuration
export MOONSHOT_API_KEY="your-key"
# Logging Level
export LOG_LEVEL="DEBUG" # INFO, WARNING, ERROR
Custom Implementation Modes¶
You can extend the KVCacheMode enum in agent.py to add new patterns:
Then implement the behavior in the corresponding methods:
- _get_system_prompt()
- _get_tools()
- _format_messages()
📝 Best Practices for KV Cache¶
Based on this demonstration, follow these practices:
- Keep System Prompts Stable: Avoid adding timestamps or request-specific data
- Maintain Consistent Tool Order: Don't shuffle or reorder tools dynamically
- Avoid Unnecessary Metadata: Don't include counters, credits, or changing values
- Use Proper Message Format: Follow the API's expected conversation structure
- Preserve Context Continuity: Avoid aggressive truncation strategies
- Cache-Aware Design: Design your prompts and context with caching in mind
🐛 Troubleshooting¶
High TTFT with Correct Implementation¶
- Check if this is the first request (cold start)
- Verify API key and model availability
- Ensure stable network connection
Zero Cache Hits¶
- Confirm using a current Kimi model (
kimi-k2.5/kimi-k2.6/kimi-k3) — these reportcached_tokens. The non-reasoningmoonshot-v1-*models do not report cached tokens, so this metric will read 0 even when caching happens server-side. - Check if context is actually stable
- Review logs for cache invalidation patterns
Tool Execution Errors¶
- Verify root directory permissions
- Check file paths are within root directory
- Ensure files exist and are readable
📚 References¶
📄 License¶
This project is part of the AI Agent Book educational materials.
源代码¶
agent.py¶
"""
KV Cache Demonstration Agent with ReAct Pattern
Demonstrates the importance of KV cache through correct and incorrect implementations.
Uses local file system tools to read and search through code files.
"""
import json
import os
import re
import time
import logging
import random
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass, field, asdict
from enum import Enum
from datetime import datetime
from openai import OpenAI
import glob as glob_module
import subprocess
def _is_reasoning_model(model) -> bool:
"""True for models that emit reasoning_content and only accept temperature=1.
On the live Moonshot endpoint the whole current Kimi family reasons:
kimi-k2.5 / kimi-k2.6 / kimi-k2.7* / kimi-k3. The legacy moonshot-v1-*
chat models do NOT reason (and also do not report cached_tokens)."""
m = str(model or "").lower().replace("/", "-")
if "gpt-5" in m:
return True
return any(tag in m for tag in ("kimi-k2.5", "kimi-k2.6", "kimi-k2.7", "kimi-k3"))
def _reasoning_safe_temperature(model, requested=1.0):
"""Reasoning models (Kimi K2.5/K2.6/K2.7/K3, GPT-5, ...) only accept
temperature=1. Return 1 for those; otherwise the requested value so
non-reasoning providers (moonshot-v1, Doubao, DeepSeek) are unchanged."""
return 1 if _is_reasoning_model(model) else requested
def _reasoning_safe_max_tokens(model, requested=2000):
"""Reasoning models spend completion budget on hidden reasoning tokens
before emitting content / tool calls. Give them enough headroom so a
tool call is not truncated away; leave non-reasoning models unchanged."""
return max(requested, 4096) if _is_reasoning_model(model) else requested
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class KVCacheMode(Enum):
"""Different KV cache optimization modes"""
CORRECT = "correct" # Correct implementation with stable context
DYNAMIC_SYSTEM = "dynamic_system" # Changing system prompt with timestamp
SHUFFLED_TOOLS = "shuffled_tools" # Shuffling tool order each request
DYNAMIC_PROFILE = "dynamic_profile" # Changing user profile with credits
SLIDING_WINDOW = "sliding_window" # Only keeping recent 5 messages
TEXT_FORMAT = "text_format" # Formatting messages as plain text
@dataclass
class ToolCall:
"""Represents a single tool call"""
name: str
arguments: Dict[str, Any]
result: Any = None
error: Optional[str] = None
timestamp: float = field(default_factory=time.time)
@dataclass
class AgentMetrics:
"""Metrics for agent performance"""
ttft: float = 0.0 # Time to first token (first iteration)
ttft_per_iteration: List[float] = field(default_factory=list) # TTFT for each iteration
total_time: float = 0.0
iterations: int = 0
tool_calls: int = 0
cache_hits: int = 0
cache_misses: int = 0
prompt_tokens: int = 0
completion_tokens: int = 0
cached_tokens: int = 0
class LocalFileTools:
"""Local implementations of file system tools"""
def __init__(self, root_dir: str = "."):
self.root_dir = os.path.abspath(root_dir)
logger.info(f"File tools initialized with root: {self.root_dir}")
def read_file(self, file_path: str, offset: int = 0, size: int = None) -> Dict[str, Any]:
"""
Read contents of a file
Args:
file_path: Path to the file relative to root directory
offset: Line number to start reading from (0-based, default: 0)
size: Number of lines to read (default: None, read all)
Returns:
Dictionary with file contents or error
"""
try:
full_path = os.path.join(self.root_dir, file_path)
# Security check - ensure path is within root_dir
real_path = os.path.realpath(full_path)
if not real_path.startswith(self.root_dir):
return {
"error": f"Access denied: Path outside root directory",
"success": False
}
with open(real_path, 'r', encoding='utf-8', errors='ignore') as f:
lines = f.readlines()
total_lines = len(lines)
# Apply offset and size
if offset < 0:
offset = 0
if offset >= total_lines:
return {
"path": file_path,
"content": "",
"total_lines": total_lines,
"lines_read": 0,
"offset": offset,
"success": True,
"message": f"Offset {offset} exceeds file length ({total_lines} lines)"
}
# Determine end line
if size is None:
end = total_lines
else:
end = min(offset + size, total_lines)
# Get the requested lines
selected_lines = lines[offset:end]
content = ''.join(selected_lines)
# Apply size limit for safety (10KB)
truncated = False
if len(content) > 10000:
content = content[:10000]
truncated = True
return {
"path": file_path,
"content": content,
"total_lines": total_lines,
"lines_read": len(selected_lines),
"offset": offset,
"end_line": end,
"truncated": truncated,
"success": True
}
except FileNotFoundError:
return {
"error": f"File not found: {file_path}",
"success": False
}
except Exception as e:
return {
"error": f"Error reading file: {str(e)}",
"success": False
}
def find(self, pattern: str = "*", directory: str = ".") -> Dict[str, Any]:
"""
Find files matching a pattern (similar to Unix find command)
Args:
pattern: File name pattern (supports wildcards, default: "*" for all files)
directory: Directory to search in (relative to root_dir)
Returns:
Dictionary with list of matching files
"""
try:
# Handle directory path properly
if directory == ".":
search_dir = self.root_dir
else:
# Remove leading/trailing slashes for consistency
directory = directory.strip('/')
search_dir = os.path.join(self.root_dir, directory)
# Security check
real_path = os.path.realpath(search_dir)
if not real_path.startswith(self.root_dir):
return {
"error": f"Access denied: Path outside root directory",
"success": False
}
# Check if directory exists
if not os.path.exists(real_path):
return {
"error": f"Directory not found: {directory}",
"success": False
}
# Use glob to find matching files
matches = []
for root, dirs, files in os.walk(real_path):
# Filter hidden directories and __pycache__
dirs[:] = [d for d in dirs if not d.startswith('.') and d != '__pycache__']
for file in files:
# Skip hidden files and .pyc files
if file.startswith('.') or file.endswith('.pyc'):
continue
if glob_module.fnmatch.fnmatch(file, pattern):
# Get path relative to root_dir (not search_dir)
full_path = os.path.join(root, file)
rel_path = os.path.relpath(full_path, self.root_dir)
matches.append(rel_path)
# Sort for consistency
matches.sort()
# Limit results for demonstration
if len(matches) > 100:
matches = matches[:100]
truncated = True
else:
truncated = False
return {
"pattern": pattern,
"directory": directory,
"matches": matches,
"count": len(matches),
"truncated": truncated,
"success": True
}
except Exception as e:
return {
"error": f"Error finding files: {str(e)}",
"success": False
}
def grep(self, pattern: str, file_path: str = None, directory: str = None) -> Dict[str, Any]:
"""
Search for pattern in files (similar to Unix grep command)
Args:
pattern: Regular expression pattern to search for
file_path: Single file to search in (optional)
directory: Directory to search in (optional)
Returns:
Dictionary with matching lines
"""
try:
matches = []
files_searched = []
if file_path:
# Search in single file
full_path = os.path.join(self.root_dir, file_path)
real_path = os.path.realpath(full_path)
if not real_path.startswith(self.root_dir):
return {
"error": f"Access denied: Path outside root directory",
"success": False
}
files_to_search = [file_path]
elif directory:
# Search in directory
search_dir = os.path.join(self.root_dir, directory)
real_path = os.path.realpath(search_dir)
if not real_path.startswith(self.root_dir):
return {
"error": f"Access denied: Path outside root directory",
"success": False
}
# Find all text files in directory
files_to_search = []
for root, dirs, files in os.walk(real_path):
dirs[:] = [d for d in dirs if not d.startswith('.')]
for file in files:
if file.endswith(('.py', '.txt', '.md', '.json', '.yaml', '.yml', '.js', '.ts', '.jsx', '.tsx')):
rel_path = os.path.relpath(os.path.join(root, file), self.root_dir)
files_to_search.append(rel_path)
if len(files_to_search) >= 50: # Limit files for demonstration
break
else:
return {
"error": "Must specify either file_path or directory",
"success": False
}
# Compile regex pattern
regex = re.compile(pattern, re.IGNORECASE)
# Search in files
for file in files_to_search:
full_path = os.path.join(self.root_dir, file)
try:
with open(full_path, 'r', encoding='utf-8', errors='ignore') as f:
lines = f.readlines()
for i, line in enumerate(lines, 1):
if regex.search(line):
matches.append({
"file": file,
"line_num": i,
"line": line.strip()[:200] # Truncate long lines
})
if len(matches) >= 100: # Limit matches
break
files_searched.append(file)
except:
continue
if len(matches) >= 100:
break
return {
"pattern": pattern,
"matches": matches,
"files_searched": len(files_searched),
"match_count": len(matches),
"truncated": len(matches) >= 100,
"success": True
}
except Exception as e:
return {
"error": f"Error searching: {str(e)}",
"success": False
}
class KVCacheAgent:
"""
ReAct Agent with different KV cache optimization modes
"""
def __init__(self, api_key: str, mode: KVCacheMode = KVCacheMode.CORRECT,
model: str = "kimi-k2.6", root_dir: str = ".",
verbose: bool = True):
"""
Initialize the agent
Args:
api_key: API key for Moonshot/Kimi
mode: KV cache optimization mode
model: Model to use
root_dir: Root directory for file operations
verbose: If True, log detailed information
"""
# 默认走 Moonshot/Kimi 官方端点；若传入的是 OpenRouter key（sk-or-…），
# 则自动回退到 OpenRouter，并把 kimi-* 模型名映射为 moonshotai/kimi-k2。
from openrouter_fallback import (
OPENROUTER_BASE_URL,
is_openrouter_key,
map_model_to_openrouter,
)
if is_openrouter_key(api_key):
base_url = OPENROUTER_BASE_URL
model = map_model_to_openrouter(model)
else:
base_url = "https://api.moonshot.cn/v1"
self.client = OpenAI(
api_key=api_key,
base_url=base_url
)
self.model = model
self.mode = mode
self.verbose = verbose
self.tools = LocalFileTools(root_dir)
# Initialize conversation history
self.conversation_history = []
self.user_credits = 100 # For dynamic profile mode
self.metrics = AgentMetrics()
# Tool definitions in OpenAI format
self.tool_definitions = [
{
"type": "function",
"function": {
"name": "read_file",
"description": "Read the contents of a file, optionally specifying a line range",
"parameters": {
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Path to the file relative to root directory"
},
"offset": {
"type": "integer",
"description": "Line number to start reading from (0-based, default: 0)",
"default": 0
},
"size": {
"type": "integer",
"description": "Number of lines to read (default: read all lines)",
"default": None
}
},
"required": ["file_path"]
}
}
},
{
"type": "function",
"function": {
"name": "find",
"description": "Find files matching a pattern",
"parameters": {
"type": "object",
"properties": {
"pattern": {
"type": "string",
"description": "File name pattern (supports wildcards like *.py)"
},
"directory": {
"type": "string",
"description": "Directory to search in (default: current directory)",
"default": "."
}
},
"required": ["pattern"]
}
}
},
{
"type": "function",
"function": {
"name": "grep",
"description": "Search for a pattern in files",
"parameters": {
"type": "object",
"properties": {
"pattern": {
"type": "string",
"description": "Regular expression pattern to search for"
},
"file_path": {
"type": "string",
"description": "Single file to search in (optional)"
},
"directory": {
"type": "string",
"description": "Directory to search in (optional)"
}
},
"required": ["pattern"]
}
}
}
]
logger.info(f"Agent initialized with mode: {mode.value}, model: {model}")
def _get_system_prompt(self) -> str:
"""Get system prompt based on mode"""
base_prompt = """You are a helpful AI assistant with access to file system tools.
You can read files, find files by pattern, and search for text within files.
Use the ReAct pattern: Reason about what to do, then Act using tools, and Observe the results.
When asked to analyze or summarize code projects, be thorough:
1. First use 'find' to discover the structure
2. Then read key files to understand the content
3. Use 'grep' to search for specific patterns if needed
4. Once you have gathered sufficient information, provide your response
Always think step by step and use tools to gather information. When you have enough information to answer the user's question, simply provide your response without calling any tools."""
if self.mode == KVCacheMode.DYNAMIC_SYSTEM:
# Add timestamp to system prompt (breaks KV cache)
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
return f"{base_prompt}\n\nCURRENT TIME: {timestamp}"
return base_prompt
def _get_tools(self) -> List[Dict]:
"""Get tool definitions based on mode"""
tools = self.tool_definitions.copy()
if self.mode == KVCacheMode.SHUFFLED_TOOLS:
# Shuffle tool order (breaks KV cache)
random.shuffle(tools)
return tools
def _get_user_profile_message(self) -> Optional[Dict]:
"""Get user profile message for dynamic profile mode"""
if self.mode == KVCacheMode.DYNAMIC_PROFILE:
self.user_credits -= 1
return {
"role": "user",
"content": f"[User Profile: Premium user with {self.user_credits} credits remaining]"
}
return None
def _format_messages(self, task: str) -> List[Dict]:
"""Format messages based on mode - recreated each iteration for incorrect modes"""
messages = []
# Add system prompt (changes each time for DYNAMIC_SYSTEM mode)
messages.append({
"role": "system",
"content": self._get_system_prompt()
})
# Add user profile if in dynamic profile mode (changes each time)
profile_msg = self._get_user_profile_message()
if profile_msg:
messages.append(profile_msg)
if self.mode == KVCacheMode.SLIDING_WINDOW:
# Preserve all system and user messages, at the most recent 6 messages
if self.conversation_history:
# Include all system and user messages, and if there are at least 6 messages, include the last 6 messages regardless of role
system_user_msgs = [msg for msg in self.conversation_history if msg.get("role") in ("system", "user")]
messages.extend(system_user_msgs)
if len(self.conversation_history) >= 6:
messages.extend(self.conversation_history[-6:])
elif self.mode == KVCacheMode.TEXT_FORMAT:
# Format all history as plain text (breaks KV cache)
# Reformatting each time breaks structured format
if self.conversation_history:
history_text = "Previous conversation:\n"
for msg in self.conversation_history:
role = msg['role'].upper()
# Handle different message types
if role == "ASSISTANT":
# Also include any content
if msg.get('content'):
history_text += f"{role}: {msg['content']}\n"
# Check for tool calls
if msg.get('tool_calls'):
history_text += f"{role}: [Making tool calls]\n"
for tool_call in msg['tool_calls']:
func_name = tool_call.get('function', {}).get('name', 'unknown')
func_args = tool_call.get('function', {}).get('arguments', '{}')
history_text += f" - Calling {func_name} with args: {func_args}\n"
elif role == "TOOL":
# Format tool responses
tool_content = msg.get('content', '')
history_text += f"TOOL RESPONSE: {tool_content}\n"
else:
# USER, SYSTEM, or other roles
content = msg.get('content', '')
if content:
history_text += f"{role}: {content}\n"
messages.append({
"role": "user",
"content": history_text
})
else:
# For CORRECT, DYNAMIC_SYSTEM, SHUFFLED_TOOLS, DYNAMIC_PROFILE modes
# Include full conversation history
messages.extend(self.conversation_history)
# Add current task (always at the end)
messages.append({
"role": "user",
"content": task
})
return messages
def _execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
"""Execute a tool and return the result"""
tool_map = {
"read_file": self.tools.read_file,
"find": self.tools.find,
"grep": self.tools.grep
}
if tool_name not in tool_map:
return {"error": f"Unknown tool: {tool_name}", "success": False}
try:
# Filter out any unexpected arguments
tool_func = tool_map[tool_name]
# Get the expected arguments for this tool
import inspect
sig = inspect.signature(tool_func)
valid_args = {}
for param_name in sig.parameters:
if param_name in arguments:
valid_args[param_name] = arguments[param_name]
# Log if any arguments were filtered
filtered = set(arguments.keys()) - set(valid_args.keys())
if filtered and self.verbose:
logger.warning(f"Filtered unexpected arguments for {tool_name}: {filtered}")
return tool_func(**valid_args)
except Exception as e:
# Return error as tool result instead of raising
error_msg = f"Tool execution error: {str(e)}"
logger.error(f"{tool_name} failed: {error_msg}")
return {"error": error_msg, "success": False}
def execute_task(self, task: str, max_iterations: int = 50) -> Dict[str, Any]:
"""
Execute a task using ReAct pattern with standard OpenAI tool calling
Args:
task: The task to execute
max_iterations: Maximum number of iterations
Returns:
Task execution result with metrics
"""
start_time = time.time()
iteration = 0
final_answer = None
tool_calls = []
# Store the original task
original_task = task
while iteration < max_iterations:
iteration += 1
# CRITICAL: Message handling for KV cache demonstration
#
# CORRECT mode: Build messages once on first iteration, then keep appending
# - Maintains stable context → KV cache works efficiently
#
# INCORRECT modes: Recreate entire messages list from history each iteration
# - Forces complete context reconstruction → KV cache invalidated
# - Within an iteration, we still append to messages for proper API flow
# - But at the start of each new iteration, we rebuild from scratch
if self.mode == KVCacheMode.CORRECT:
# Correct mode: Build messages once, then keep using same list
if iteration == 1:
messages = self._format_messages(original_task)
else:
# Incorrect modes: Recreate messages from history each iteration
# This forces cache invalidation due to context changes
messages = self._format_messages(original_task)
# Prepare request
request_data = {
"model": self.model,
"messages": messages,
"temperature": _reasoning_safe_temperature(self.model, 0.7),
"max_tokens": _reasoning_safe_max_tokens(self.model, 2000)
}
# Add tools for all modes (TEXT_FORMAT still needs tools to work)
# TEXT_FORMAT only affects how conversation history is formatted, not tool availability
request_data["tools"] = self._get_tools()
request_data["tool_choice"] = "auto"
# Make API call
api_start = time.time()
try:
response = self.client.chat.completions.create(**request_data)
# Record TTFT for this iteration
iteration_ttft = time.time() - api_start
self.metrics.ttft_per_iteration.append(iteration_ttft)
# Record first iteration TTFT separately for backwards compatibility
if iteration == 1:
self.metrics.ttft = iteration_ttft
# Extract response
message = response.choices[0].message
# Print assistant content to console (always show, not just verbose)
if message.content:
print(f"\n🤖 Assistant (Iteration {iteration}):")
print("-" * 40)
print(message.content)
print("-" * 40)
# Log token usage and cache information
if hasattr(response, 'usage'):
usage = response.usage
self.metrics.prompt_tokens += usage.prompt_tokens
self.metrics.completion_tokens += usage.completion_tokens
# Check for cached tokens (Kimi specific)
# The cached_tokens field appears directly in the usage object
cached = 0
if hasattr(usage, 'cached_tokens'):
# Direct attribute on usage object
cached = usage.cached_tokens if usage.cached_tokens is not None else 0
self.metrics.cached_tokens += cached
if cached > 0:
self.metrics.cache_hits += 1
else:
self.metrics.cache_misses += 1
else:
# Try alternative locations
if hasattr(usage, 'prompt_tokens_details'):
details = usage.prompt_tokens_details
if details and hasattr(details, 'cached_tokens'):
cached = details.cached_tokens if details.cached_tokens is not None else 0
self.metrics.cached_tokens += cached
if cached > 0:
self.metrics.cache_hits += 1
else:
self.metrics.cache_misses += 1
# Debug logging when verbose and no cached tokens field found
if self.verbose and iteration > 1 and cached == 0:
logger.debug(f"Usage object attributes: {dir(usage)}")
logger.debug(f"Usage data: {usage}")
if self.verbose:
# Log with TTFT for this iteration
cache_info = f", cached={cached}" if cached > 0 else ""
logger.info(f"Iteration {iteration} - TTFT: {iteration_ttft:.3f}s, "
f"Tokens: prompt={usage.prompt_tokens}, "
f"completion={usage.completion_tokens}"
f"{cache_info}")
# Handle tool calls using standard OpenAI format
if hasattr(message, 'tool_calls') and message.tool_calls:
# Add the assistant message with tool calls
# Always append to messages for current iteration
messages.append(message.model_dump())
# Also append to history for next iteration
self.conversation_history.append(message.model_dump())
for tool_call in message.tool_calls:
function_name = tool_call.function.name
# Parse arguments safely
try:
function_args = json.loads(tool_call.function.arguments)
except json.JSONDecodeError as e:
logger.error(f"Failed to parse tool arguments: {e}")
function_args = {}
result = {"error": f"Invalid tool arguments: {str(e)}", "success": False}
else:
if self.verbose:
logger.info(f"Executing tool: {function_name} with args: {function_args}")
# Execute tool (errors are handled internally and returned as results)
result = self._execute_tool(function_name, function_args)
# Record tool call
tc = ToolCall(name=function_name, arguments=function_args, result=result)
tool_calls.append(tc)
# Print tool result summary
if result.get("success"):
# Success - show brief summary
if function_name == "read_file":
lines_info = f"{result.get('lines_read', 'unknown')} lines"
if result.get('offset', 0) > 0 or result.get('size'):
lines_info += f" (lines {result.get('offset', 0)}-{result.get('end_line', '?')})"
print(f" ✓ {function_name}: Read {lines_info}")
elif function_name == "find":
print(f" ✓ {function_name}: Found {result.get('count', 0)} files")
elif function_name == "grep":
print(f" ✓ {function_name}: Found {result.get('match_count', 0)} matches")
else:
print(f" ✓ {function_name}: Success")
else:
# Error - show the error message
print(f" ✗ {function_name}: {result.get('error', 'Unknown error')}")
# Add tool result as proper tool message (including errors)
tool_message = {
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result)
}
# Always append to messages for current iteration
messages.append(tool_message)
# Also append to history for next iteration
self.conversation_history.append(tool_message)
# Log if tool returned an error
if not result.get("success", True):
if self.verbose:
logger.warning(f"Tool {function_name} returned error: {result.get('error', 'Unknown error')}")
elif message.content:
# No tool calls - consider this the final answer
final_answer = message.content
# Always append to messages for current iteration
messages.append(message.model_dump())
# Also append to history for next iteration
self.conversation_history.append(message.model_dump())
if self.verbose:
logger.info("No tool calls in response - considering as final answer")
break
except Exception as e:
logger.error(f"Error in iteration {iteration}: {str(e)}")
break
# Calculate final metrics
self.metrics.total_time = time.time() - start_time
self.metrics.iterations = iteration
self.metrics.tool_calls = len(tool_calls)
return {
"success": final_answer is not None,
"final_answer": final_answer,
"iterations": iteration,
"tool_calls": tool_calls,
"metrics": self.metrics,
"mode": self.mode.value
}
def compare_implementations(api_key: str, task: str, root_dir: str = ".",
model: str = "kimi-k2.6") -> Dict[str, Any]:
"""
Compare different KV cache implementations
Args:
api_key: API key for Kimi
task: Task to execute
root_dir: Root directory for file operations
model: Model to use for all modes
Returns:
Comparison results
"""
results = {}
for mode in KVCacheMode:
logger.info(f"\n{'='*60}")
logger.info(f"Testing mode: {mode.value}")
logger.info(f"{'='*60}")
agent = KVCacheAgent(api_key=api_key, mode=mode, model=model, root_dir=root_dir, verbose=True)
result = agent.execute_task(task)
results[mode.value] = {
"success": result["success"],
"iterations": result["iterations"],
"tool_calls": result["tool_calls"],
"metrics": asdict(result["metrics"])
}
# Log summary
metrics = result["metrics"]
logger.info(f"\nMode: {mode.value}")
logger.info(f"First TTFT: {metrics.ttft:.3f}s")
# Log TTFT progression
if metrics.ttft_per_iteration:
ttft_summary = ", ".join([f"{t:.3f}s" for t in metrics.ttft_per_iteration[:5]])
if len(metrics.ttft_per_iteration) > 5:
ttft_summary += f"... ({len(metrics.ttft_per_iteration)} total)"
logger.info(f"TTFT per iteration: [{ttft_summary}]")
# Calculate TTFT improvement from first to last
if len(metrics.ttft_per_iteration) > 1:
improvement = (metrics.ttft_per_iteration[0] - metrics.ttft_per_iteration[-1]) / metrics.ttft_per_iteration[0] * 100
logger.info(f"TTFT improvement: {improvement:.1f}% (first vs last)")
logger.info(f"Total Time: {metrics.total_time:.3f}s")
logger.info(f"Cached Tokens: {metrics.cached_tokens}")
logger.info(f"Cache Hits: {metrics.cache_hits}")
logger.info(f"Cache Misses: {metrics.cache_misses}")
logger.info(f"Total Tokens: {metrics.prompt_tokens + metrics.completion_tokens}")
return results
demo_quick.py¶
#!/usr/bin/env python3
"""
Quick demonstration of KV cache impact
Shows the difference between correct and incorrect implementations
"""
import os
import sys
from agent import KVCacheAgent, KVCacheMode
def main():
"""Run a quick demo comparing correct vs incorrect implementation"""
# Get API key. 优先 Moonshot/Kimi；缺失时回退 OPENROUTER_API_KEY
# （KVCacheAgent 会自动切换到 OpenRouter 端点并映射模型名）。
api_key = (os.getenv("MOONSHOT_API_KEY") or os.getenv("KIMI_API_KEY")
or os.getenv("OPENROUTER_API_KEY"))
if not api_key:
print("❌ Please set MOONSHOT_API_KEY (or KIMI_API_KEY / OPENROUTER_API_KEY)")
print(" export MOONSHOT_API_KEY='your-api-key-here'")
sys.exit(1)
print("🚀 KV Cache Quick Demo")
print("="*60)
# Simple task that requires multiple tool calls
task = """Please do the following:
1. Find all Python files in the week1 directory
2. Read the main.py file from the context project
3. Search for the word 'agent' in week1 files
4. Provide a brief summary of what you found"""
print(f"📝 Task: {task}")
print("="*60)
# Test 1: Correct implementation
print("\n✅ Testing CORRECT implementation (with KV cache)...")
print("-"*60)
agent_correct = KVCacheAgent(
api_key=api_key,
mode=KVCacheMode.CORRECT,
root_dir="../..",
verbose=False # Set to True for detailed logs
)
result_correct = agent_correct.execute_task(task, max_iterations=10)
metrics_correct = result_correct["metrics"]
print(f"✓ TTFT: {metrics_correct.ttft:.3f}s")
print(f"✓ Total Time: {metrics_correct.total_time:.3f}s")
print(f"✓ Cached Tokens: {metrics_correct.cached_tokens:,}")
print(f"✓ Cache Hits: {metrics_correct.cache_hits}")
print(f"✓ Total Tokens Used: {metrics_correct.prompt_tokens + metrics_correct.completion_tokens:,}")
# Test 2: Incorrect implementation (dynamic system prompt)
print("\n❌ Testing INCORRECT implementation (dynamic system prompt)...")
print("-"*60)
agent_incorrect = KVCacheAgent(
api_key=api_key,
mode=KVCacheMode.DYNAMIC_SYSTEM,
root_dir="../..",
verbose=False
)
result_incorrect = agent_incorrect.execute_task(task, max_iterations=10)
metrics_incorrect = result_incorrect["metrics"]
print(f"✗ TTFT: {metrics_incorrect.ttft:.3f}s")
print(f"✗ Total Time: {metrics_incorrect.total_time:.3f}s")
print(f"✗ Cached Tokens: {metrics_incorrect.cached_tokens:,}")
print(f"✗ Cache Hits: {metrics_incorrect.cache_hits}")
print(f"✗ Total Tokens Used: {metrics_incorrect.prompt_tokens + metrics_incorrect.completion_tokens:,}")
# Comparison
print("\n📊 Performance Impact:")
print("="*60)
ttft_diff = ((metrics_incorrect.ttft - metrics_correct.ttft) / metrics_correct.ttft) * 100
time_diff = ((metrics_incorrect.total_time - metrics_correct.total_time) / metrics_correct.total_time) * 100
cache_lost = metrics_correct.cached_tokens - metrics_incorrect.cached_tokens
print(f"⚡ TTFT increased by: {ttft_diff:.1f}%")
print(f"⏱️ Total time increased by: {time_diff:.1f}%")
print(f"💾 Cache tokens lost: {cache_lost:,}")
if ttft_diff > 50:
print("\n⚠️ Dynamic system prompts severely impact performance!")
print(" Even small context changes can invalidate the entire KV cache.")
print("\n💡 Key Takeaway:")
print(" Maintaining stable context is crucial for LLM performance.")
print(" Small implementation details can have major performance impacts!")
if __name__ == "__main__":
main()
main.py¶
"""
Main script to demonstrate KV cache importance
Runs the ReAct agent with different implementations and compares performance
"""
import os
import sys
import glob
import json
import argparse
import logging
from typing import Dict, List, Any
from datetime import datetime
from dataclasses import asdict
from agent import KVCacheAgent, KVCacheMode, AgentMetrics, compare_implementations
# Default model (Moonshot / Kimi). The whole current Kimi family (k2.5/k2.6/
# k2.7/k3) reports cached_tokens for automatic prefix caching AND reasons, so it
# only accepts temperature=1 (agent.py handles that automatically). kimi-k2.6 has
# the lightest reasoning footprint of the cache-reporting models, giving the
# cleanest TTFT while still exposing the prefix-cache hit metric this demo needs.
# (The non-reasoning moonshot-v1-* models do NOT report cached_tokens, so they
# cannot demonstrate the cache effect.)
DEFAULT_MODEL = "kimi-k2.6"
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('kv_cache_demo.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Metrics helpers (shared by live comparison and offline report)
# ---------------------------------------------------------------------------
def _coerce_metrics(metrics: Any) -> Dict[str, Any]:
"""Normalize a stored metrics value into a plain dict.
Handles both formats found in result files:
- dict: produced by --compare (asdict) and by the fixed --mode path
- str : legacy single-mode files that stored repr(AgentMetrics(...))
because json.dump used default=str
"""
if isinstance(metrics, dict):
return metrics
if isinstance(metrics, str) and metrics.startswith("AgentMetrics("):
# Safe eval: only AgentMetrics is exposed, no builtins.
try:
obj = eval(metrics, {"__builtins__": {}}, {"AgentMetrics": AgentMetrics})
return asdict(obj)
except Exception as e: # pragma: no cover - defensive
logger.warning(f"Could not parse legacy metrics string: {e}")
return {}
def _avg_ttft(m: Dict[str, Any]) -> float:
"""Average TTFT across iterations, falling back to first-iteration TTFT."""
lst = m.get("ttft_per_iteration") or []
return sum(lst) / len(lst) if lst else float(m.get("ttft", 0.0) or 0.0)
def _hit_rate(m: Dict[str, Any]) -> float:
total = (m.get("cache_hits", 0) or 0) + (m.get("cache_misses", 0) or 0)
return (m.get("cache_hits", 0) or 0) / total * 100 if total else 0.0
def _billable_tokens(m: Dict[str, Any], cache_price_ratio: float) -> float:
"""Illustrative billable prompt tokens under a prompt-cache discount.
cached tokens are charged at cache_price_ratio of the normal price; the
rest at full price. This is a transparent function of the *measured*
token counts and a user-supplied ratio - it is not a fabricated
provider-specific price.
"""
prompt = m.get("prompt_tokens", 0) or 0
cached = m.get("cached_tokens", 0) or 0
cached = min(cached, prompt)
return (prompt - cached) + cached * cache_price_ratio
def print_comparison_table(results: Dict[str, Any], cache_price_ratio: float = 0.1) -> None:
"""Render the cross-strategy comparison table (latency / cache / cost)."""
print(f"\n{'Mode':<16} {'Iters':<6} {'1st TTFT':<10} {'Avg TTFT':<10} "
f"{'Total(s)':<10} {'Prompt':<9} {'Cached':<9} {'Hit%':<7} "
f"{'Cache%':<8} {'Bill.Tok':<10} {'Save%':<7}")
print("-" * 112)
for mode, data in results.items():
m = _coerce_metrics(data.get("metrics", {}))
prompt = m.get("prompt_tokens", 0) or 0
cached = m.get("cached_tokens", 0) or 0
iters = data.get("iterations", m.get("iterations", 0)) or 0
cache_pct = cached / prompt * 100 if prompt else 0.0
billable = _billable_tokens(m, cache_price_ratio)
save_pct = (prompt - billable) / prompt * 100 if prompt else 0.0
print(f"{mode:<16} {iters:<6} {float(m.get('ttft', 0.0) or 0.0):<10.3f} "
f"{_avg_ttft(m):<10.3f} {float(m.get('total_time', 0.0) or 0.0):<10.3f} "
f"{prompt:<9,} {cached:<9,} {_hit_rate(m):<7.1f} "
f"{cache_pct:<8.1f} {billable:<10,.0f} {save_pct:<7.1f}")
print("-" * 112)
print(f"注：Bill.Tok / Save% 假设缓存 token 按正常价的 {cache_price_ratio:.0%} 计费"
f"（可用 --cache-price-ratio 调整），仅为成本示意，非某家服务商实际报价。")
def load_result_files(paths: List[str]) -> Dict[str, Any]:
"""Load result_*.json files into a {mode: {...}} dict for offline reporting."""
results: Dict[str, Any] = {}
for path in sorted(paths):
try:
with open(path, 'r') as f:
data = json.load(f)
except Exception as e:
logger.warning(f"Skipping {path}: {e}")
continue
# A comparison_*.json holds many modes; a result_*.json holds one.
if "mode" not in data and all(isinstance(v, dict) and "metrics" in v
for v in data.values()):
for mode, entry in data.items():
results[mode] = {"metrics": _coerce_metrics(entry.get("metrics", {})),
"iterations": entry.get("iterations"),
"_source": path}
else:
mode = data.get("mode", os.path.splitext(os.path.basename(path))[0])
results[mode] = {"metrics": _coerce_metrics(data.get("metrics", {})),
"iterations": data.get("iterations"),
"_source": path}
return results
def run_report(inputs: List[str] = None, cache_price_ratio: float = 0.1) -> None:
"""Offline: build the comparison table from existing result_*.json files.
No API key required - reads previously saved runs so the final result is
legible in one command without re-hitting the model.
"""
if not inputs:
inputs = ["result_*.json", "comparison_*.json"]
paths: List[str] = []
for item in inputs:
if os.path.isdir(item):
paths.extend(glob.glob(os.path.join(item, "result_*.json")))
paths.extend(glob.glob(os.path.join(item, "comparison_*.json")))
else:
paths.extend(glob.glob(item))
paths = sorted(set(paths))
if not paths:
logger.error("未找到任何 result_*.json / comparison_*.json 结果文件。"
"请先运行 --mode 或 --compare 生成结果，或用 --input 指定路径。")
sys.exit(1)
results = load_result_files(paths)
print("\n" + "=" * 112)
print("KV CACHE 离线对比报告（基于已保存的实测结果）")
print("=" * 112)
print(f"数据来源（{len(paths)} 个文件）:")
for mode, data in results.items():
print(f" • {mode:<16} ← {os.path.basename(data.get('_source', '?'))}")
print_comparison_table(results, cache_price_ratio)
print("\n📝 说明：不同结果文件可能来自不同任务/时间，绝对数值仅供同一次运行内横向对比；"
"如需严格对照，请用 --compare 在同一任务下一次性生成全部模式的数据。")
def create_summary_task() -> str:
"""Create a task that requires reading multiple files"""
return """Please analyze and summarize all the projects in the week1 and week2 directories.
For each project:
1. Find all Python files
2. Read the main files and understand the functionality
3. Identify the key features and purpose
4. Provide a comprehensive summary
Start with week1 projects, then move to week2. Be thorough in your analysis."""
def run_single_mode(api_key: str, mode: str, task: str = None, root_dir: str = "../..",
model: str = DEFAULT_MODEL, output: str = None):
"""
Run agent in a single mode
Args:
api_key: API key for Kimi
mode: KV cache mode to use
task: Custom task (optional)
root_dir: Root directory for file operations (default: "../.." = /projects from kv-cache dir)
model: Model to use
output: Output path for the result JSON (optional; auto-named if omitted)
"""
# Parse mode
mode_map = {
"correct": KVCacheMode.CORRECT,
"dynamic_system": KVCacheMode.DYNAMIC_SYSTEM,
"shuffled_tools": KVCacheMode.SHUFFLED_TOOLS,
"dynamic_profile": KVCacheMode.DYNAMIC_PROFILE,
"sliding_window": KVCacheMode.SLIDING_WINDOW,
"text_format": KVCacheMode.TEXT_FORMAT
}
if mode not in mode_map:
logger.error(f"Invalid mode: {mode}")
logger.info(f"Valid modes: {', '.join(mode_map.keys())}")
return
# Use default task if not provided
if not task:
task = create_summary_task()
logger.info(f"Running in mode: {mode}")
logger.info(f"Task: {task}")
logger.info("="*80)
# Create agent and execute task
agent = KVCacheAgent(
api_key=api_key,
mode=mode_map[mode],
model=model,
root_dir=root_dir,
verbose=True
)
result = agent.execute_task(task, max_iterations=30)
# Print results
print("\n" + "="*80)
print(f"EXECUTION RESULTS - Mode: {mode}")
print("="*80)
metrics = result["metrics"]
print(f"\n📊 Performance Metrics:")
print(f" • Time to First Token (TTFT): {metrics.ttft:.3f} seconds")
# Show TTFT progression
if metrics.ttft_per_iteration:
print(f" • TTFT per iteration:")
for i, ttft in enumerate(metrics.ttft_per_iteration, 1):
print(f" Iteration {i}: {ttft:.3f}s")
# Show improvement
if len(metrics.ttft_per_iteration) > 1:
first_ttft = metrics.ttft_per_iteration[0]
last_ttft = metrics.ttft_per_iteration[-1]
avg_after_first = sum(metrics.ttft_per_iteration[1:]) / len(metrics.ttft_per_iteration[1:])
print(f" • TTFT Analysis:")
print(f" First iteration: {first_ttft:.3f}s")
print(f" Last iteration: {last_ttft:.3f}s")
print(f" Average (after first): {avg_after_first:.3f}s")
improvement = (first_ttft - last_ttft) / first_ttft * 100
print(f" Improvement: {improvement:.1f}%")
print(f" • Total Execution Time: {metrics.total_time:.3f} seconds")
print(f" • Iterations: {result['iterations']}")
print(f" • Tool Calls: {len(result['tool_calls'])}")
print(f"\n🔄 Cache Statistics:")
print(f" • Cached Tokens: {metrics.cached_tokens:,}")
print(f" • Cache Hits: {metrics.cache_hits}")
print(f" • Cache Misses: {metrics.cache_misses}")
if metrics.cache_hits + metrics.cache_misses > 0:
hit_rate = metrics.cache_hits / (metrics.cache_hits + metrics.cache_misses) * 100
print(f" • Cache Hit Rate: {hit_rate:.1f}%")
print(f"\n💰 Token Usage:")
print(f" • Prompt Tokens: {metrics.prompt_tokens:,}")
print(f" • Completion Tokens: {metrics.completion_tokens:,}")
print(f" • Total Tokens: {metrics.prompt_tokens + metrics.completion_tokens:,}")
if metrics.prompt_tokens > 0:
cache_ratio = metrics.cached_tokens / metrics.prompt_tokens * 100
print(f" • Cache Ratio: {cache_ratio:.1f}% of prompt tokens cached")
# Show tool calls summary
if result["tool_calls"]:
print(f"\n🔧 Tool Calls Summary:")
tool_counts = {}
for tc in result["tool_calls"]:
tool_counts[tc.name] = tool_counts.get(tc.name, 0) + 1
for tool_name, count in tool_counts.items():
print(f" • {tool_name}: {count} calls")
# Save detailed results
output_file = output or f"result_{mode}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(output_file, 'w') as f:
# Convert to serializable format. Store metrics as a dict (via asdict)
# so the file can be re-loaded later by --report; tool calls likewise.
result_copy = result.copy()
result_copy["metrics"] = asdict(result["metrics"])
result_copy["tool_calls"] = [
{
"name": tc.name,
"arguments": tc.arguments,
"timestamp": tc.timestamp
}
for tc in result["tool_calls"]
]
json.dump(result_copy, f, indent=2, default=str)
print(f"\n💾 Detailed results saved to: {output_file}")
def select_mode_interactive():
"""
Interactive mode selection menu
Returns:
Selected mode string or None for all modes
"""
modes = [
("correct", "✅ Correct Implementation - Optimal KV cache usage"),
("dynamic_system", "❌ Dynamic System Prompt - Adds timestamps"),
("shuffled_tools", "❌ Shuffled Tools - Randomizes tool order"),
("dynamic_profile", "❌ Dynamic Profile - Updates user credits"),
("sliding_window", "❌ Sliding Window - Keeps only recent messages"),
("text_format", "❌ Text Format - Plain text instead of structured"),
("compare", "📊 Compare All - Run all modes and compare"),
]
print("\n" + "="*60)
print("KV CACHE DEMONSTRATION - MODE SELECTION")
print("="*60)
print("\nSelect a mode to run:\n")
for i, (mode, description) in enumerate(modes, 1):
print(f" {i}. {description}")
print("\n 0. Exit")
print("-"*60)
while True:
try:
choice = input("\nEnter your choice (0-7): ").strip()
choice_num = int(choice)
if choice_num == 0:
print("Exiting...")
sys.exit(0)
elif 1 <= choice_num <= 6:
selected = modes[choice_num - 1][0]
print(f"\n✓ Selected: {modes[choice_num - 1][1]}")
return selected
elif choice_num == 7:
print("\n✓ Selected: Compare all modes")
return "compare"
else:
print("Invalid choice. Please enter a number between 0 and 7.")
except ValueError:
print("Invalid input. Please enter a number.")
except KeyboardInterrupt:
print("\n\nExiting...")
sys.exit(0)
def run_comparison(api_key: str, task: str = None, root_dir: str = "../..",
model: str = DEFAULT_MODEL, output: str = None,
cache_price_ratio: float = 0.1):
"""
Run comparison across all modes
Args:
api_key: API key for Kimi
task: Custom task (optional)
root_dir: Root directory for file operations (default: "../.." = /projects from kv-cache dir)
model: Model to use for all modes
output: Output path for the comparison JSON (optional; auto-named if omitted)
cache_price_ratio: Assumed price of a cached token vs a normal token (cost column)
"""
# Use default task if not provided
if not task:
task = create_summary_task()
logger.info("Starting KV Cache Comparison Study")
logger.info(f"Task: {task[:200]}...")
logger.info("="*80)
# Run comparison
results = compare_implementations(api_key, task, root_dir, model=model)
# Print comparison table
print("\n" + "="*112)
print("KV CACHE COMPARISON RESULTS")
print("="*112)
print_comparison_table(results, cache_price_ratio)
# Analyze results
print("\n" + "="*80)
print("ANALYSIS")
print("="*80)
# Find best and worst performers
correct_metrics = results["correct"]["metrics"]
print("\n🏆 Performance Impact (compared to correct implementation):")
for mode, data in results.items():
if mode == "correct":
continue
metrics = data["metrics"]
ttft_diff = ((metrics["ttft"] - correct_metrics["ttft"]) / correct_metrics["ttft"]) * 100
total_diff = ((metrics["total_time"] - correct_metrics["total_time"]) / correct_metrics["total_time"]) * 100
cache_diff = correct_metrics["cached_tokens"] - metrics["cached_tokens"]
print(f"\n{mode}:")
print(f" • TTFT: {'+' if ttft_diff > 0 else ''}{ttft_diff:.1f}% "
f"({'slower' if ttft_diff > 0 else 'faster'})")
print(f" • Total Time: {'+' if total_diff > 0 else ''}{total_diff:.1f}% "
f"({'slower' if total_diff > 0 else 'faster'})")
print(f" • Lost Cached Tokens: {cache_diff:,}")
# Show TTFT progression comparison
print("\n📈 TTFT Progression (first 5 iterations):")
for mode, data in results.items():
metrics = data["metrics"]
ttft_list = metrics.get("ttft_per_iteration", [])[:5]
if ttft_list:
ttft_str = " → ".join([f"{t:.2f}s" for t in ttft_list])
print(f" {mode:<20}: {ttft_str}")
# Key insights
print("\n📝 Key Insights:")
print(" 1. The correct implementation maintains stable context for optimal KV cache usage")
print(" 2. TTFT improves dramatically after first iteration when cache is utilized")
print(" 3. Dynamic system prompts invalidate the entire cache on each request")
print(" 4. Shuffling tools breaks cache even though the functionality is identical")
print(" 5. Dynamic user profiles add unnecessary context changes")
print(" 6. Sliding windows may seem to reduce context but actually harm cache efficiency")
print(" 7. Text formatting breaks the structured message format that enables caching")
# Save comparison results
output_file = output or f"comparison_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(output_file, 'w') as f:
json.dump(results, f, indent=2, default=str)
print(f"\n💾 Comparison results saved to: {output_file}")
def main():
"""Main entry point"""
parser = argparse.ArgumentParser(
description="KV Cache 实验：用 ReAct Agent 对比不同上下文构造策略对前缀缓存"
"（KV Cache / Prompt Cache）命中率、TTFT 延迟与成本的影响。",
epilog="示例：\n"
" python main.py --mode correct # 运行单个策略\n"
" python main.py --compare # 一次跑完所有策略并打印对比表\n"
" python main.py --report # 离线：读取已有 result_*.json 打印对比表（无需 API Key）\n"
" python main.py --mode sliding_window --model kimi-k2.6 --output run.json\n"
"\n可选策略（--mode）：correct, dynamic_system, shuffled_tools,\n"
" dynamic_profile, sliding_window, text_format",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("--api-key", type=str,
help="Moonshot/Kimi API Key（也可用环境变量 MOONSHOT_API_KEY）")
parser.add_argument("--model", type=str, default=DEFAULT_MODEL,
help=f"使用的模型名（默认：{DEFAULT_MODEL}）")
parser.add_argument("--mode", type=str,
help="运行单个策略：correct / dynamic_system / shuffled_tools / "
"dynamic_profile / sliding_window / text_format")
parser.add_argument("--compare", action="store_true",
help="依次运行全部策略并打印横向对比表（需要 API Key）")
parser.add_argument("--report", action="store_true",
help="离线模式：从已保存的 result_*.json / comparison_*.json 生成对比表（无需 API Key）")
parser.add_argument("--input", type=str, nargs="*", default=None,
help="配合 --report：指定结果文件、通配符或目录（默认：当前目录下的 result_*.json 与 comparison_*.json）")
parser.add_argument("--output", type=str,
help="结果 JSON 的输出路径（默认按模式和时间戳自动命名）")
parser.add_argument("--cache-price-ratio", type=float, default=0.1,
help="成本估算中缓存 token 相对正常 token 的计费比例（默认：0.1，即缓存读取按一折计），仅作示意")
parser.add_argument("--task", type=str, help="自定义任务描述（默认：分析并总结项目代码）")
parser.add_argument("--root-dir", type=str, default="../..",
help="文件工具的根目录（默认：../.. 即仓库根，供 Agent 读取代码）")
parser.add_argument("--interactive", action="store_true", default=True,
help="交互式菜单选择策略（默认开启）")
parser.add_argument("--no-interactive", dest="interactive", action="store_false",
help="关闭交互式菜单")
args = parser.parse_args()
# Offline report needs no API key - handle it first.
if args.report:
run_report(args.input, args.cache_price_ratio)
return
# Get API key. 优先 Moonshot/Kimi 官方 key；缺失时回退到 OPENROUTER_API_KEY
# （KVCacheAgent 会据此自动切换到 OpenRouter 端点并映射模型名）。
api_key = (args.api_key or os.getenv("MOONSHOT_API_KEY")
or os.getenv("KIMI_API_KEY") or os.getenv("OPENROUTER_API_KEY"))
if not api_key:
logger.error("请通过 --api-key 或环境变量 MOONSHOT_API_KEY / KIMI_API_KEY / "
"OPENROUTER_API_KEY 提供 API Key；"
"若只想查看已有结果，可使用 --report（无需 API Key）。")
sys.exit(1)
# Run based on mode
if args.compare:
# Explicit --compare flag overrides interactive mode
run_comparison(api_key, args.task, args.root_dir, args.model, args.output,
args.cache_price_ratio)
elif args.mode:
# Explicit --mode flag overrides interactive mode
run_single_mode(api_key, args.mode, args.task, args.root_dir, args.model, args.output)
elif args.interactive and not args.task:
# Interactive mode selection (default)
selected_mode = select_mode_interactive()
if selected_mode == "compare":
run_comparison(api_key, args.task, args.root_dir, args.model, args.output,
args.cache_price_ratio)
else:
run_single_mode(api_key, selected_mode, args.task, args.root_dir, args.model, args.output)
else:
# If task is provided without mode, ask which mode to use
if args.task:
print(f"\n📝 Custom task provided: {args.task}")
selected_mode = select_mode_interactive()
if selected_mode == "compare":
run_comparison(api_key, args.task, args.root_dir, args.model, args.output,
args.cache_price_ratio)
else:
run_single_mode(api_key, selected_mode, args.task, args.root_dir, args.model, args.output)
else:
# Fallback to interactive mode
selected_mode = select_mode_interactive()
if selected_mode == "compare":
run_comparison(api_key, args.task, args.root_dir, args.model, args.output,
args.cache_price_ratio)
else:
run_single_mode(api_key, selected_mode, args.task, args.root_dir, args.model, args.output)
if __name__ == "__main__":
main()
openrouter_fallback.py¶
"""Universal OpenRouter fallback helper (Chapter 2 experiments).
Goal: every experiment keeps working when the direct provider key is missing
but ``OPENROUTER_API_KEY`` is present. Default behavior is fully preserved:
* If a primary provider key is present -> use the primary provider unchanged.
* Else if ``OPENROUTER_API_KEY`` is present -> route through OpenRouter
(base_url=https://openrouter.ai/api/v1) and translate the model id.
* Else raise a clear error listing every accepted key.
Model translation (only applied when the OpenRouter fallback is active):
* ids already containing "/" are passed through unchanged;
* gpt-* / o1* / o3* / o4* / chatgpt* -> "openai/<id>";
* claude-* -> "anthropic/claude-opus-4.8";
* kimi-* -> "moonshotai/kimi-k2.6";
* anything else -> passed through unchanged.
"""
import os
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
def is_openrouter_key(api_key):
"""OpenRouter keys reliably start with ``sk-or-``."""
return bool(api_key) and api_key.startswith("sk-or-")
def map_model_to_openrouter(model):
"""Translate a bare provider model id into an OpenRouter-qualified id."""
if not model or "/" in model:
return model
low = model.lower()
if low.startswith(("gpt-", "gpt5", "o1", "o3", "o4", "chatgpt")):
return "openai/" + model
if low.startswith("claude"):
return "anthropic/claude-opus-4.8"
if low.startswith("kimi"):
return "moonshotai/kimi-k2.6"
return model
def resolve_llm(model=None, primary_keys=("OPENAI_API_KEY",), primary_base_url=None):
"""Resolve ``(api_key, base_url, model)`` honoring the primary->OpenRouter fallback.
Args:
model: requested model id (may be remapped when the fallback activates).
primary_keys: env var names checked in order for the primary provider.
primary_base_url: base_url used when a primary key is found
(``None`` means the OpenAI SDK default / official endpoint).
"""
or_key = os.getenv("OPENROUTER_API_KEY")
# gpt-5.x (incl. gpt-5.6*) needs OpenAI org-verification on the direct API;
# when an OpenRouter key is present, prefer routing these ids through it.
if or_key and model and model.lower().startswith("gpt-5"):
return or_key, OPENROUTER_BASE_URL, map_model_to_openrouter(model)
for env in primary_keys:
key = os.getenv(env)
if key:
return key, primary_base_url, model
if or_key:
return or_key, OPENROUTER_BASE_URL, map_model_to_openrouter(model)
accepted = ", ".join(list(primary_keys) + ["OPENROUTER_API_KEY"])
raise RuntimeError(
"No LLM API key found. Set one of the primary keys or the universal "
"fallback. Accepted keys: " + accepted + ". See env.example."
)
test_api.py¶
#!/usr/bin/env python3
"""
Test script to verify the updated agent works with standard OpenAI tool calling
"""
import os
import sys
import json
from agent import KVCacheAgent, KVCacheMode
def test_tool_calling():
"""Test that the agent correctly uses OpenAI tool calling format"""
# Get API key
api_key = os.getenv("MOONSHOT_API_KEY")
if not api_key:
print("❌ Please set MOONSHOT_API_KEY environment variable")
sys.exit(1)
print("🧪 Testing Standard OpenAI Tool Calling Format")
print("="*60)
# Simple task that requires tool calls
task = "Find all Python files in the week1/context directory and tell me how many there are."
print(f"📝 Task: {task}")
print("-"*60)
# Create agent with correct implementation
agent = KVCacheAgent(
api_key=api_key,
mode=KVCacheMode.CORRECT,
root_dir="../..",
verbose=True # Enable verbose to see tool calls
)
# Execute task
result = agent.execute_task(task, max_iterations=5)
# Check results
print("\n" + "="*60)
print("📊 Results:")
print(f"✓ Success: {result['success']}")
print(f"✓ Iterations: {result['iterations']}")
print(f"✓ Tool Calls Made: {len(result['tool_calls'])}")
if result['tool_calls']:
print("\n🔧 Tool Calls:")
for tc in result['tool_calls']:
print(f" • {tc.name}({tc.arguments})")
if tc.result and tc.result.get('success'):
if tc.name == 'find':
print(f" → Found {tc.result.get('count', 0)} files")
if result['final_answer']:
print(f"\n💬 Final Answer:")
print(f" {result['final_answer'][:200]}...")
# Test metrics
metrics = result['metrics']
print(f"\n📈 Performance Metrics:")
print(f" • TTFT: {metrics.ttft:.3f}s")
print(f" • Total Time: {metrics.total_time:.3f}s")
print(f" • Cached Tokens: {metrics.cached_tokens}")
print("\n✅ Tool calling test completed successfully!")
if __name__ == "__main__":
test_tool_calling()
test_cache_invalidation.py¶
#!/usr/bin/env python3
"""
Test script to verify KV cache is properly invalidated in incorrect modes
"""
import os
import sys
import logging
from agent import KVCacheAgent, KVCacheMode
# Set up logging to see details
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def test_cache_invalidation():
"""Test that incorrect modes properly invalidate KV cache each iteration"""
# Get API key
api_key = os.getenv("MOONSHOT_API_KEY")
if not api_key:
print("❌ Please set MOONSHOT_API_KEY environment variable")
sys.exit(1)
print("🔬 Testing KV Cache Invalidation")
print("="*60)
# Simple task that requires multiple iterations
task = "Find Python files in week1/context and tell me how many there are."
print(f"Task: {task}")
print("-"*40)
# Test 1: CORRECT mode (should use cache)
print("\n1️⃣ Testing CORRECT mode (should use cache):")
agent_correct = KVCacheAgent(
api_key=api_key,
mode=KVCacheMode.CORRECT,
root_dir="../..",
verbose=True
)
result_correct = agent_correct.execute_task(task, max_iterations=5)
metrics_correct = result_correct["metrics"]
print(f"\n Results for CORRECT mode:")
print(f" • Iterations: {result_correct['iterations']}")
print(f" • TTFT per iteration: {[f'{t:.2f}s' for t in metrics_correct.ttft_per_iteration]}")
print(f" • Cached tokens: {metrics_correct.cached_tokens}")
print(f" • Cache hits: {metrics_correct.cache_hits}")
# Test 2: DYNAMIC_SYSTEM mode (should NOT use cache)
print("\n2️⃣ Testing DYNAMIC_SYSTEM mode (should NOT use cache):")
agent_dynamic = KVCacheAgent(
api_key=api_key,
mode=KVCacheMode.DYNAMIC_SYSTEM,
root_dir="../..",
verbose=True
)
result_dynamic = agent_dynamic.execute_task(task, max_iterations=5)
metrics_dynamic = result_dynamic["metrics"]
print(f"\n Results for DYNAMIC_SYSTEM mode:")
print(f" • Iterations: {result_dynamic['iterations']}")
print(f" • TTFT per iteration: {[f'{t:.2f}s' for t in metrics_dynamic.ttft_per_iteration]}")
print(f" • Cached tokens: {metrics_dynamic.cached_tokens}")
print(f" • Cache hits: {metrics_dynamic.cache_hits}")
# Analysis
print("\n" + "="*60)
print("📊 ANALYSIS:")
print("-"*40)
# Check TTFT improvement
if len(metrics_correct.ttft_per_iteration) > 1:
correct_improvement = (metrics_correct.ttft_per_iteration[0] - metrics_correct.ttft_per_iteration[-1]) / metrics_correct.ttft_per_iteration[0] * 100
print(f"CORRECT mode TTFT improvement: {correct_improvement:.1f}%")
if len(metrics_dynamic.ttft_per_iteration) > 1:
dynamic_improvement = (metrics_dynamic.ttft_per_iteration[0] - metrics_dynamic.ttft_per_iteration[-1]) / metrics_dynamic.ttft_per_iteration[0] * 100
print(f"DYNAMIC mode TTFT improvement: {dynamic_improvement:.1f}%")
# Verify cache behavior
print("\n✅ Verification:")
if metrics_correct.cached_tokens > 0:
print(f" ✓ CORRECT mode used cache: {metrics_correct.cached_tokens} tokens")
else:
print(f" ✗ CORRECT mode did NOT use cache (unexpected!)")
if metrics_dynamic.cached_tokens == 0:
print(f" ✓ DYNAMIC mode did NOT use cache (expected)")
else:
print(f" ✗ DYNAMIC mode used cache: {metrics_dynamic.cached_tokens} tokens (unexpected!)")
# Check TTFT consistency
print("\n🔍 TTFT Consistency Check:")
if len(metrics_correct.ttft_per_iteration) > 2:
# CORRECT mode should show improvement after first iteration
first_ttft = metrics_correct.ttft_per_iteration[0]
avg_rest = sum(metrics_correct.ttft_per_iteration[1:]) / len(metrics_correct.ttft_per_iteration[1:])
if avg_rest < first_ttft * 0.7: # At least 30% improvement
print(f" ✓ CORRECT mode shows cache benefit (first: {first_ttft:.2f}s, avg rest: {avg_rest:.2f}s)")
else:
print(f" ⚠️ CORRECT mode improvement less than expected")
if len(metrics_dynamic.ttft_per_iteration) > 2:
# DYNAMIC mode should NOT show significant improvement
all_ttfts = metrics_dynamic.ttft_per_iteration
min_ttft = min(all_ttfts)
max_ttft = max(all_ttfts)
if (max_ttft - min_ttft) / max_ttft < 0.3: # Less than 30% variation
print(f" ✓ DYNAMIC mode shows consistent TTFT (no cache benefit)")
else:
print(f" ⚠️ DYNAMIC mode shows unexpected TTFT variation")
print("\n💡 Key Finding:")
print("The CORRECT mode should show significant TTFT improvement after the first")
print("iteration due to KV cache, while incorrect modes should maintain")
print("consistently high TTFT because the cache is invalidated on each iteration.")
if __name__ == "__main__":
test_cache_invalidation()
test_cached_tokens.py¶
#!/usr/bin/env python3
"""
Test script to verify cached tokens are being parsed correctly from Kimi API
"""
import os
import sys
from agent import KVCacheAgent, KVCacheMode
def test_cached_tokens():
"""Test that cached tokens are correctly parsed from API response"""
# Get API key
api_key = os.getenv("MOONSHOT_API_KEY")
if not api_key:
print("❌ Please set MOONSHOT_API_KEY environment variable")
sys.exit(1)
print("🔍 Testing Cached Tokens Parsing")
print("="*60)
# Simple task that requires a few iterations
task = "Find Python files in week1/context directory and tell me how many there are."
print(f"Task: {task}")
print("-"*40)
# Run with correct implementation (should use cache)
print("\nRunning agent with CORRECT implementation...")
agent = KVCacheAgent(
api_key=api_key,
mode=KVCacheMode.CORRECT,
root_dir="../..",
verbose=True # Enable verbose to see token logging
)
result = agent.execute_task(task, max_iterations=5)
metrics = result["metrics"]
print("\n" + "="*60)
print("📊 Cache Token Results:")
print(f" • Total iterations: {result['iterations']}")
print(f" • Cached tokens accumulated: {metrics.cached_tokens}")
print(f" • Cache hits: {metrics.cache_hits}")
print(f" • Cache misses: {metrics.cache_misses}")
# Check each iteration's TTFT
if metrics.ttft_per_iteration:
print(f"\n • TTFT per iteration:")
for i, ttft in enumerate(metrics.ttft_per_iteration, 1):
status = "🔴 No cache" if i == 1 else "🟢 With cache"
print(f" Iteration {i}: {ttft:.3f}s {status}")
# Verify cache is working
print("\n✅ Verification:")
if metrics.cached_tokens > 0:
print(f" ✓ Cached tokens detected: {metrics.cached_tokens}")
else:
print(f" ⚠️ No cached tokens detected - cache may not be working")
if len(metrics.ttft_per_iteration) > 1:
first_ttft = metrics.ttft_per_iteration[0]
second_ttft = metrics.ttft_per_iteration[1]
if second_ttft < first_ttft * 0.8: # At least 20% improvement
print(f" ✓ TTFT improved from {first_ttft:.3f}s to {second_ttft:.3f}s")
else:
print(f" ⚠️ TTFT did not improve significantly")
print("\n💡 Note: Kimi API should return cached_tokens in the usage object")
print(" starting from the second iteration when context is stable.")
if __name__ == "__main__":
test_cached_tokens()
test_completion.py¶
#!/usr/bin/env python3
"""
Test script to verify the agent correctly identifies final answers
when no tool calls are made
"""
import os
import sys
from agent import KVCacheAgent, KVCacheMode
def test_completion_logic():
"""Test that the agent correctly handles responses without tool calls as final answers"""
# Get API key
api_key = os.getenv("MOONSHOT_API_KEY")
if not api_key:
print("❌ Please set MOONSHOT_API_KEY environment variable")
sys.exit(1)
print("🧪 Testing Final Answer Detection")
print("="*60)
# Test 1: Simple question that doesn't require tools
print("\n1️⃣ Test: Simple question without tools")
task1 = "What is 2 + 2? Just tell me the answer, no need to use any tools."
agent = KVCacheAgent(
api_key=api_key,
mode=KVCacheMode.CORRECT,
root_dir="../..",
verbose=False
)
result = agent.execute_task(task1, max_iterations=5)
print(f" Task: {task1}")
print(f" ✓ Completed in {result['iterations']} iteration(s)")
print(f" ✓ Tool calls: {len(result['tool_calls'])}")
print(f" ✓ Has final answer: {result['success']}")
if result['final_answer']:
print(f" Answer: {result['final_answer'][:100]}")
# Test 2: Question that requires tools
print("\n2️⃣ Test: Question requiring tools")
task2 = "How many Python files are in the week1/context directory?"
agent2 = KVCacheAgent(
api_key=api_key,
mode=KVCacheMode.CORRECT,
root_dir="../..",
verbose=False
)
result2 = agent2.execute_task(task2, max_iterations=5)
print(f" Task: {task2}")
print(f" ✓ Completed in {result2['iterations']} iteration(s)")
print(f" ✓ Tool calls: {len(result2['tool_calls'])}")
print(f" ✓ Has final answer: {result2['success']}")
if result2['tool_calls']:
print(" Tools used:")
for tc in result2['tool_calls']:
print(f" • {tc.name}")
# Test 3: Multi-step task
print("\n3️⃣ Test: Multi-step task")
task3 = "Find Python files in week1/context, then tell me if there's a file named 'agent.py'"
agent3 = KVCacheAgent(
api_key=api_key,
mode=KVCacheMode.CORRECT,
root_dir="../..",
verbose=False
)
result3 = agent3.execute_task(task3, max_iterations=10)
print(f" Task: {task3}")
print(f" ✓ Completed in {result3['iterations']} iteration(s)")
print(f" ✓ Tool calls: {len(result3['tool_calls'])}")
print(f" ✓ Has final answer: {result3['success']}")
# Summary
print("\n" + "="*60)
print("📊 Summary:")
print(f" • Test 1 (no tools): {result['iterations']} iterations, {len(result['tool_calls'])} tools")
print(f" • Test 2 (with tools): {result2['iterations']} iterations, {len(result2['tool_calls'])} tools")
print(f" • Test 3 (multi-step): {result3['iterations']} iterations, {len(result3['tool_calls'])} tools")
print("\n✅ The agent correctly:")
print(" 1. Identifies final answers when no tools are needed")
print(" 2. Uses tools when necessary to gather information")
print(" 3. Provides final answer after tool execution")
print("\nNo explicit 'final answer' keyword needed!")
if __name__ == "__main__":
test_completion_logic()
test_error_handling.py¶
#!/usr/bin/env python3
"""
Test script to verify error handling in tool execution
"""
import os
import sys
import json
from agent import KVCacheAgent, KVCacheMode, LocalFileTools
def test_error_handling():
"""Test that the agent continues when tools return errors"""
print("🧪 Testing Error Handling in Tool Execution")
print("="*60)
# Test local tools directly first
print("\n1️⃣ Testing direct tool error handling:")
tools = LocalFileTools(root_dir="../..")
# Test with invalid arguments
print(" Testing read_file with extra 'limit' parameter...")
# The tool should ignore the extra parameter
result = tools.read_file("week1/context/README.md")
print(f" Result: {'✓ Success' if result.get('success') else '✗ Error'}")
# Test with non-existent file
print(" Testing read_file with non-existent file...")
result = tools.read_file("non_existent_file.txt")
print(f" Result: {'✓ Error handled' if not result.get('success') else '✗ Unexpected success'}")
print(f" Error message: {result.get('error', 'N/A')}")
# Test security boundary
print(" Testing security boundary...")
result = tools.read_file("../../../../etc/passwd")
print(f" Result: {'✓ Access denied' if 'Access denied' in result.get('error', '') else '✗ Security issue'}")
# Test agent with API if key is available
api_key = os.getenv("MOONSHOT_API_KEY")
if api_key:
print("\n2️⃣ Testing agent error recovery:")
agent = KVCacheAgent(
api_key=api_key,
mode=KVCacheMode.CORRECT,
root_dir="../..",
verbose=True
)
# Task that might trigger errors
task = """Please do the following:
1. Try to read a file that doesn't exist: 'non_existent_file.txt'
2. Then find Python files in week1/context directory
3. Tell me what you found"""
print(f" Task: {task[:100]}...")
result = agent.execute_task(task, max_iterations=10)
print(f"\n ✓ Completed in {result['iterations']} iterations")
print(f" ✓ Tool calls made: {len(result['tool_calls'])}")
# Check if any tools returned errors
error_count = 0
for tc in result['tool_calls']:
if tc.result and not tc.result.get('success', True):
error_count += 1
print(f" • Tool error in {tc.name}: {tc.result.get('error', 'Unknown')[:50]}...")
print(f" ✓ Errors encountered and handled: {error_count}")
print(f" ✓ Agent continued despite errors: {result['success']}")
if result['final_answer']:
print(f"\n Final answer provided despite errors:")
print(f" {result['final_answer'][:200]}...")
else:
print("\n⚠️ Skipping agent test (no API key)")
print("\n" + "="*60)
print("✅ Error handling test complete!")
print("\nKey findings:")
print(" • Tools return errors as results instead of throwing exceptions")
print(" • Agent continues execution even when tools fail")
print(" • Unexpected arguments are filtered out safely")
print(" • Security boundaries are enforced")
if __name__ == "__main__":
test_error_handling()
test_file_range.py¶
#!/usr/bin/env python3
"""
Test script for the read_file tool with offset and size parameters
"""
from agent import LocalFileTools
def test_file_range_reading():
"""Test reading files with offset and size parameters"""
print("🧪 Testing File Range Reading")
print("="*60)
# Initialize tools
tools = LocalFileTools(root_dir="../..")
# Test file
test_file = "week1/context/agent.py"
# Test 1: Read first 10 lines
print("\n1️⃣ Reading first 10 lines:")
result = tools.read_file(test_file, offset=0, size=10)
if result["success"]:
print(f" ✓ Read {result['lines_read']} lines from total {result['total_lines']}")
print(f" Range: lines {result['offset']}-{result['end_line']}")
print(f" First line: {result['content'].split(chr(10))[0][:50]}...")
else:
print(f" ✗ Error: {result['error']}")
# Test 2: Read lines 100-110
print("\n2️⃣ Reading lines 100-110:")
result = tools.read_file(test_file, offset=100, size=10)
if result["success"]:
print(f" ✓ Read {result['lines_read']} lines")
print(f" Range: lines {result['offset']}-{result['end_line']}")
lines = result['content'].split('\n')
if lines:
print(f" Sample: {lines[0][:60]}...")
else:
print(f" ✗ Error: {result['error']}")
# Test 3: Read from offset 250 with size 500 (as specified)
print("\n3️⃣ Reading from offset 250, size 500:")
result = tools.read_file(test_file, offset=250, size=500)
if result["success"]:
print(f" ✓ Read {result['lines_read']} lines")
print(f" Range: lines {result['offset']}-{result['end_line']}")
print(f" Total file has {result['total_lines']} lines")
else:
print(f" ✗ Error: {result['error']}")
# Test 4: Read without size (from offset to end)
print("\n4️⃣ Reading from offset 700 to end:")
result = tools.read_file(test_file, offset=700)
if result["success"]:
print(f" ✓ Read {result['lines_read']} lines")
print(f" Range: lines {result['offset']}-{result['end_line']}")
else:
print(f" ✗ Error: {result['error']}")
# Test 5: Offset beyond file length
print("\n5️⃣ Testing offset beyond file length:")
result = tools.read_file(test_file, offset=10000, size=10)
if result["success"]:
print(f" ✓ Handled gracefully: {result.get('message', 'No error')}")
print(f" Lines read: {result['lines_read']}")
else:
print(f" Result: {result}")
# Test 6: Read entire file (no offset, no size)
print("\n6️⃣ Reading entire file (default behavior):")
result = tools.read_file("week1/context/README.md")
if result["success"]:
print(f" ✓ Read entire file")
print(f" Total lines: {result['total_lines']}")
print(f" Lines read: {result['lines_read']}")
print(f" Truncated: {result.get('truncated', False)}")
else:
print(f" ✗ Error: {result['error']}")
# Test 7: Compare with limit parameter (the user's original request)
print("\n7️⃣ API-style usage (offset=250, size=500):")
result = tools.read_file("week2/local_llm_serving/main.py", offset=250, size=500)
if result["success"]:
print(f" ✓ Successfully read lines {result['offset']}-{result['end_line']}")
print(f" Lines read: {result['lines_read']}")
print(f" File has {result['total_lines']} total lines")
# Show a sample of the content
lines = result['content'].split('\n')[:3]
print("\n First 3 lines of content:")
for i, line in enumerate(lines):
print(f" Line {250+i}: {line[:60]}..." if len(line) > 60 else f" Line {250+i}: {line}")
print("\n" + "="*60)
print("✅ File range reading tests complete!")
print("\nThe read_file tool now supports:")
print(" • offset: Starting line number (0-based)")
print(" • size: Number of lines to read")
print(" • Handles edge cases gracefully")
print(" • Maintains security boundaries")
if __name__ == "__main__":
test_file_range_reading()
test_interactive.py¶
#!/usr/bin/env python3
"""
Test script for interactive mode selection
"""
import sys
from main import select_mode_interactive
def test_mode_selection():
"""Test the interactive mode selection without running the agent"""
print("🧪 Testing Interactive Mode Selection")
print("(This is a test - no agent will actually run)")
# Test the selection menu
selected = select_mode_interactive()
print("\n" + "="*60)
if selected == "compare":
print("✅ You selected: Compare all modes")
print("In real usage, this would run all 6 implementations and compare them.")
else:
print(f"✅ You selected: {selected}")
print(f"In real usage, this would run the '{selected}' implementation.")
print("\nTest complete!")
if __name__ == "__main__":
test_mode_selection()
test_message_flow.py¶
#!/usr/bin/env python3
"""
Test script to verify message flow in correct vs incorrect modes
"""
def simulate_message_flow():
"""Simulate how messages are handled in different modes"""
print("🔍 Testing Message Flow Logic")
print("="*60)
# Simulate CORRECT mode
print("\n✅ CORRECT Mode:")
print("-"*40)
messages_correct = None
history_correct = []
for iteration in range(1, 4):
print(f"\nIteration {iteration}:")
if iteration == 1:
# First iteration: create messages
messages_correct = ["system", "task"]
print(f" • Created messages: {messages_correct}")
else:
print(f" • Using existing messages: {messages_correct}")
# Simulate tool call
print(f" • API returns tool call")
messages_correct.append(f"assistant_iter{iteration}")
history_correct.append(f"assistant_iter{iteration}")
# Simulate tool result
print(f" • Tool executed")
messages_correct.append(f"tool_result_iter{iteration}")
history_correct.append(f"tool_result_iter{iteration}")
print(f" • Messages now: {messages_correct}")
print(f" • History now: {history_correct}")
# Simulate INCORRECT mode
print("\n\n❌ INCORRECT Mode (e.g., DYNAMIC_SYSTEM):")
print("-"*40)
history_incorrect = []
for iteration in range(1, 4):
print(f"\nIteration {iteration}:")
# Always recreate messages from history
messages_incorrect = ["system_with_timestamp", "task"] + history_incorrect
print(f" • Recreated messages: {messages_incorrect}")
# Simulate tool call
print(f" • API returns tool call")
messages_incorrect.append(f"assistant_iter{iteration}")
history_incorrect.append(f"assistant_iter{iteration}")
# Simulate tool result
print(f" • Tool executed")
messages_incorrect.append(f"tool_result_iter{iteration}")
history_incorrect.append(f"tool_result_iter{iteration}")
print(f" • Messages now: {messages_incorrect}")
print(f" • History now: {history_incorrect}")
print("\n\n📊 Key Observations:")
print("="*60)
print("\n1. CORRECT Mode:")
print(" • Messages list persists across iterations")
print(" • Each iteration adds to the same list")
print(" • Context remains stable → KV cache works")
print("\n2. INCORRECT Mode:")
print(" • Messages list recreated each iteration")
print(" • System prompt changes (timestamp)")
print(" • Context changes → KV cache invalidated")
print("\n3. Both Modes:")
print(" • Within an iteration, tool results are appended")
print(" • This ensures the API sees complete conversation")
print(" • History is maintained for reconstruction")
if __name__ == "__main__":
simulate_message_flow()
test_tools.py¶
#!/usr/bin/env python3
"""
Test script for local file system tools
Validates that read_file, find, and grep work correctly
"""
import os
import json
from agent import LocalFileTools
def test_file_tools():
"""Test the local file system tools"""
print("🧪 Testing Local File System Tools")
print("="*60)
# Initialize tools with project root
tools = LocalFileTools(root_dir="../..")
# Test 1: Find Python files
print("\n1️⃣ Testing 'find' command...")
print(" Finding *.py files in week1/context directory...")
result = tools.find("*.py", "week1/context")
if result["success"]:
print(f" ✓ Found {result['count']} Python files")
if result["matches"]:
print(f" Sample files: {result['matches'][:3]}")
else:
print(f" ✗ Error: {result['error']}")
# Test 2: Read a file
print("\n2️⃣ Testing 'read_file' command...")
test_file = "week1/context/README.md"
print(f" Reading {test_file}...")
result = tools.read_file(test_file)
if result["success"]:
print(f" ✓ Read file successfully ({result['size']} bytes)")
print(f" First 100 chars: {result['content'][:100]}...")
else:
print(f" ✗ Error: {result['error']}")
# Test 3: Grep for a pattern
print("\n3️⃣ Testing 'grep' command...")
print(" Searching for 'agent' in week1 directory...")
result = tools.grep("agent", directory="week1")
if result["success"]:
print(f" ✓ Found {result['match_count']} matches in {result['files_searched']} files")
if result["matches"]:
sample = result["matches"][0]
print(f" Sample match: {sample['file']}:{sample['line_num']} - {sample['line'][:50]}...")
else:
print(f" ✗ Error: {result['error']}")
# Test 4: Security check - try to access outside root
print("\n4️⃣ Testing security boundaries...")
print(" Attempting to read file outside root directory...")
result = tools.read_file("../../../../../../etc/passwd")
if not result["success"] and "Access denied" in result.get("error", ""):
print(" ✓ Security check passed - access denied as expected")
else:
print(" ⚠️ Security check result:", result.get("error", "Unexpected result"))
# Test 5: Grep in specific file
print("\n5️⃣ Testing 'grep' in specific file...")
print(" Searching for 'class' in week1/context/agent.py...")
result = tools.grep("class", file_path="week1/context/agent.py")
if result["success"]:
print(f" ✓ Found {result['match_count']} matches")
if result["matches"]:
for match in result["matches"][:3]:
print(f" Line {match['line_num']}: {match['line'][:60]}...")
else:
print(f" ✗ Error: {result['error']}")
print("\n" + "="*60)
print("✅ Tool testing complete!")
print("\nAll tools are working correctly and can be used by the ReAct agent.")
print("Security boundaries are properly enforced.")
def test_pattern_matching():
"""Test various pattern matching scenarios"""
print("\n🔍 Testing Pattern Matching Capabilities")
print("="*60)
tools = LocalFileTools(root_dir="../..")
# Test different file patterns
patterns = [
("*.md", "week1", "Markdown files"),
("*.py", "week2", "Python files"),
("README*", ".", "README files"),
("test_*.py", "week1", "Test files"),
]
for pattern, directory, description in patterns:
print(f"\n• Finding {description}: {pattern} in {directory}")
result = tools.find(pattern, directory)
if result["success"]:
print(f" Found {result['count']} files")
else:
print(f" Error: {result['error']}")
# Test different grep patterns
print("\n📝 Testing Grep Patterns")
print("-"*40)
grep_tests = [
(r"def \w+\(", "week1/context/agent.py", "Function definitions"),
(r"import \w+", "week1/context/main.py", "Import statements"),
(r"TODO|FIXME", "week1", "TODO/FIXME comments"),
(r"^\s*class", "week1/context/agent.py", "Class definitions"),
]
for pattern, target, description in grep_tests:
print(f"\n• Searching for {description}: {pattern}")
if "/" in target:
result = tools.grep(pattern, file_path=target)
else:
result = tools.grep(pattern, directory=target)
if result["success"]:
print(f" Found {result['match_count']} matches")
else:
print(f" Error: {result['error']}")
if __name__ == "__main__":
# Run basic tests
test_file_tools()
# Run pattern matching tests
test_pattern_matching()
print("\n🎉 All tests completed successfully!")
test_ttft.py¶
#!/usr/bin/env python3
"""
Test script to demonstrate TTFT tracking across iterations
Shows how cache usage improves response times
"""
import os
import sys
from agent import KVCacheAgent, KVCacheMode
def test_ttft_tracking():
"""Test and display TTFT tracking across iterations"""
# Get API key
api_key = os.getenv("MOONSHOT_API_KEY")
if not api_key:
print("❌ Please set MOONSHOT_API_KEY environment variable")
sys.exit(1)
print("📊 TTFT Tracking Demonstration")
print("="*60)
# Task that requires multiple iterations
task = """Analyze the week1/context directory:
1. Find all Python files
2. Read the agent.py file (first 100 lines)
3. Search for classes in the code
4. Provide a summary of what you found"""
print(f"Task: {task[:100]}...")
print("="*60)
# Test with correct implementation (should show cache benefits)
print("\n✅ CORRECT Implementation (with KV cache):")
print("-"*40)
agent = KVCacheAgent(
api_key=api_key,
mode=KVCacheMode.CORRECT,
root_dir="../..",
verbose=False # Set to True to see detailed logs
)
result = agent.execute_task(task, max_iterations=10)
metrics = result["metrics"]
# Display TTFT progression
print(f"Iterations completed: {result['iterations']}")
print(f"Tool calls made: {len(result['tool_calls'])}")
print(f"\nTTFT per iteration:")
for i, ttft in enumerate(metrics.ttft_per_iteration, 1):
bar_length = int(ttft * 10) # Visual bar representation
bar = "█" * min(bar_length, 50)
print(f" Iter {i:2d}: {ttft:6.3f}s {bar}")
# Calculate statistics
if len(metrics.ttft_per_iteration) > 1:
first = metrics.ttft_per_iteration[0]
last = metrics.ttft_per_iteration[-1]
avg_all = sum(metrics.ttft_per_iteration) / len(metrics.ttft_per_iteration)
avg_after_first = sum(metrics.ttft_per_iteration[1:]) / len(metrics.ttft_per_iteration[1:])
print(f"\n📈 Performance Analysis:")
print(f" • First iteration: {first:.3f}s (cold start)")
print(f" • Last iteration: {last:.3f}s")
print(f" • Average (all): {avg_all:.3f}s")
print(f" • Average (cached): {avg_after_first:.3f}s")
print(f" • Speed improvement: {(first - last) / first * 100:.1f}%")
print(f" • Cached tokens: {metrics.cached_tokens:,}")
# Compare with dynamic system prompt (no cache benefits)
print("\n" + "="*60)
print("❌ DYNAMIC SYSTEM Implementation (breaks KV cache):")
print("-"*40)
agent2 = KVCacheAgent(
api_key=api_key,
mode=KVCacheMode.DYNAMIC_SYSTEM,
root_dir="../..",
verbose=False
)
result2 = agent2.execute_task(task, max_iterations=10)
metrics2 = result2["metrics"]
print(f"Iterations completed: {result2['iterations']}")
print(f"Tool calls made: {len(result2['tool_calls'])}")
print(f"\nTTFT per iteration:")
for i, ttft in enumerate(metrics2.ttft_per_iteration, 1):
bar_length = int(ttft * 10)
bar = "█" * min(bar_length, 50)
print(f" Iter {i:2d}: {ttft:6.3f}s {bar}")
if len(metrics2.ttft_per_iteration) > 1:
first2 = metrics2.ttft_per_iteration[0]
last2 = metrics2.ttft_per_iteration[-1]
avg_all2 = sum(metrics2.ttft_per_iteration) / len(metrics2.ttft_per_iteration)
print(f"\n📉 Performance Analysis:")
print(f" • First iteration: {first2:.3f}s")
print(f" • Last iteration: {last2:.3f}s")
print(f" • Average (all): {avg_all2:.3f}s")
print(f" • Speed improvement: {(first2 - last2) / first2 * 100:.1f}% (minimal)")
print(f" • Cached tokens: {metrics2.cached_tokens:,} (should be 0)")
# Comparison
print("\n" + "="*60)
print("🔬 COMPARISON:")
print("-"*40)
if metrics.ttft_per_iteration and metrics2.ttft_per_iteration:
avg1 = sum(metrics.ttft_per_iteration) / len(metrics.ttft_per_iteration)
avg2 = sum(metrics2.ttft_per_iteration) / len(metrics2.ttft_per_iteration)
print(f"Average TTFT:")
print(f" • Correct (with cache): {avg1:.3f}s")
print(f" • Dynamic (no cache): {avg2:.3f}s")
print(f" • Difference: {avg2 - avg1:.3f}s slower without cache")
print(f" • Performance penalty: {(avg2 - avg1) / avg1 * 100:.1f}% slower")
print(f"\nCache Usage:")
print(f" • Correct: {metrics.cached_tokens:,} tokens cached")
print(f" • Dynamic: {metrics2.cached_tokens:,} tokens cached")
print("\n💡 Key Observation:")
print("The correct implementation shows significant TTFT improvement after the")
print("first iteration due to KV cache, while dynamic system prompt maintains")
print("consistently high TTFT because the cache is invalidated on each request.")
if __name__ == "__main__":
test_ttft_tracking()
result_correct_20260718_kimi_k2_6.json¶
{
"success": true,
"iterations": 3,
"tool_calls": [
"ToolCall(name='find', arguments={'pattern': '*.py'}, result={'pattern': '*.py', 'directory': '.', 'matches': ['agent.py', 'demo_quick.py', 'main.py', 'test_api.py', 'test_cache_invalidation.py', 'test_cached_tokens.py', 'test_completion.py', 'test_error_handling.py', 'test_file_range.py', 'test_interactive.py', 'test_message_flow.py', 'test_tools.py', 'test_ttft.py'], 'count': 13, 'truncated': False, 'success': True}, error=None, timestamp=1784337508.700185)",
"ToolCall(name='read_file', arguments={'file_path': 'main.py'}, result={'path': 'main.py', 'content': '\"\"\"\\nMain script to demonstrate KV cache importance\\nRuns the ReAct agent with different implementations and compares performance\\n\"\"\"\\n\\nimport os\\nimport sys\\nimport glob\\nimport json\\nimport argparse\\nimport logging\\nfrom typing import Dict, List, Any\\nfrom datetime import datetime\\nfrom dataclasses import asdict\\nfrom agent import KVCacheAgent, KVCacheMode, AgentMetrics, compare_implementations\\n\\n# Default model (Moonshot / Kimi). The whole current Kimi family (k2.5/k2.6/\\n# k2.7/k3) reports cached_tokens for automatic prefix caching AND reasons, so it\\n# only accepts temperature=1 (agent.py handles that automatically). kimi-k2.6 has\\n# the lightest reasoning footprint of the cache-reporting models, giving the\\n# cleanest TTFT while still exposing the prefix-cache hit metric this demo needs.\\n# (The non-reasoning moonshot-v1-* models do NOT report cached_tokens, so they\\n# cannot demonstrate the cache effect.)\\nDEFAULT_MODEL = \"kimi-k2.6\"\\n\\n# Configure logging\\nlogging.basicConfig(\\n level=logging.INFO,\\n format=\\'%(asctime)s - %(levelname)s - %(message)s\\',\\n handlers=[\\n logging.FileHandler(\\'kv_cache_demo.log\\'),\\n logging.StreamHandler()\\n ]\\n)\\nlogger = logging.getLogger(__name__)\\n\\n\\n# ---------------------------------------------------------------------------\\n# Metrics helpers (shared by live comparison and offline report)\\n# ---------------------------------------------------------------------------\\n\\ndef _coerce_metrics(metrics: Any) -> Dict[str, Any]:\\n \"\"\"Normalize a stored metrics value into a plain dict.\\n\\n Handles both formats found in result files:\\n - dict: produced by --compare (asdict) and by the fixed --mode path\\n - str : legacy single-mode files that stored repr(AgentMetrics(...))\\n because json.dump used default=str\\n \"\"\"\\n if isinstance(metrics, dict):\\n return metrics\\n if isinstance(metrics, str) and metrics.startswith(\"AgentMetrics(\"):\\n # Safe eval: only AgentMetrics is exposed, no builtins.\\n try:\\n obj = eval(metrics, {\"__builtins__\": {}}, {\"AgentMetrics\": AgentMetrics})\\n return asdict(obj)\\n except Exception as e: # pragma: no cover - defensive\\n logger.warning(f\"Could not parse legacy metrics string: {e}\")\\n return {}\\n\\n\\ndef _avg_ttft(m: Dict[str, Any]) -> float:\\n \"\"\"Average TTFT across iterations, falling back to first-iteration TTFT.\"\"\"\\n lst = m.get(\"ttft_per_iteration\") or []\\n return sum(lst) / len(lst) if lst else float(m.get(\"ttft\", 0.0) or 0.0)\\n\\n\\ndef _hit_rate(m: Dict[str, Any]) -> float:\\n total = (m.get(\"cache_hits\", 0) or 0) + (m.get(\"cache_misses\", 0) or 0)\\n return (m.get(\"cache_hits\", 0) or 0) / total * 100 if total else 0.0\\n\\n\\ndef _billable_tokens(m: Dict[str, Any], cache_price_ratio: float) -> float:\\n \"\"\"Illustrative billable prompt tokens under a prompt-cache discount.\\n\\n cached tokens are charged at cache_price_ratio of the normal price; the\\n rest at full price. This is a transparent function of the *measured*\\n token counts and a user-supplied ratio - it is not a fabricated\\n provider-specific price.\\n \"\"\"\\n prompt = m.get(\"prompt_tokens\", 0) or 0\\n cached = m.get(\"cached_tokens\", 0) or 0\\n cached = min(cached, prompt)\\n return (prompt - cached) + cached * cache_price_ratio\\n\\n\\ndef print_comparison_table(results: Dict[str, Any], cache_price_ratio: float = 0.1) -> None:\\n \"\"\"Render the cross-strategy comparison table (latency / cache / cost).\"\"\"\\n print(f\"\\\\n{\\'Mode\\':<16} {\\'Iters\\':<6} {\\'1st TTFT\\':<10} {\\'Avg TTFT\\':<10} \"\\n f\"{\\'Total(s)\\':<10} {\\'Prompt\\':<9} {\\'Cached\\':<9} {\\'Hit%\\':<7} \"\\n f\"{\\'Cache%\\':<8} {\\'Bill.Tok\\':<10} {\\'Save%\\':<7}\")\\n print(\"-\" * 112)\\n\\n for mode, data in results.items():\\n m = _coerce_metrics(data.get(\"metrics\", {}))\\n prompt = m.get(\"prompt_tokens\", 0) or 0\\n cached = m.get(\"cached_tokens\", 0) or 0\\n iters = data.get(\"iterations\", m.get(\"iterations\", 0)) or 0\\n cache_pct = cached / prompt * 100 if prompt else 0.0\\n billable = _billable_tokens(m, cache_price_ratio)\\n save_pct = (prompt - billable) / prompt * 100 if prompt else 0.0\\n\\n print(f\"{mode:<16} {iters:<6} {float(m.get(\\'ttft\\', 0.0) or 0.0):<10.3f} \"\\n f\"{_avg_ttft(m):<10.3f} {float(m.get(\\'total_time\\', 0.0) or 0.0):<10.3f} \"\\n f\"{prompt:<9,} {cached:<9,} {_hit_rate(m):<7.1f} \"\\n f\"{cache_pct:<8.1f} {billable:<10,.0f} {save_pct:<7.1f}\")\\n\\n print(\"-\" * 112)\\n print(f\"\u6ce8\uff1aBill.Tok / Save% \u5047\u8bbe\u7f13\u5b58 token \u6309\u6b63\u5e38\u4ef7\u7684 {cache_price_ratio:.0%} \u8ba1\u8d39\"\\n f\"\uff08\u53ef\u7528 --cache-price-ratio \u8c03\u6574\uff09\uff0c\u4ec5\u4e3a\u6210\u672c\u793a\u610f\uff0c\u975e\u67d0\u5bb6\u670d\u52a1\u5546\u5b9e\u9645\u62a5\u4ef7\u3002\")\\n\\n\\ndef load_result_files(paths: List[str]) -> Dict[str, Any]:\\n \"\"\"Load result_*.json files into a {mode: {...}} dict for offline reporting.\"\"\"\\n results: Dict[str, Any] = {}\\n for path in sorted(paths):\\n try:\\n with open(path, \\'r\\') as f:\\n data = json.load(f)\\n except Exception as e:\\n logger.warning(f\"Skipping {path}: {e}\")\\n continue\\n\\n # A comparison_*.json holds many modes; a result_*.json holds one.\\n if \"mode\" not in data and all(isinstance(v, dict) and \"metrics\" in v\\n for v in data.values()):\\n for mode, entry in data.items():\\n results[mode] = {\"metrics\": _coerce_metrics(entry.get(\"metrics\", {})),\\n \"iterations\": entry.get(\"iterations\"),\\n \"_source\": path}\\n else:\\n mode = data.get(\"mode\", os.path.splitext(os.path.basename(path))[0])\\n results[mode] = {\"metrics\": _coerce_metrics(data.get(\"metrics\", {})),\\n \"iterations\": data.get(\"iterations\"),\\n \"_source\": path}\\n return results\\n\\n\\ndef run_report(inputs: List[str] = None, cache_price_ratio: float = 0.1) -> None:\\n \"\"\"Offline: build the comparison table from existing result_*.json files.\\n\\n No API key required - reads previously saved runs so the final result is\\n legible in one command without re-hitting the model.\\n \"\"\"\\n if not inputs:\\n inputs = [\"result_*.json\", \"comparison_*.json\"]\\n\\n paths: List[str] = []\\n for item in inputs:\\n if os.path.isdir(item):\\n paths.extend(glob.glob(os.path.join(item, \"result_*.json\")))\\n paths.extend(glob.glob(os.path.join(item, \"comparison_*.json\")))\\n else:\\n paths.extend(glob.glob(item))\\n\\n paths = sorted(set(paths))\\n if not paths:\\n logger.error(\"\u672a\u627e\u5230\u4efb\u4f55 result_*.json / comparison_*.json \u7ed3\u679c\u6587\u4ef6\u3002\"\\n \"\u8bf7\u5148\u8fd0\u884c --mode \u6216 --compare \u751f\u6210\u7ed3\u679c\uff0c\u6216\u7528 --input \u6307\u5b9a\u8def\u5f84\u3002\")\\n sys.exit(1)\\n\\n results = load_result_files(paths)\\n\\n print(\"\\\\n\" + \"=\" * 112)\\n print(\"KV CACHE \u79bb\u7ebf\u5bf9\u6bd4\u62a5\u544a\uff08\u57fa\u4e8e\u5df2\u4fdd\u5b58\u7684\u5b9e\u6d4b\u7ed3\u679c\uff09\")\\n print(\"=\" * 112)\\n print(f\"\u6570\u636e\u6765\u6e90\uff08{len(paths)} \u4e2a\u6587\u4ef6\uff09:\")\\n for mode, data in results.items():\\n print(f\" \u2022 {mode:<16} \u2190 {os.path.basename(data.get(\\'_source\\', \\'?\\'))}\")\\n\\n print_comparison_table(results, cache_price_ratio)\\n\\n print(\"\\\\n\ud83d\udcdd \u8bf4\u660e\uff1a\u4e0d\u540c\u7ed3\u679c\u6587\u4ef6\u53ef\u80fd\u6765\u81ea\u4e0d\u540c\u4efb\u52a1/\u65f6\u95f4\uff0c\u7edd\u5bf9\u6570\u503c\u4ec5\u4f9b\u540c\u4e00\u6b21\u8fd0\u884c\u5185\u6a2a\u5411\u5bf9\u6bd4\uff1b\"\\n \"\u5982\u9700\u4e25\u683c\u5bf9\u7167\uff0c\u8bf7\u7528 --compare \u5728\u540c\u4e00\u4efb\u52a1\u4e0b\u4e00\u6b21\u6027\u751f\u6210\u5168\u90e8\u6a21\u5f0f\u7684\u6570\u636e\u3002\")\\n\\n\\ndef create_summary_task() -> str:\\n \"\"\"Create a task that requires reading multiple files\"\"\"\\n return \"\"\"Please analyze and summarize all the projects in the week1 and week2 directories.\\nFor each project:\\n1. Find all Python files\\n2. Read the main files and understand the functionality\\n3. Identify the key features and purpose\\n4. Provide a comprehensive summary\\n\\nStart with week1 projects, then move to week2. Be thorough in your analysis.\"\"\"\\n\\n\\ndef run_single_mode(api_key: str, mode: str, task: str = None, root_dir: str = \"../..\",\\n model: str = DEFAULT_MODEL, output: str = None):\\n \"\"\"\\n Run agent in a single mode\\n\\n Args:\\n api_key: API key for Kimi\\n mode: KV cache mode to use\\n task: Custom task (optional)\\n root_dir: Root directory for file operations (default: \"../..\" = /projects from kv-cache dir)\\n model: Model to use\\n output: Output path for the result JSON (optional; auto-named if omitted)\\n \"\"\"\\n # Parse mode\\n mode_map = {\\n \"correct\": KVCacheMode.CORRECT,\\n \"dynamic_system\": KVCacheMode.DYNAMIC_SYSTEM,\\n \"shuffled_tools\": KVCacheMode.SHUFFLED_TOOLS,\\n \"dynamic_profile\": KVCacheMode.DYNAMIC_PROFILE,\\n \"sliding_window\": KVCacheMode.SLIDING_WINDOW,\\n \"text_format\": KVCacheMode.TEXT_FORMAT\\n }\\n \\n if mode not in mode_map:\\n logger.error(f\"Invalid mode: {mode}\")\\n logger.info(f\"Valid modes: {\\', \\'.join(mode_map.keys())}\")\\n return\\n \\n # Use default task if not provided\\n if not task:\\n task = create_summary_task()\\n \\n logger.info(f\"Running in mode: {mode}\")\\n logger.info(f\"Task: {task}\")\\n logger.info(\"=\"*80)\\n \\n # Create agent and execute task\\n agent = KVCacheAgent(\\n api_key=api_key,\\n mode=mode_map[mode],\\n model=model,\\n root_dir=root_dir,\\n verbose=True\\n )\\n \\n result = agent.execute_task(task, max_iterations=30)\\n \\n # Print results\\n print(\"\\\\n\" + \"=\"*80)\\n print(f\"EXECUTION RESULTS - Mode: {mode}\")\\n print(\"=\"*80)\\n \\n metrics = result[\"metrics\"]\\n print(f\"\\\\n\ud83d\udcca Performance Metrics:\")\\n print(f\" \u2022 Time to First Token (TTFT): {metrics.ttft:.3f} seconds\")\\n \\n # Show TTFT progression\\n if metrics.ttft_per_iteration:\\n print(f\" \u2022 TTFT per iteration:\")\\n for i, ttft in enumerate(metrics.ttft_per_iteration, 1):\\n print(f\" Iteration {i}: {ttft:.3f}s\")\\n\\n # Show improvement\\n if len(metrics.ttft_per_iteration) > 1:\\n first_ttft = metrics.ttft_per_iteration[0]\\n last_ttft = metrics.ttft_per_iteration[-1]\\n avg_after_first = sum(metrics.ttft_per_iteration[1:]) / len(metrics.ttft_per_iteration[1:])\\n print(f\" \u2022 TTFT Analysis:\")\\n print(f\" First iteration: {first_ttft:.3f}s\")\\n print(f\" Last iteration: {', 'total_lines': 537, 'lines_read': 537, 'offset': 0, 'end_line': 537, 'truncated': True, 'success': True}, error=None, timestamp=1784337510.6666272)",
"ToolCall(name='read_file', arguments={'file_path': 'agent.py'}, result={'path': 'agent.py', 'content': '\"\"\"\\nKV Cache Demonstration Agent with ReAct Pattern\\nDemonstrates the importance of KV cache through correct and incorrect implementations.\\nUses local file system tools to read and search through code files.\\n\"\"\"\\n\\nimport json\\nimport os\\nimport re\\nimport time\\nimport logging\\nimport random\\nfrom typing import List, Dict, Any, Optional, Tuple\\nfrom dataclasses import dataclass, field, asdict\\nfrom enum import Enum\\nfrom datetime import datetime\\nfrom openai import OpenAI\\nimport glob as glob_module\\nimport subprocess\\n\\n\\ndef _is_reasoning_model(model) -> bool:\\n \"\"\"True for models that emit reasoning_content and only accept temperature=1.\\n\\n On the live Moonshot endpoint the whole current Kimi family reasons:\\n kimi-k2.5 / kimi-k2.6 / kimi-k2.7* / kimi-k3. The legacy moonshot-v1-*\\n chat models do NOT reason (and also do not report cached_tokens).\"\"\"\\n m = str(model or \"\").lower().replace(\"/\", \"-\")\\n if \"gpt-5\" in m:\\n return True\\n return any(tag in m for tag in (\"kimi-k2.5\", \"kimi-k2.6\", \"kimi-k2.7\", \"kimi-k3\"))\\n\\n\\ndef _reasoning_safe_temperature(model, requested=1.0):\\n \"\"\"Reasoning models (Kimi K2.5/K2.6/K2.7/K3, GPT-5, ...) only accept\\n temperature=1. Return 1 for those; otherwise the requested value so\\n non-reasoning providers (moonshot-v1, Doubao, DeepSeek) are unchanged.\"\"\"\\n return 1 if _is_reasoning_model(model) else requested\\n\\n\\ndef _reasoning_safe_max_tokens(model, requested=2000):\\n \"\"\"Reasoning models spend completion budget on hidden reasoning tokens\\n before emitting content / tool calls. Give them enough headroom so a\\n tool call is not truncated away; leave non-reasoning models unchanged.\"\"\"\\n return max(requested, 4096) if _is_reasoning_model(model) else requested\\n\\n\\n# Configure logging\\nlogging.basicConfig(level=logging.INFO, format=\\'%(asctime)s - %(levelname)s - %(message)s\\')\\nlogger = logging.getLogger(__name__)\\n\\n\\nclass KVCacheMode(Enum):\\n \"\"\"Different KV cache optimization modes\"\"\"\\n CORRECT = \"correct\" # Correct implementation with stable context\\n DYNAMIC_SYSTEM = \"dynamic_system\" # Changing system prompt with timestamp\\n SHUFFLED_TOOLS = \"shuffled_tools\" # Shuffling tool order each request\\n DYNAMIC_PROFILE = \"dynamic_profile\" # Changing user profile with credits\\n SLIDING_WINDOW = \"sliding_window\" # Only keeping recent 5 messages\\n TEXT_FORMAT = \"text_format\" # Formatting messages as plain text\\n\\n\\n@dataclass\\nclass ToolCall:\\n \"\"\"Represents a single tool call\"\"\"\\n name: str\\n arguments: Dict[str, Any]\\n result: Any = None\\n error: Optional[str] = None\\n timestamp: float = field(default_factory=time.time)\\n\\n\\n@dataclass\\nclass AgentMetrics:\\n \"\"\"Metrics for agent performance\"\"\"\\n ttft: float = 0.0 # Time to first token (first iteration)\\n ttft_per_iteration: List[float] = field(default_factory=list) # TTFT for each iteration\\n total_time: float = 0.0\\n iterations: int = 0\\n tool_calls: int = 0\\n cache_hits: int = 0\\n cache_misses: int = 0\\n prompt_tokens: int = 0\\n completion_tokens: int = 0\\n cached_tokens: int = 0\\n\\n\\nclass LocalFileTools:\\n \"\"\"Local implementations of file system tools\"\"\"\\n \\n def __init__(self, root_dir: str = \".\"):\\n self.root_dir = os.path.abspath(root_dir)\\n logger.info(f\"File tools initialized with root: {self.root_dir}\")\\n \\n def read_file(self, file_path: str, offset: int = 0, size: int = None) -> Dict[str, Any]:\\n \"\"\"\\n Read contents of a file\\n \\n Args:\\n file_path: Path to the file relative to root directory\\n offset: Line number to start reading from (0-based, default: 0)\\n size: Number of lines to read (default: None, read all)\\n \\n Returns:\\n Dictionary with file contents or error\\n \"\"\"\\n try:\\n full_path = os.path.join(self.root_dir, file_path)\\n \\n # Security check - ensure path is within root_dir\\n real_path = os.path.realpath(full_path)\\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n with open(real_path, \\'r\\', encoding=\\'utf-8\\', errors=\\'ignore\\') as f:\\n lines = f.readlines()\\n \\n total_lines = len(lines)\\n \\n # Apply offset and size\\n if offset < 0:\\n offset = 0\\n if offset >= total_lines:\\n return {\\n \"path\": file_path,\\n \"content\": \"\",\\n \"total_lines\": total_lines,\\n \"lines_read\": 0,\\n \"offset\": offset,\\n \"success\": True,\\n \"message\": f\"Offset {offset} exceeds file length ({total_lines} lines)\"\\n }\\n \\n # Determine end line\\n if size is None:\\n end = total_lines\\n else:\\n end = min(offset + size, total_lines)\\n \\n # Get the requested lines\\n selected_lines = lines[offset:end]\\n content = \\'\\'.join(selected_lines)\\n \\n # Apply size limit for safety (10KB)\\n truncated = False\\n if len(content) > 10000:\\n content = content[:10000]\\n truncated = True\\n \\n return {\\n \"path\": file_path,\\n \"content\": content,\\n \"total_lines\": total_lines,\\n \"lines_read\": len(selected_lines),\\n \"offset\": offset,\\n \"end_line\": end,\\n \"truncated\": truncated,\\n \"success\": True\\n }\\n except FileNotFoundError:\\n return {\\n \"error\": f\"File not found: {file_path}\",\\n \"success\": False\\n }\\n except Exception as e:\\n return {\\n \"error\": f\"Error reading file: {str(e)}\",\\n \"success\": False\\n }\\n \\n def find(self, pattern: str = \"*\", directory: str = \".\") -> Dict[str, Any]:\\n \"\"\"\\n Find files matching a pattern (similar to Unix find command)\\n \\n Args:\\n pattern: File name pattern (supports wildcards, default: \"*\" for all files)\\n directory: Directory to search in (relative to root_dir)\\n \\n Returns:\\n Dictionary with list of matching files\\n \"\"\"\\n try:\\n # Handle directory path properly\\n if directory == \".\":\\n search_dir = self.root_dir\\n else:\\n # Remove leading/trailing slashes for consistency\\n directory = directory.strip(\\'/\\')\\n search_dir = os.path.join(self.root_dir, directory)\\n \\n # Security check\\n real_path = os.path.realpath(search_dir)\\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n # Check if directory exists\\n if not os.path.exists(real_path):\\n return {\\n \"error\": f\"Directory not found: {directory}\",\\n \"success\": False\\n }\\n \\n # Use glob to find matching files\\n matches = []\\n for root, dirs, files in os.walk(real_path):\\n # Filter hidden directories and __pycache__\\n dirs[:] = [d for d in dirs if not d.startswith(\\'.\\') and d != \\'__pycache__\\']\\n \\n for file in files:\\n # Skip hidden files and .pyc files\\n if file.startswith(\\'.\\') or file.endswith(\\'.pyc\\'):\\n continue\\n \\n if glob_module.fnmatch.fnmatch(file, pattern):\\n # Get path relative to root_dir (not search_dir)\\n full_path = os.path.join(root, file)\\n rel_path = os.path.relpath(full_path, self.root_dir)\\n matches.append(rel_path)\\n \\n # Sort for consistency\\n matches.sort()\\n \\n # Limit results for demonstration\\n if len(matches) > 100:\\n matches = matches[:100]\\n truncated = True\\n else:\\n truncated = False\\n \\n return {\\n \"pattern\": pattern,\\n \"directory\": directory,\\n \"matches\": matches,\\n \"count\": len(matches),\\n \"truncated\": truncated,\\n \"success\": True\\n }\\n except Exception as e:\\n return {\\n \"error\": f\"Error finding files: {str(e)}\",\\n \"success\": False\\n }\\n \\n def grep(self, pattern: str, file_path: str = None, directory: str = None) -> Dict[str, Any]:\\n \"\"\"\\n Search for pattern in files (similar to Unix grep command)\\n \\n Args:\\n pattern: Regular expression pattern to search for\\n file_path: Single file to search in (optional)\\n directory: Directory to search in (optional)\\n \\n Returns:\\n Dictionary with matching lines\\n \"\"\"\\n try:\\n matches = []\\n files_searched = []\\n \\n if file_path:\\n # Search in single file\\n full_path = os.path.join(self.root_dir, file_path)\\n real_path = os.path.realpath(full_path)\\n \\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n ', 'total_lines': 869, 'lines_read': 869, 'offset': 0, 'end_line': 869, 'truncated': True, 'success': True}, error=None, timestamp=1784337510.6668859)"
],
"metrics": {
"ttft": 2.3279571533203125,
"ttft_per_iteration": [
2.3279571533203125,
1.966052770614624,
13.868276834487915
],
"total_time": 18.16346788406372,
"iterations": 3,
"tool_calls": 3,
"cache_hits": 2,
"cache_misses": 0,
"prompt_tokens": 7567,
"completion_tokens": 620,
"cached_tokens": 768
},
"mode": "correct",
"model": "kimi-k2.6",
"task": "Find all Python files in this directory; read main.py and agent.py; summarize in 3 sentences."
}
result_dynamic_profile_20260718_kimi_k2_6.json¶
{
"success": true,
"iterations": 3,
"tool_calls": [
"ToolCall(name='find', arguments={'pattern': '*.py'}, result={'pattern': '*.py', 'directory': '.', 'matches': ['agent.py', 'demo_quick.py', 'main.py', 'test_api.py', 'test_cache_invalidation.py', 'test_cached_tokens.py', 'test_completion.py', 'test_error_handling.py', 'test_file_range.py', 'test_interactive.py', 'test_message_flow.py', 'test_tools.py', 'test_ttft.py'], 'count': 13, 'truncated': False, 'success': True}, error=None, timestamp=1784337584.4136388)",
"ToolCall(name='read_file', arguments={'file_path': 'main.py'}, result={'path': 'main.py', 'content': '\"\"\"\\nMain script to demonstrate KV cache importance\\nRuns the ReAct agent with different implementations and compares performance\\n\"\"\"\\n\\nimport os\\nimport sys\\nimport glob\\nimport json\\nimport argparse\\nimport logging\\nfrom typing import Dict, List, Any\\nfrom datetime import datetime\\nfrom dataclasses import asdict\\nfrom agent import KVCacheAgent, KVCacheMode, AgentMetrics, compare_implementations\\n\\n# Default model (Moonshot / Kimi). The whole current Kimi family (k2.5/k2.6/\\n# k2.7/k3) reports cached_tokens for automatic prefix caching AND reasons, so it\\n# only accepts temperature=1 (agent.py handles that automatically). kimi-k2.6 has\\n# the lightest reasoning footprint of the cache-reporting models, giving the\\n# cleanest TTFT while still exposing the prefix-cache hit metric this demo needs.\\n# (The non-reasoning moonshot-v1-* models do NOT report cached_tokens, so they\\n# cannot demonstrate the cache effect.)\\nDEFAULT_MODEL = \"kimi-k2.6\"\\n\\n# Configure logging\\nlogging.basicConfig(\\n level=logging.INFO,\\n format=\\'%(asctime)s - %(levelname)s - %(message)s\\',\\n handlers=[\\n logging.FileHandler(\\'kv_cache_demo.log\\'),\\n logging.StreamHandler()\\n ]\\n)\\nlogger = logging.getLogger(__name__)\\n\\n\\n# ---------------------------------------------------------------------------\\n# Metrics helpers (shared by live comparison and offline report)\\n# ---------------------------------------------------------------------------\\n\\ndef _coerce_metrics(metrics: Any) -> Dict[str, Any]:\\n \"\"\"Normalize a stored metrics value into a plain dict.\\n\\n Handles both formats found in result files:\\n - dict: produced by --compare (asdict) and by the fixed --mode path\\n - str : legacy single-mode files that stored repr(AgentMetrics(...))\\n because json.dump used default=str\\n \"\"\"\\n if isinstance(metrics, dict):\\n return metrics\\n if isinstance(metrics, str) and metrics.startswith(\"AgentMetrics(\"):\\n # Safe eval: only AgentMetrics is exposed, no builtins.\\n try:\\n obj = eval(metrics, {\"__builtins__\": {}}, {\"AgentMetrics\": AgentMetrics})\\n return asdict(obj)\\n except Exception as e: # pragma: no cover - defensive\\n logger.warning(f\"Could not parse legacy metrics string: {e}\")\\n return {}\\n\\n\\ndef _avg_ttft(m: Dict[str, Any]) -> float:\\n \"\"\"Average TTFT across iterations, falling back to first-iteration TTFT.\"\"\"\\n lst = m.get(\"ttft_per_iteration\") or []\\n return sum(lst) / len(lst) if lst else float(m.get(\"ttft\", 0.0) or 0.0)\\n\\n\\ndef _hit_rate(m: Dict[str, Any]) -> float:\\n total = (m.get(\"cache_hits\", 0) or 0) + (m.get(\"cache_misses\", 0) or 0)\\n return (m.get(\"cache_hits\", 0) or 0) / total * 100 if total else 0.0\\n\\n\\ndef _billable_tokens(m: Dict[str, Any], cache_price_ratio: float) -> float:\\n \"\"\"Illustrative billable prompt tokens under a prompt-cache discount.\\n\\n cached tokens are charged at cache_price_ratio of the normal price; the\\n rest at full price. This is a transparent function of the *measured*\\n token counts and a user-supplied ratio - it is not a fabricated\\n provider-specific price.\\n \"\"\"\\n prompt = m.get(\"prompt_tokens\", 0) or 0\\n cached = m.get(\"cached_tokens\", 0) or 0\\n cached = min(cached, prompt)\\n return (prompt - cached) + cached * cache_price_ratio\\n\\n\\ndef print_comparison_table(results: Dict[str, Any], cache_price_ratio: float = 0.1) -> None:\\n \"\"\"Render the cross-strategy comparison table (latency / cache / cost).\"\"\"\\n print(f\"\\\\n{\\'Mode\\':<16} {\\'Iters\\':<6} {\\'1st TTFT\\':<10} {\\'Avg TTFT\\':<10} \"\\n f\"{\\'Total(s)\\':<10} {\\'Prompt\\':<9} {\\'Cached\\':<9} {\\'Hit%\\':<7} \"\\n f\"{\\'Cache%\\':<8} {\\'Bill.Tok\\':<10} {\\'Save%\\':<7}\")\\n print(\"-\" * 112)\\n\\n for mode, data in results.items():\\n m = _coerce_metrics(data.get(\"metrics\", {}))\\n prompt = m.get(\"prompt_tokens\", 0) or 0\\n cached = m.get(\"cached_tokens\", 0) or 0\\n iters = data.get(\"iterations\", m.get(\"iterations\", 0)) or 0\\n cache_pct = cached / prompt * 100 if prompt else 0.0\\n billable = _billable_tokens(m, cache_price_ratio)\\n save_pct = (prompt - billable) / prompt * 100 if prompt else 0.0\\n\\n print(f\"{mode:<16} {iters:<6} {float(m.get(\\'ttft\\', 0.0) or 0.0):<10.3f} \"\\n f\"{_avg_ttft(m):<10.3f} {float(m.get(\\'total_time\\', 0.0) or 0.0):<10.3f} \"\\n f\"{prompt:<9,} {cached:<9,} {_hit_rate(m):<7.1f} \"\\n f\"{cache_pct:<8.1f} {billable:<10,.0f} {save_pct:<7.1f}\")\\n\\n print(\"-\" * 112)\\n print(f\"\u6ce8\uff1aBill.Tok / Save% \u5047\u8bbe\u7f13\u5b58 token \u6309\u6b63\u5e38\u4ef7\u7684 {cache_price_ratio:.0%} \u8ba1\u8d39\"\\n f\"\uff08\u53ef\u7528 --cache-price-ratio \u8c03\u6574\uff09\uff0c\u4ec5\u4e3a\u6210\u672c\u793a\u610f\uff0c\u975e\u67d0\u5bb6\u670d\u52a1\u5546\u5b9e\u9645\u62a5\u4ef7\u3002\")\\n\\n\\ndef load_result_files(paths: List[str]) -> Dict[str, Any]:\\n \"\"\"Load result_*.json files into a {mode: {...}} dict for offline reporting.\"\"\"\\n results: Dict[str, Any] = {}\\n for path in sorted(paths):\\n try:\\n with open(path, \\'r\\') as f:\\n data = json.load(f)\\n except Exception as e:\\n logger.warning(f\"Skipping {path}: {e}\")\\n continue\\n\\n # A comparison_*.json holds many modes; a result_*.json holds one.\\n if \"mode\" not in data and all(isinstance(v, dict) and \"metrics\" in v\\n for v in data.values()):\\n for mode, entry in data.items():\\n results[mode] = {\"metrics\": _coerce_metrics(entry.get(\"metrics\", {})),\\n \"iterations\": entry.get(\"iterations\"),\\n \"_source\": path}\\n else:\\n mode = data.get(\"mode\", os.path.splitext(os.path.basename(path))[0])\\n results[mode] = {\"metrics\": _coerce_metrics(data.get(\"metrics\", {})),\\n \"iterations\": data.get(\"iterations\"),\\n \"_source\": path}\\n return results\\n\\n\\ndef run_report(inputs: List[str] = None, cache_price_ratio: float = 0.1) -> None:\\n \"\"\"Offline: build the comparison table from existing result_*.json files.\\n\\n No API key required - reads previously saved runs so the final result is\\n legible in one command without re-hitting the model.\\n \"\"\"\\n if not inputs:\\n inputs = [\"result_*.json\", \"comparison_*.json\"]\\n\\n paths: List[str] = []\\n for item in inputs:\\n if os.path.isdir(item):\\n paths.extend(glob.glob(os.path.join(item, \"result_*.json\")))\\n paths.extend(glob.glob(os.path.join(item, \"comparison_*.json\")))\\n else:\\n paths.extend(glob.glob(item))\\n\\n paths = sorted(set(paths))\\n if not paths:\\n logger.error(\"\u672a\u627e\u5230\u4efb\u4f55 result_*.json / comparison_*.json \u7ed3\u679c\u6587\u4ef6\u3002\"\\n \"\u8bf7\u5148\u8fd0\u884c --mode \u6216 --compare \u751f\u6210\u7ed3\u679c\uff0c\u6216\u7528 --input \u6307\u5b9a\u8def\u5f84\u3002\")\\n sys.exit(1)\\n\\n results = load_result_files(paths)\\n\\n print(\"\\\\n\" + \"=\" * 112)\\n print(\"KV CACHE \u79bb\u7ebf\u5bf9\u6bd4\u62a5\u544a\uff08\u57fa\u4e8e\u5df2\u4fdd\u5b58\u7684\u5b9e\u6d4b\u7ed3\u679c\uff09\")\\n print(\"=\" * 112)\\n print(f\"\u6570\u636e\u6765\u6e90\uff08{len(paths)} \u4e2a\u6587\u4ef6\uff09:\")\\n for mode, data in results.items():\\n print(f\" \u2022 {mode:<16} \u2190 {os.path.basename(data.get(\\'_source\\', \\'?\\'))}\")\\n\\n print_comparison_table(results, cache_price_ratio)\\n\\n print(\"\\\\n\ud83d\udcdd \u8bf4\u660e\uff1a\u4e0d\u540c\u7ed3\u679c\u6587\u4ef6\u53ef\u80fd\u6765\u81ea\u4e0d\u540c\u4efb\u52a1/\u65f6\u95f4\uff0c\u7edd\u5bf9\u6570\u503c\u4ec5\u4f9b\u540c\u4e00\u6b21\u8fd0\u884c\u5185\u6a2a\u5411\u5bf9\u6bd4\uff1b\"\\n \"\u5982\u9700\u4e25\u683c\u5bf9\u7167\uff0c\u8bf7\u7528 --compare \u5728\u540c\u4e00\u4efb\u52a1\u4e0b\u4e00\u6b21\u6027\u751f\u6210\u5168\u90e8\u6a21\u5f0f\u7684\u6570\u636e\u3002\")\\n\\n\\ndef create_summary_task() -> str:\\n \"\"\"Create a task that requires reading multiple files\"\"\"\\n return \"\"\"Please analyze and summarize all the projects in the week1 and week2 directories.\\nFor each project:\\n1. Find all Python files\\n2. Read the main files and understand the functionality\\n3. Identify the key features and purpose\\n4. Provide a comprehensive summary\\n\\nStart with week1 projects, then move to week2. Be thorough in your analysis.\"\"\"\\n\\n\\ndef run_single_mode(api_key: str, mode: str, task: str = None, root_dir: str = \"../..\",\\n model: str = DEFAULT_MODEL, output: str = None):\\n \"\"\"\\n Run agent in a single mode\\n\\n Args:\\n api_key: API key for Kimi\\n mode: KV cache mode to use\\n task: Custom task (optional)\\n root_dir: Root directory for file operations (default: \"../..\" = /projects from kv-cache dir)\\n model: Model to use\\n output: Output path for the result JSON (optional; auto-named if omitted)\\n \"\"\"\\n # Parse mode\\n mode_map = {\\n \"correct\": KVCacheMode.CORRECT,\\n \"dynamic_system\": KVCacheMode.DYNAMIC_SYSTEM,\\n \"shuffled_tools\": KVCacheMode.SHUFFLED_TOOLS,\\n \"dynamic_profile\": KVCacheMode.DYNAMIC_PROFILE,\\n \"sliding_window\": KVCacheMode.SLIDING_WINDOW,\\n \"text_format\": KVCacheMode.TEXT_FORMAT\\n }\\n \\n if mode not in mode_map:\\n logger.error(f\"Invalid mode: {mode}\")\\n logger.info(f\"Valid modes: {\\', \\'.join(mode_map.keys())}\")\\n return\\n \\n # Use default task if not provided\\n if not task:\\n task = create_summary_task()\\n \\n logger.info(f\"Running in mode: {mode}\")\\n logger.info(f\"Task: {task}\")\\n logger.info(\"=\"*80)\\n \\n # Create agent and execute task\\n agent = KVCacheAgent(\\n api_key=api_key,\\n mode=mode_map[mode],\\n model=model,\\n root_dir=root_dir,\\n verbose=True\\n )\\n \\n result = agent.execute_task(task, max_iterations=30)\\n \\n # Print results\\n print(\"\\\\n\" + \"=\"*80)\\n print(f\"EXECUTION RESULTS - Mode: {mode}\")\\n print(\"=\"*80)\\n \\n metrics = result[\"metrics\"]\\n print(f\"\\\\n\ud83d\udcca Performance Metrics:\")\\n print(f\" \u2022 Time to First Token (TTFT): {metrics.ttft:.3f} seconds\")\\n \\n # Show TTFT progression\\n if metrics.ttft_per_iteration:\\n print(f\" \u2022 TTFT per iteration:\")\\n for i, ttft in enumerate(metrics.ttft_per_iteration, 1):\\n print(f\" Iteration {i}: {ttft:.3f}s\")\\n\\n # Show improvement\\n if len(metrics.ttft_per_iteration) > 1:\\n first_ttft = metrics.ttft_per_iteration[0]\\n last_ttft = metrics.ttft_per_iteration[-1]\\n avg_after_first = sum(metrics.ttft_per_iteration[1:]) / len(metrics.ttft_per_iteration[1:])\\n print(f\" \u2022 TTFT Analysis:\")\\n print(f\" First iteration: {first_ttft:.3f}s\")\\n print(f\" Last iteration: {', 'total_lines': 537, 'lines_read': 537, 'offset': 0, 'end_line': 537, 'truncated': True, 'success': True}, error=None, timestamp=1784337587.088365)",
"ToolCall(name='read_file', arguments={'file_path': 'agent.py'}, result={'path': 'agent.py', 'content': '\"\"\"\\nKV Cache Demonstration Agent with ReAct Pattern\\nDemonstrates the importance of KV cache through correct and incorrect implementations.\\nUses local file system tools to read and search through code files.\\n\"\"\"\\n\\nimport json\\nimport os\\nimport re\\nimport time\\nimport logging\\nimport random\\nfrom typing import List, Dict, Any, Optional, Tuple\\nfrom dataclasses import dataclass, field, asdict\\nfrom enum import Enum\\nfrom datetime import datetime\\nfrom openai import OpenAI\\nimport glob as glob_module\\nimport subprocess\\n\\n\\ndef _is_reasoning_model(model) -> bool:\\n \"\"\"True for models that emit reasoning_content and only accept temperature=1.\\n\\n On the live Moonshot endpoint the whole current Kimi family reasons:\\n kimi-k2.5 / kimi-k2.6 / kimi-k2.7* / kimi-k3. The legacy moonshot-v1-*\\n chat models do NOT reason (and also do not report cached_tokens).\"\"\"\\n m = str(model or \"\").lower().replace(\"/\", \"-\")\\n if \"gpt-5\" in m:\\n return True\\n return any(tag in m for tag in (\"kimi-k2.5\", \"kimi-k2.6\", \"kimi-k2.7\", \"kimi-k3\"))\\n\\n\\ndef _reasoning_safe_temperature(model, requested=1.0):\\n \"\"\"Reasoning models (Kimi K2.5/K2.6/K2.7/K3, GPT-5, ...) only accept\\n temperature=1. Return 1 for those; otherwise the requested value so\\n non-reasoning providers (moonshot-v1, Doubao, DeepSeek) are unchanged.\"\"\"\\n return 1 if _is_reasoning_model(model) else requested\\n\\n\\ndef _reasoning_safe_max_tokens(model, requested=2000):\\n \"\"\"Reasoning models spend completion budget on hidden reasoning tokens\\n before emitting content / tool calls. Give them enough headroom so a\\n tool call is not truncated away; leave non-reasoning models unchanged.\"\"\"\\n return max(requested, 4096) if _is_reasoning_model(model) else requested\\n\\n\\n# Configure logging\\nlogging.basicConfig(level=logging.INFO, format=\\'%(asctime)s - %(levelname)s - %(message)s\\')\\nlogger = logging.getLogger(__name__)\\n\\n\\nclass KVCacheMode(Enum):\\n \"\"\"Different KV cache optimization modes\"\"\"\\n CORRECT = \"correct\" # Correct implementation with stable context\\n DYNAMIC_SYSTEM = \"dynamic_system\" # Changing system prompt with timestamp\\n SHUFFLED_TOOLS = \"shuffled_tools\" # Shuffling tool order each request\\n DYNAMIC_PROFILE = \"dynamic_profile\" # Changing user profile with credits\\n SLIDING_WINDOW = \"sliding_window\" # Only keeping recent 5 messages\\n TEXT_FORMAT = \"text_format\" # Formatting messages as plain text\\n\\n\\n@dataclass\\nclass ToolCall:\\n \"\"\"Represents a single tool call\"\"\"\\n name: str\\n arguments: Dict[str, Any]\\n result: Any = None\\n error: Optional[str] = None\\n timestamp: float = field(default_factory=time.time)\\n\\n\\n@dataclass\\nclass AgentMetrics:\\n \"\"\"Metrics for agent performance\"\"\"\\n ttft: float = 0.0 # Time to first token (first iteration)\\n ttft_per_iteration: List[float] = field(default_factory=list) # TTFT for each iteration\\n total_time: float = 0.0\\n iterations: int = 0\\n tool_calls: int = 0\\n cache_hits: int = 0\\n cache_misses: int = 0\\n prompt_tokens: int = 0\\n completion_tokens: int = 0\\n cached_tokens: int = 0\\n\\n\\nclass LocalFileTools:\\n \"\"\"Local implementations of file system tools\"\"\"\\n \\n def __init__(self, root_dir: str = \".\"):\\n self.root_dir = os.path.abspath(root_dir)\\n logger.info(f\"File tools initialized with root: {self.root_dir}\")\\n \\n def read_file(self, file_path: str, offset: int = 0, size: int = None) -> Dict[str, Any]:\\n \"\"\"\\n Read contents of a file\\n \\n Args:\\n file_path: Path to the file relative to root directory\\n offset: Line number to start reading from (0-based, default: 0)\\n size: Number of lines to read (default: None, read all)\\n \\n Returns:\\n Dictionary with file contents or error\\n \"\"\"\\n try:\\n full_path = os.path.join(self.root_dir, file_path)\\n \\n # Security check - ensure path is within root_dir\\n real_path = os.path.realpath(full_path)\\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n with open(real_path, \\'r\\', encoding=\\'utf-8\\', errors=\\'ignore\\') as f:\\n lines = f.readlines()\\n \\n total_lines = len(lines)\\n \\n # Apply offset and size\\n if offset < 0:\\n offset = 0\\n if offset >= total_lines:\\n return {\\n \"path\": file_path,\\n \"content\": \"\",\\n \"total_lines\": total_lines,\\n \"lines_read\": 0,\\n \"offset\": offset,\\n \"success\": True,\\n \"message\": f\"Offset {offset} exceeds file length ({total_lines} lines)\"\\n }\\n \\n # Determine end line\\n if size is None:\\n end = total_lines\\n else:\\n end = min(offset + size, total_lines)\\n \\n # Get the requested lines\\n selected_lines = lines[offset:end]\\n content = \\'\\'.join(selected_lines)\\n \\n # Apply size limit for safety (10KB)\\n truncated = False\\n if len(content) > 10000:\\n content = content[:10000]\\n truncated = True\\n \\n return {\\n \"path\": file_path,\\n \"content\": content,\\n \"total_lines\": total_lines,\\n \"lines_read\": len(selected_lines),\\n \"offset\": offset,\\n \"end_line\": end,\\n \"truncated\": truncated,\\n \"success\": True\\n }\\n except FileNotFoundError:\\n return {\\n \"error\": f\"File not found: {file_path}\",\\n \"success\": False\\n }\\n except Exception as e:\\n return {\\n \"error\": f\"Error reading file: {str(e)}\",\\n \"success\": False\\n }\\n \\n def find(self, pattern: str = \"*\", directory: str = \".\") -> Dict[str, Any]:\\n \"\"\"\\n Find files matching a pattern (similar to Unix find command)\\n \\n Args:\\n pattern: File name pattern (supports wildcards, default: \"*\" for all files)\\n directory: Directory to search in (relative to root_dir)\\n \\n Returns:\\n Dictionary with list of matching files\\n \"\"\"\\n try:\\n # Handle directory path properly\\n if directory == \".\":\\n search_dir = self.root_dir\\n else:\\n # Remove leading/trailing slashes for consistency\\n directory = directory.strip(\\'/\\')\\n search_dir = os.path.join(self.root_dir, directory)\\n \\n # Security check\\n real_path = os.path.realpath(search_dir)\\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n # Check if directory exists\\n if not os.path.exists(real_path):\\n return {\\n \"error\": f\"Directory not found: {directory}\",\\n \"success\": False\\n }\\n \\n # Use glob to find matching files\\n matches = []\\n for root, dirs, files in os.walk(real_path):\\n # Filter hidden directories and __pycache__\\n dirs[:] = [d for d in dirs if not d.startswith(\\'.\\') and d != \\'__pycache__\\']\\n \\n for file in files:\\n # Skip hidden files and .pyc files\\n if file.startswith(\\'.\\') or file.endswith(\\'.pyc\\'):\\n continue\\n \\n if glob_module.fnmatch.fnmatch(file, pattern):\\n # Get path relative to root_dir (not search_dir)\\n full_path = os.path.join(root, file)\\n rel_path = os.path.relpath(full_path, self.root_dir)\\n matches.append(rel_path)\\n \\n # Sort for consistency\\n matches.sort()\\n \\n # Limit results for demonstration\\n if len(matches) > 100:\\n matches = matches[:100]\\n truncated = True\\n else:\\n truncated = False\\n \\n return {\\n \"pattern\": pattern,\\n \"directory\": directory,\\n \"matches\": matches,\\n \"count\": len(matches),\\n \"truncated\": truncated,\\n \"success\": True\\n }\\n except Exception as e:\\n return {\\n \"error\": f\"Error finding files: {str(e)}\",\\n \"success\": False\\n }\\n \\n def grep(self, pattern: str, file_path: str = None, directory: str = None) -> Dict[str, Any]:\\n \"\"\"\\n Search for pattern in files (similar to Unix grep command)\\n \\n Args:\\n pattern: Regular expression pattern to search for\\n file_path: Single file to search in (optional)\\n directory: Directory to search in (optional)\\n \\n Returns:\\n Dictionary with matching lines\\n \"\"\"\\n try:\\n matches = []\\n files_searched = []\\n \\n if file_path:\\n # Search in single file\\n full_path = os.path.join(self.root_dir, file_path)\\n real_path = os.path.realpath(full_path)\\n \\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n ', 'total_lines': 869, 'lines_read': 869, 'offset': 0, 'end_line': 869, 'truncated': True, 'success': True}, error=None, timestamp=1784337587.0890272)"
],
"metrics": {
"ttft": 2.2061221599578857,
"ttft_per_iteration": [
2.2061221599578857,
2.673752784729004,
13.079047203063965
],
"total_time": 17.96152114868164,
"iterations": 3,
"tool_calls": 3,
"cache_hits": 3,
"cache_misses": 0,
"prompt_tokens": 7652,
"completion_tokens": 618,
"cached_tokens": 768
},
"mode": "dynamic_profile",
"model": "kimi-k2.6",
"task": "Find all Python files in this directory; read main.py and agent.py; summarize in 3 sentences."
}
result_dynamic_system_20260718_kimi_k2_6.json¶
{
"success": true,
"iterations": 3,
"tool_calls": [
"ToolCall(name='find', arguments={'pattern': '*.py'}, result={'pattern': '*.py', 'directory': '.', 'matches': ['agent.py', 'demo_quick.py', 'main.py', 'test_api.py', 'test_cache_invalidation.py', 'test_cached_tokens.py', 'test_completion.py', 'test_error_handling.py', 'test_file_range.py', 'test_interactive.py', 'test_message_flow.py', 'test_tools.py', 'test_ttft.py'], 'count': 13, 'truncated': False, 'success': True}, error=None, timestamp=1784337527.048646)",
"ToolCall(name='read_file', arguments={'file_path': 'main.py'}, result={'path': 'main.py', 'content': '\"\"\"\\nMain script to demonstrate KV cache importance\\nRuns the ReAct agent with different implementations and compares performance\\n\"\"\"\\n\\nimport os\\nimport sys\\nimport glob\\nimport json\\nimport argparse\\nimport logging\\nfrom typing import Dict, List, Any\\nfrom datetime import datetime\\nfrom dataclasses import asdict\\nfrom agent import KVCacheAgent, KVCacheMode, AgentMetrics, compare_implementations\\n\\n# Default model (Moonshot / Kimi). The whole current Kimi family (k2.5/k2.6/\\n# k2.7/k3) reports cached_tokens for automatic prefix caching AND reasons, so it\\n# only accepts temperature=1 (agent.py handles that automatically). kimi-k2.6 has\\n# the lightest reasoning footprint of the cache-reporting models, giving the\\n# cleanest TTFT while still exposing the prefix-cache hit metric this demo needs.\\n# (The non-reasoning moonshot-v1-* models do NOT report cached_tokens, so they\\n# cannot demonstrate the cache effect.)\\nDEFAULT_MODEL = \"kimi-k2.6\"\\n\\n# Configure logging\\nlogging.basicConfig(\\n level=logging.INFO,\\n format=\\'%(asctime)s - %(levelname)s - %(message)s\\',\\n handlers=[\\n logging.FileHandler(\\'kv_cache_demo.log\\'),\\n logging.StreamHandler()\\n ]\\n)\\nlogger = logging.getLogger(__name__)\\n\\n\\n# ---------------------------------------------------------------------------\\n# Metrics helpers (shared by live comparison and offline report)\\n# ---------------------------------------------------------------------------\\n\\ndef _coerce_metrics(metrics: Any) -> Dict[str, Any]:\\n \"\"\"Normalize a stored metrics value into a plain dict.\\n\\n Handles both formats found in result files:\\n - dict: produced by --compare (asdict) and by the fixed --mode path\\n - str : legacy single-mode files that stored repr(AgentMetrics(...))\\n because json.dump used default=str\\n \"\"\"\\n if isinstance(metrics, dict):\\n return metrics\\n if isinstance(metrics, str) and metrics.startswith(\"AgentMetrics(\"):\\n # Safe eval: only AgentMetrics is exposed, no builtins.\\n try:\\n obj = eval(metrics, {\"__builtins__\": {}}, {\"AgentMetrics\": AgentMetrics})\\n return asdict(obj)\\n except Exception as e: # pragma: no cover - defensive\\n logger.warning(f\"Could not parse legacy metrics string: {e}\")\\n return {}\\n\\n\\ndef _avg_ttft(m: Dict[str, Any]) -> float:\\n \"\"\"Average TTFT across iterations, falling back to first-iteration TTFT.\"\"\"\\n lst = m.get(\"ttft_per_iteration\") or []\\n return sum(lst) / len(lst) if lst else float(m.get(\"ttft\", 0.0) or 0.0)\\n\\n\\ndef _hit_rate(m: Dict[str, Any]) -> float:\\n total = (m.get(\"cache_hits\", 0) or 0) + (m.get(\"cache_misses\", 0) or 0)\\n return (m.get(\"cache_hits\", 0) or 0) / total * 100 if total else 0.0\\n\\n\\ndef _billable_tokens(m: Dict[str, Any], cache_price_ratio: float) -> float:\\n \"\"\"Illustrative billable prompt tokens under a prompt-cache discount.\\n\\n cached tokens are charged at cache_price_ratio of the normal price; the\\n rest at full price. This is a transparent function of the *measured*\\n token counts and a user-supplied ratio - it is not a fabricated\\n provider-specific price.\\n \"\"\"\\n prompt = m.get(\"prompt_tokens\", 0) or 0\\n cached = m.get(\"cached_tokens\", 0) or 0\\n cached = min(cached, prompt)\\n return (prompt - cached) + cached * cache_price_ratio\\n\\n\\ndef print_comparison_table(results: Dict[str, Any], cache_price_ratio: float = 0.1) -> None:\\n \"\"\"Render the cross-strategy comparison table (latency / cache / cost).\"\"\"\\n print(f\"\\\\n{\\'Mode\\':<16} {\\'Iters\\':<6} {\\'1st TTFT\\':<10} {\\'Avg TTFT\\':<10} \"\\n f\"{\\'Total(s)\\':<10} {\\'Prompt\\':<9} {\\'Cached\\':<9} {\\'Hit%\\':<7} \"\\n f\"{\\'Cache%\\':<8} {\\'Bill.Tok\\':<10} {\\'Save%\\':<7}\")\\n print(\"-\" * 112)\\n\\n for mode, data in results.items():\\n m = _coerce_metrics(data.get(\"metrics\", {}))\\n prompt = m.get(\"prompt_tokens\", 0) or 0\\n cached = m.get(\"cached_tokens\", 0) or 0\\n iters = data.get(\"iterations\", m.get(\"iterations\", 0)) or 0\\n cache_pct = cached / prompt * 100 if prompt else 0.0\\n billable = _billable_tokens(m, cache_price_ratio)\\n save_pct = (prompt - billable) / prompt * 100 if prompt else 0.0\\n\\n print(f\"{mode:<16} {iters:<6} {float(m.get(\\'ttft\\', 0.0) or 0.0):<10.3f} \"\\n f\"{_avg_ttft(m):<10.3f} {float(m.get(\\'total_time\\', 0.0) or 0.0):<10.3f} \"\\n f\"{prompt:<9,} {cached:<9,} {_hit_rate(m):<7.1f} \"\\n f\"{cache_pct:<8.1f} {billable:<10,.0f} {save_pct:<7.1f}\")\\n\\n print(\"-\" * 112)\\n print(f\"\u6ce8\uff1aBill.Tok / Save% \u5047\u8bbe\u7f13\u5b58 token \u6309\u6b63\u5e38\u4ef7\u7684 {cache_price_ratio:.0%} \u8ba1\u8d39\"\\n f\"\uff08\u53ef\u7528 --cache-price-ratio \u8c03\u6574\uff09\uff0c\u4ec5\u4e3a\u6210\u672c\u793a\u610f\uff0c\u975e\u67d0\u5bb6\u670d\u52a1\u5546\u5b9e\u9645\u62a5\u4ef7\u3002\")\\n\\n\\ndef load_result_files(paths: List[str]) -> Dict[str, Any]:\\n \"\"\"Load result_*.json files into a {mode: {...}} dict for offline reporting.\"\"\"\\n results: Dict[str, Any] = {}\\n for path in sorted(paths):\\n try:\\n with open(path, \\'r\\') as f:\\n data = json.load(f)\\n except Exception as e:\\n logger.warning(f\"Skipping {path}: {e}\")\\n continue\\n\\n # A comparison_*.json holds many modes; a result_*.json holds one.\\n if \"mode\" not in data and all(isinstance(v, dict) and \"metrics\" in v\\n for v in data.values()):\\n for mode, entry in data.items():\\n results[mode] = {\"metrics\": _coerce_metrics(entry.get(\"metrics\", {})),\\n \"iterations\": entry.get(\"iterations\"),\\n \"_source\": path}\\n else:\\n mode = data.get(\"mode\", os.path.splitext(os.path.basename(path))[0])\\n results[mode] = {\"metrics\": _coerce_metrics(data.get(\"metrics\", {})),\\n \"iterations\": data.get(\"iterations\"),\\n \"_source\": path}\\n return results\\n\\n\\ndef run_report(inputs: List[str] = None, cache_price_ratio: float = 0.1) -> None:\\n \"\"\"Offline: build the comparison table from existing result_*.json files.\\n\\n No API key required - reads previously saved runs so the final result is\\n legible in one command without re-hitting the model.\\n \"\"\"\\n if not inputs:\\n inputs = [\"result_*.json\", \"comparison_*.json\"]\\n\\n paths: List[str] = []\\n for item in inputs:\\n if os.path.isdir(item):\\n paths.extend(glob.glob(os.path.join(item, \"result_*.json\")))\\n paths.extend(glob.glob(os.path.join(item, \"comparison_*.json\")))\\n else:\\n paths.extend(glob.glob(item))\\n\\n paths = sorted(set(paths))\\n if not paths:\\n logger.error(\"\u672a\u627e\u5230\u4efb\u4f55 result_*.json / comparison_*.json \u7ed3\u679c\u6587\u4ef6\u3002\"\\n \"\u8bf7\u5148\u8fd0\u884c --mode \u6216 --compare \u751f\u6210\u7ed3\u679c\uff0c\u6216\u7528 --input \u6307\u5b9a\u8def\u5f84\u3002\")\\n sys.exit(1)\\n\\n results = load_result_files(paths)\\n\\n print(\"\\\\n\" + \"=\" * 112)\\n print(\"KV CACHE \u79bb\u7ebf\u5bf9\u6bd4\u62a5\u544a\uff08\u57fa\u4e8e\u5df2\u4fdd\u5b58\u7684\u5b9e\u6d4b\u7ed3\u679c\uff09\")\\n print(\"=\" * 112)\\n print(f\"\u6570\u636e\u6765\u6e90\uff08{len(paths)} \u4e2a\u6587\u4ef6\uff09:\")\\n for mode, data in results.items():\\n print(f\" \u2022 {mode:<16} \u2190 {os.path.basename(data.get(\\'_source\\', \\'?\\'))}\")\\n\\n print_comparison_table(results, cache_price_ratio)\\n\\n print(\"\\\\n\ud83d\udcdd \u8bf4\u660e\uff1a\u4e0d\u540c\u7ed3\u679c\u6587\u4ef6\u53ef\u80fd\u6765\u81ea\u4e0d\u540c\u4efb\u52a1/\u65f6\u95f4\uff0c\u7edd\u5bf9\u6570\u503c\u4ec5\u4f9b\u540c\u4e00\u6b21\u8fd0\u884c\u5185\u6a2a\u5411\u5bf9\u6bd4\uff1b\"\\n \"\u5982\u9700\u4e25\u683c\u5bf9\u7167\uff0c\u8bf7\u7528 --compare \u5728\u540c\u4e00\u4efb\u52a1\u4e0b\u4e00\u6b21\u6027\u751f\u6210\u5168\u90e8\u6a21\u5f0f\u7684\u6570\u636e\u3002\")\\n\\n\\ndef create_summary_task() -> str:\\n \"\"\"Create a task that requires reading multiple files\"\"\"\\n return \"\"\"Please analyze and summarize all the projects in the week1 and week2 directories.\\nFor each project:\\n1. Find all Python files\\n2. Read the main files and understand the functionality\\n3. Identify the key features and purpose\\n4. Provide a comprehensive summary\\n\\nStart with week1 projects, then move to week2. Be thorough in your analysis.\"\"\"\\n\\n\\ndef run_single_mode(api_key: str, mode: str, task: str = None, root_dir: str = \"../..\",\\n model: str = DEFAULT_MODEL, output: str = None):\\n \"\"\"\\n Run agent in a single mode\\n\\n Args:\\n api_key: API key for Kimi\\n mode: KV cache mode to use\\n task: Custom task (optional)\\n root_dir: Root directory for file operations (default: \"../..\" = /projects from kv-cache dir)\\n model: Model to use\\n output: Output path for the result JSON (optional; auto-named if omitted)\\n \"\"\"\\n # Parse mode\\n mode_map = {\\n \"correct\": KVCacheMode.CORRECT,\\n \"dynamic_system\": KVCacheMode.DYNAMIC_SYSTEM,\\n \"shuffled_tools\": KVCacheMode.SHUFFLED_TOOLS,\\n \"dynamic_profile\": KVCacheMode.DYNAMIC_PROFILE,\\n \"sliding_window\": KVCacheMode.SLIDING_WINDOW,\\n \"text_format\": KVCacheMode.TEXT_FORMAT\\n }\\n \\n if mode not in mode_map:\\n logger.error(f\"Invalid mode: {mode}\")\\n logger.info(f\"Valid modes: {\\', \\'.join(mode_map.keys())}\")\\n return\\n \\n # Use default task if not provided\\n if not task:\\n task = create_summary_task()\\n \\n logger.info(f\"Running in mode: {mode}\")\\n logger.info(f\"Task: {task}\")\\n logger.info(\"=\"*80)\\n \\n # Create agent and execute task\\n agent = KVCacheAgent(\\n api_key=api_key,\\n mode=mode_map[mode],\\n model=model,\\n root_dir=root_dir,\\n verbose=True\\n )\\n \\n result = agent.execute_task(task, max_iterations=30)\\n \\n # Print results\\n print(\"\\\\n\" + \"=\"*80)\\n print(f\"EXECUTION RESULTS - Mode: {mode}\")\\n print(\"=\"*80)\\n \\n metrics = result[\"metrics\"]\\n print(f\"\\\\n\ud83d\udcca Performance Metrics:\")\\n print(f\" \u2022 Time to First Token (TTFT): {metrics.ttft:.3f} seconds\")\\n \\n # Show TTFT progression\\n if metrics.ttft_per_iteration:\\n print(f\" \u2022 TTFT per iteration:\")\\n for i, ttft in enumerate(metrics.ttft_per_iteration, 1):\\n print(f\" Iteration {i}: {ttft:.3f}s\")\\n\\n # Show improvement\\n if len(metrics.ttft_per_iteration) > 1:\\n first_ttft = metrics.ttft_per_iteration[0]\\n last_ttft = metrics.ttft_per_iteration[-1]\\n avg_after_first = sum(metrics.ttft_per_iteration[1:]) / len(metrics.ttft_per_iteration[1:])\\n print(f\" \u2022 TTFT Analysis:\")\\n print(f\" First iteration: {first_ttft:.3f}s\")\\n print(f\" Last iteration: {', 'total_lines': 537, 'lines_read': 537, 'offset': 0, 'end_line': 537, 'truncated': True, 'success': True}, error=None, timestamp=1784337529.004626)",
"ToolCall(name='read_file', arguments={'file_path': 'agent.py'}, result={'path': 'agent.py', 'content': '\"\"\"\\nKV Cache Demonstration Agent with ReAct Pattern\\nDemonstrates the importance of KV cache through correct and incorrect implementations.\\nUses local file system tools to read and search through code files.\\n\"\"\"\\n\\nimport json\\nimport os\\nimport re\\nimport time\\nimport logging\\nimport random\\nfrom typing import List, Dict, Any, Optional, Tuple\\nfrom dataclasses import dataclass, field, asdict\\nfrom enum import Enum\\nfrom datetime import datetime\\nfrom openai import OpenAI\\nimport glob as glob_module\\nimport subprocess\\n\\n\\ndef _is_reasoning_model(model) -> bool:\\n \"\"\"True for models that emit reasoning_content and only accept temperature=1.\\n\\n On the live Moonshot endpoint the whole current Kimi family reasons:\\n kimi-k2.5 / kimi-k2.6 / kimi-k2.7* / kimi-k3. The legacy moonshot-v1-*\\n chat models do NOT reason (and also do not report cached_tokens).\"\"\"\\n m = str(model or \"\").lower().replace(\"/\", \"-\")\\n if \"gpt-5\" in m:\\n return True\\n return any(tag in m for tag in (\"kimi-k2.5\", \"kimi-k2.6\", \"kimi-k2.7\", \"kimi-k3\"))\\n\\n\\ndef _reasoning_safe_temperature(model, requested=1.0):\\n \"\"\"Reasoning models (Kimi K2.5/K2.6/K2.7/K3, GPT-5, ...) only accept\\n temperature=1. Return 1 for those; otherwise the requested value so\\n non-reasoning providers (moonshot-v1, Doubao, DeepSeek) are unchanged.\"\"\"\\n return 1 if _is_reasoning_model(model) else requested\\n\\n\\ndef _reasoning_safe_max_tokens(model, requested=2000):\\n \"\"\"Reasoning models spend completion budget on hidden reasoning tokens\\n before emitting content / tool calls. Give them enough headroom so a\\n tool call is not truncated away; leave non-reasoning models unchanged.\"\"\"\\n return max(requested, 4096) if _is_reasoning_model(model) else requested\\n\\n\\n# Configure logging\\nlogging.basicConfig(level=logging.INFO, format=\\'%(asctime)s - %(levelname)s - %(message)s\\')\\nlogger = logging.getLogger(__name__)\\n\\n\\nclass KVCacheMode(Enum):\\n \"\"\"Different KV cache optimization modes\"\"\"\\n CORRECT = \"correct\" # Correct implementation with stable context\\n DYNAMIC_SYSTEM = \"dynamic_system\" # Changing system prompt with timestamp\\n SHUFFLED_TOOLS = \"shuffled_tools\" # Shuffling tool order each request\\n DYNAMIC_PROFILE = \"dynamic_profile\" # Changing user profile with credits\\n SLIDING_WINDOW = \"sliding_window\" # Only keeping recent 5 messages\\n TEXT_FORMAT = \"text_format\" # Formatting messages as plain text\\n\\n\\n@dataclass\\nclass ToolCall:\\n \"\"\"Represents a single tool call\"\"\"\\n name: str\\n arguments: Dict[str, Any]\\n result: Any = None\\n error: Optional[str] = None\\n timestamp: float = field(default_factory=time.time)\\n\\n\\n@dataclass\\nclass AgentMetrics:\\n \"\"\"Metrics for agent performance\"\"\"\\n ttft: float = 0.0 # Time to first token (first iteration)\\n ttft_per_iteration: List[float] = field(default_factory=list) # TTFT for each iteration\\n total_time: float = 0.0\\n iterations: int = 0\\n tool_calls: int = 0\\n cache_hits: int = 0\\n cache_misses: int = 0\\n prompt_tokens: int = 0\\n completion_tokens: int = 0\\n cached_tokens: int = 0\\n\\n\\nclass LocalFileTools:\\n \"\"\"Local implementations of file system tools\"\"\"\\n \\n def __init__(self, root_dir: str = \".\"):\\n self.root_dir = os.path.abspath(root_dir)\\n logger.info(f\"File tools initialized with root: {self.root_dir}\")\\n \\n def read_file(self, file_path: str, offset: int = 0, size: int = None) -> Dict[str, Any]:\\n \"\"\"\\n Read contents of a file\\n \\n Args:\\n file_path: Path to the file relative to root directory\\n offset: Line number to start reading from (0-based, default: 0)\\n size: Number of lines to read (default: None, read all)\\n \\n Returns:\\n Dictionary with file contents or error\\n \"\"\"\\n try:\\n full_path = os.path.join(self.root_dir, file_path)\\n \\n # Security check - ensure path is within root_dir\\n real_path = os.path.realpath(full_path)\\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n with open(real_path, \\'r\\', encoding=\\'utf-8\\', errors=\\'ignore\\') as f:\\n lines = f.readlines()\\n \\n total_lines = len(lines)\\n \\n # Apply offset and size\\n if offset < 0:\\n offset = 0\\n if offset >= total_lines:\\n return {\\n \"path\": file_path,\\n \"content\": \"\",\\n \"total_lines\": total_lines,\\n \"lines_read\": 0,\\n \"offset\": offset,\\n \"success\": True,\\n \"message\": f\"Offset {offset} exceeds file length ({total_lines} lines)\"\\n }\\n \\n # Determine end line\\n if size is None:\\n end = total_lines\\n else:\\n end = min(offset + size, total_lines)\\n \\n # Get the requested lines\\n selected_lines = lines[offset:end]\\n content = \\'\\'.join(selected_lines)\\n \\n # Apply size limit for safety (10KB)\\n truncated = False\\n if len(content) > 10000:\\n content = content[:10000]\\n truncated = True\\n \\n return {\\n \"path\": file_path,\\n \"content\": content,\\n \"total_lines\": total_lines,\\n \"lines_read\": len(selected_lines),\\n \"offset\": offset,\\n \"end_line\": end,\\n \"truncated\": truncated,\\n \"success\": True\\n }\\n except FileNotFoundError:\\n return {\\n \"error\": f\"File not found: {file_path}\",\\n \"success\": False\\n }\\n except Exception as e:\\n return {\\n \"error\": f\"Error reading file: {str(e)}\",\\n \"success\": False\\n }\\n \\n def find(self, pattern: str = \"*\", directory: str = \".\") -> Dict[str, Any]:\\n \"\"\"\\n Find files matching a pattern (similar to Unix find command)\\n \\n Args:\\n pattern: File name pattern (supports wildcards, default: \"*\" for all files)\\n directory: Directory to search in (relative to root_dir)\\n \\n Returns:\\n Dictionary with list of matching files\\n \"\"\"\\n try:\\n # Handle directory path properly\\n if directory == \".\":\\n search_dir = self.root_dir\\n else:\\n # Remove leading/trailing slashes for consistency\\n directory = directory.strip(\\'/\\')\\n search_dir = os.path.join(self.root_dir, directory)\\n \\n # Security check\\n real_path = os.path.realpath(search_dir)\\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n # Check if directory exists\\n if not os.path.exists(real_path):\\n return {\\n \"error\": f\"Directory not found: {directory}\",\\n \"success\": False\\n }\\n \\n # Use glob to find matching files\\n matches = []\\n for root, dirs, files in os.walk(real_path):\\n # Filter hidden directories and __pycache__\\n dirs[:] = [d for d in dirs if not d.startswith(\\'.\\') and d != \\'__pycache__\\']\\n \\n for file in files:\\n # Skip hidden files and .pyc files\\n if file.startswith(\\'.\\') or file.endswith(\\'.pyc\\'):\\n continue\\n \\n if glob_module.fnmatch.fnmatch(file, pattern):\\n # Get path relative to root_dir (not search_dir)\\n full_path = os.path.join(root, file)\\n rel_path = os.path.relpath(full_path, self.root_dir)\\n matches.append(rel_path)\\n \\n # Sort for consistency\\n matches.sort()\\n \\n # Limit results for demonstration\\n if len(matches) > 100:\\n matches = matches[:100]\\n truncated = True\\n else:\\n truncated = False\\n \\n return {\\n \"pattern\": pattern,\\n \"directory\": directory,\\n \"matches\": matches,\\n \"count\": len(matches),\\n \"truncated\": truncated,\\n \"success\": True\\n }\\n except Exception as e:\\n return {\\n \"error\": f\"Error finding files: {str(e)}\",\\n \"success\": False\\n }\\n \\n def grep(self, pattern: str, file_path: str = None, directory: str = None) -> Dict[str, Any]:\\n \"\"\"\\n Search for pattern in files (similar to Unix grep command)\\n \\n Args:\\n pattern: Regular expression pattern to search for\\n file_path: Single file to search in (optional)\\n directory: Directory to search in (optional)\\n \\n Returns:\\n Dictionary with matching lines\\n \"\"\"\\n try:\\n matches = []\\n files_searched = []\\n \\n if file_path:\\n # Search in single file\\n full_path = os.path.join(self.root_dir, file_path)\\n real_path = os.path.realpath(full_path)\\n \\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n ', 'total_lines': 869, 'lines_read': 869, 'offset': 0, 'end_line': 869, 'truncated': True, 'success': True}, error=None, timestamp=1784337529.0049899)"
],
"metrics": {
"ttft": 2.4965670108795166,
"ttft_per_iteration": [
2.4965670108795166,
1.9555277824401855,
19.803843021392822
],
"total_time": 24.259823322296143,
"iterations": 3,
"tool_calls": 3,
"cache_hits": 3,
"cache_misses": 0,
"prompt_tokens": 7639,
"completion_tokens": 878,
"cached_tokens": 768
},
"mode": "dynamic_system",
"model": "kimi-k2.6",
"task": "Find all Python files in this directory; read main.py and agent.py; summarize in 3 sentences."
}
result_shuffled_tools_20260718_kimi_k2_6.json¶
{
"success": true,
"iterations": 3,
"tool_calls": [
"ToolCall(name='find', arguments={'pattern': '*.py'}, result={'pattern': '*.py', 'directory': '.', 'matches': ['agent.py', 'demo_quick.py', 'main.py', 'test_api.py', 'test_cache_invalidation.py', 'test_cached_tokens.py', 'test_completion.py', 'test_error_handling.py', 'test_file_range.py', 'test_interactive.py', 'test_message_flow.py', 'test_tools.py', 'test_ttft.py'], 'count': 13, 'truncated': False, 'success': True}, error=None, timestamp=1784337556.6428301)",
"ToolCall(name='read_file', arguments={'file_path': 'main.py'}, result={'path': 'main.py', 'content': '\"\"\"\\nMain script to demonstrate KV cache importance\\nRuns the ReAct agent with different implementations and compares performance\\n\"\"\"\\n\\nimport os\\nimport sys\\nimport glob\\nimport json\\nimport argparse\\nimport logging\\nfrom typing import Dict, List, Any\\nfrom datetime import datetime\\nfrom dataclasses import asdict\\nfrom agent import KVCacheAgent, KVCacheMode, AgentMetrics, compare_implementations\\n\\n# Default model (Moonshot / Kimi). The whole current Kimi family (k2.5/k2.6/\\n# k2.7/k3) reports cached_tokens for automatic prefix caching AND reasons, so it\\n# only accepts temperature=1 (agent.py handles that automatically). kimi-k2.6 has\\n# the lightest reasoning footprint of the cache-reporting models, giving the\\n# cleanest TTFT while still exposing the prefix-cache hit metric this demo needs.\\n# (The non-reasoning moonshot-v1-* models do NOT report cached_tokens, so they\\n# cannot demonstrate the cache effect.)\\nDEFAULT_MODEL = \"kimi-k2.6\"\\n\\n# Configure logging\\nlogging.basicConfig(\\n level=logging.INFO,\\n format=\\'%(asctime)s - %(levelname)s - %(message)s\\',\\n handlers=[\\n logging.FileHandler(\\'kv_cache_demo.log\\'),\\n logging.StreamHandler()\\n ]\\n)\\nlogger = logging.getLogger(__name__)\\n\\n\\n# ---------------------------------------------------------------------------\\n# Metrics helpers (shared by live comparison and offline report)\\n# ---------------------------------------------------------------------------\\n\\ndef _coerce_metrics(metrics: Any) -> Dict[str, Any]:\\n \"\"\"Normalize a stored metrics value into a plain dict.\\n\\n Handles both formats found in result files:\\n - dict: produced by --compare (asdict) and by the fixed --mode path\\n - str : legacy single-mode files that stored repr(AgentMetrics(...))\\n because json.dump used default=str\\n \"\"\"\\n if isinstance(metrics, dict):\\n return metrics\\n if isinstance(metrics, str) and metrics.startswith(\"AgentMetrics(\"):\\n # Safe eval: only AgentMetrics is exposed, no builtins.\\n try:\\n obj = eval(metrics, {\"__builtins__\": {}}, {\"AgentMetrics\": AgentMetrics})\\n return asdict(obj)\\n except Exception as e: # pragma: no cover - defensive\\n logger.warning(f\"Could not parse legacy metrics string: {e}\")\\n return {}\\n\\n\\ndef _avg_ttft(m: Dict[str, Any]) -> float:\\n \"\"\"Average TTFT across iterations, falling back to first-iteration TTFT.\"\"\"\\n lst = m.get(\"ttft_per_iteration\") or []\\n return sum(lst) / len(lst) if lst else float(m.get(\"ttft\", 0.0) or 0.0)\\n\\n\\ndef _hit_rate(m: Dict[str, Any]) -> float:\\n total = (m.get(\"cache_hits\", 0) or 0) + (m.get(\"cache_misses\", 0) or 0)\\n return (m.get(\"cache_hits\", 0) or 0) / total * 100 if total else 0.0\\n\\n\\ndef _billable_tokens(m: Dict[str, Any], cache_price_ratio: float) -> float:\\n \"\"\"Illustrative billable prompt tokens under a prompt-cache discount.\\n\\n cached tokens are charged at cache_price_ratio of the normal price; the\\n rest at full price. This is a transparent function of the *measured*\\n token counts and a user-supplied ratio - it is not a fabricated\\n provider-specific price.\\n \"\"\"\\n prompt = m.get(\"prompt_tokens\", 0) or 0\\n cached = m.get(\"cached_tokens\", 0) or 0\\n cached = min(cached, prompt)\\n return (prompt - cached) + cached * cache_price_ratio\\n\\n\\ndef print_comparison_table(results: Dict[str, Any], cache_price_ratio: float = 0.1) -> None:\\n \"\"\"Render the cross-strategy comparison table (latency / cache / cost).\"\"\"\\n print(f\"\\\\n{\\'Mode\\':<16} {\\'Iters\\':<6} {\\'1st TTFT\\':<10} {\\'Avg TTFT\\':<10} \"\\n f\"{\\'Total(s)\\':<10} {\\'Prompt\\':<9} {\\'Cached\\':<9} {\\'Hit%\\':<7} \"\\n f\"{\\'Cache%\\':<8} {\\'Bill.Tok\\':<10} {\\'Save%\\':<7}\")\\n print(\"-\" * 112)\\n\\n for mode, data in results.items():\\n m = _coerce_metrics(data.get(\"metrics\", {}))\\n prompt = m.get(\"prompt_tokens\", 0) or 0\\n cached = m.get(\"cached_tokens\", 0) or 0\\n iters = data.get(\"iterations\", m.get(\"iterations\", 0)) or 0\\n cache_pct = cached / prompt * 100 if prompt else 0.0\\n billable = _billable_tokens(m, cache_price_ratio)\\n save_pct = (prompt - billable) / prompt * 100 if prompt else 0.0\\n\\n print(f\"{mode:<16} {iters:<6} {float(m.get(\\'ttft\\', 0.0) or 0.0):<10.3f} \"\\n f\"{_avg_ttft(m):<10.3f} {float(m.get(\\'total_time\\', 0.0) or 0.0):<10.3f} \"\\n f\"{prompt:<9,} {cached:<9,} {_hit_rate(m):<7.1f} \"\\n f\"{cache_pct:<8.1f} {billable:<10,.0f} {save_pct:<7.1f}\")\\n\\n print(\"-\" * 112)\\n print(f\"\u6ce8\uff1aBill.Tok / Save% \u5047\u8bbe\u7f13\u5b58 token \u6309\u6b63\u5e38\u4ef7\u7684 {cache_price_ratio:.0%} \u8ba1\u8d39\"\\n f\"\uff08\u53ef\u7528 --cache-price-ratio \u8c03\u6574\uff09\uff0c\u4ec5\u4e3a\u6210\u672c\u793a\u610f\uff0c\u975e\u67d0\u5bb6\u670d\u52a1\u5546\u5b9e\u9645\u62a5\u4ef7\u3002\")\\n\\n\\ndef load_result_files(paths: List[str]) -> Dict[str, Any]:\\n \"\"\"Load result_*.json files into a {mode: {...}} dict for offline reporting.\"\"\"\\n results: Dict[str, Any] = {}\\n for path in sorted(paths):\\n try:\\n with open(path, \\'r\\') as f:\\n data = json.load(f)\\n except Exception as e:\\n logger.warning(f\"Skipping {path}: {e}\")\\n continue\\n\\n # A comparison_*.json holds many modes; a result_*.json holds one.\\n if \"mode\" not in data and all(isinstance(v, dict) and \"metrics\" in v\\n for v in data.values()):\\n for mode, entry in data.items():\\n results[mode] = {\"metrics\": _coerce_metrics(entry.get(\"metrics\", {})),\\n \"iterations\": entry.get(\"iterations\"),\\n \"_source\": path}\\n else:\\n mode = data.get(\"mode\", os.path.splitext(os.path.basename(path))[0])\\n results[mode] = {\"metrics\": _coerce_metrics(data.get(\"metrics\", {})),\\n \"iterations\": data.get(\"iterations\"),\\n \"_source\": path}\\n return results\\n\\n\\ndef run_report(inputs: List[str] = None, cache_price_ratio: float = 0.1) -> None:\\n \"\"\"Offline: build the comparison table from existing result_*.json files.\\n\\n No API key required - reads previously saved runs so the final result is\\n legible in one command without re-hitting the model.\\n \"\"\"\\n if not inputs:\\n inputs = [\"result_*.json\", \"comparison_*.json\"]\\n\\n paths: List[str] = []\\n for item in inputs:\\n if os.path.isdir(item):\\n paths.extend(glob.glob(os.path.join(item, \"result_*.json\")))\\n paths.extend(glob.glob(os.path.join(item, \"comparison_*.json\")))\\n else:\\n paths.extend(glob.glob(item))\\n\\n paths = sorted(set(paths))\\n if not paths:\\n logger.error(\"\u672a\u627e\u5230\u4efb\u4f55 result_*.json / comparison_*.json \u7ed3\u679c\u6587\u4ef6\u3002\"\\n \"\u8bf7\u5148\u8fd0\u884c --mode \u6216 --compare \u751f\u6210\u7ed3\u679c\uff0c\u6216\u7528 --input \u6307\u5b9a\u8def\u5f84\u3002\")\\n sys.exit(1)\\n\\n results = load_result_files(paths)\\n\\n print(\"\\\\n\" + \"=\" * 112)\\n print(\"KV CACHE \u79bb\u7ebf\u5bf9\u6bd4\u62a5\u544a\uff08\u57fa\u4e8e\u5df2\u4fdd\u5b58\u7684\u5b9e\u6d4b\u7ed3\u679c\uff09\")\\n print(\"=\" * 112)\\n print(f\"\u6570\u636e\u6765\u6e90\uff08{len(paths)} \u4e2a\u6587\u4ef6\uff09:\")\\n for mode, data in results.items():\\n print(f\" \u2022 {mode:<16} \u2190 {os.path.basename(data.get(\\'_source\\', \\'?\\'))}\")\\n\\n print_comparison_table(results, cache_price_ratio)\\n\\n print(\"\\\\n\ud83d\udcdd \u8bf4\u660e\uff1a\u4e0d\u540c\u7ed3\u679c\u6587\u4ef6\u53ef\u80fd\u6765\u81ea\u4e0d\u540c\u4efb\u52a1/\u65f6\u95f4\uff0c\u7edd\u5bf9\u6570\u503c\u4ec5\u4f9b\u540c\u4e00\u6b21\u8fd0\u884c\u5185\u6a2a\u5411\u5bf9\u6bd4\uff1b\"\\n \"\u5982\u9700\u4e25\u683c\u5bf9\u7167\uff0c\u8bf7\u7528 --compare \u5728\u540c\u4e00\u4efb\u52a1\u4e0b\u4e00\u6b21\u6027\u751f\u6210\u5168\u90e8\u6a21\u5f0f\u7684\u6570\u636e\u3002\")\\n\\n\\ndef create_summary_task() -> str:\\n \"\"\"Create a task that requires reading multiple files\"\"\"\\n return \"\"\"Please analyze and summarize all the projects in the week1 and week2 directories.\\nFor each project:\\n1. Find all Python files\\n2. Read the main files and understand the functionality\\n3. Identify the key features and purpose\\n4. Provide a comprehensive summary\\n\\nStart with week1 projects, then move to week2. Be thorough in your analysis.\"\"\"\\n\\n\\ndef run_single_mode(api_key: str, mode: str, task: str = None, root_dir: str = \"../..\",\\n model: str = DEFAULT_MODEL, output: str = None):\\n \"\"\"\\n Run agent in a single mode\\n\\n Args:\\n api_key: API key for Kimi\\n mode: KV cache mode to use\\n task: Custom task (optional)\\n root_dir: Root directory for file operations (default: \"../..\" = /projects from kv-cache dir)\\n model: Model to use\\n output: Output path for the result JSON (optional; auto-named if omitted)\\n \"\"\"\\n # Parse mode\\n mode_map = {\\n \"correct\": KVCacheMode.CORRECT,\\n \"dynamic_system\": KVCacheMode.DYNAMIC_SYSTEM,\\n \"shuffled_tools\": KVCacheMode.SHUFFLED_TOOLS,\\n \"dynamic_profile\": KVCacheMode.DYNAMIC_PROFILE,\\n \"sliding_window\": KVCacheMode.SLIDING_WINDOW,\\n \"text_format\": KVCacheMode.TEXT_FORMAT\\n }\\n \\n if mode not in mode_map:\\n logger.error(f\"Invalid mode: {mode}\")\\n logger.info(f\"Valid modes: {\\', \\'.join(mode_map.keys())}\")\\n return\\n \\n # Use default task if not provided\\n if not task:\\n task = create_summary_task()\\n \\n logger.info(f\"Running in mode: {mode}\")\\n logger.info(f\"Task: {task}\")\\n logger.info(\"=\"*80)\\n \\n # Create agent and execute task\\n agent = KVCacheAgent(\\n api_key=api_key,\\n mode=mode_map[mode],\\n model=model,\\n root_dir=root_dir,\\n verbose=True\\n )\\n \\n result = agent.execute_task(task, max_iterations=30)\\n \\n # Print results\\n print(\"\\\\n\" + \"=\"*80)\\n print(f\"EXECUTION RESULTS - Mode: {mode}\")\\n print(\"=\"*80)\\n \\n metrics = result[\"metrics\"]\\n print(f\"\\\\n\ud83d\udcca Performance Metrics:\")\\n print(f\" \u2022 Time to First Token (TTFT): {metrics.ttft:.3f} seconds\")\\n \\n # Show TTFT progression\\n if metrics.ttft_per_iteration:\\n print(f\" \u2022 TTFT per iteration:\")\\n for i, ttft in enumerate(metrics.ttft_per_iteration, 1):\\n print(f\" Iteration {i}: {ttft:.3f}s\")\\n\\n # Show improvement\\n if len(metrics.ttft_per_iteration) > 1:\\n first_ttft = metrics.ttft_per_iteration[0]\\n last_ttft = metrics.ttft_per_iteration[-1]\\n avg_after_first = sum(metrics.ttft_per_iteration[1:]) / len(metrics.ttft_per_iteration[1:])\\n print(f\" \u2022 TTFT Analysis:\")\\n print(f\" First iteration: {first_ttft:.3f}s\")\\n print(f\" Last iteration: {', 'total_lines': 537, 'lines_read': 537, 'offset': 0, 'end_line': 537, 'truncated': True, 'success': True}, error=None, timestamp=1784337561.854176)",
"ToolCall(name='read_file', arguments={'file_path': 'agent.py'}, result={'path': 'agent.py', 'content': '\"\"\"\\nKV Cache Demonstration Agent with ReAct Pattern\\nDemonstrates the importance of KV cache through correct and incorrect implementations.\\nUses local file system tools to read and search through code files.\\n\"\"\"\\n\\nimport json\\nimport os\\nimport re\\nimport time\\nimport logging\\nimport random\\nfrom typing import List, Dict, Any, Optional, Tuple\\nfrom dataclasses import dataclass, field, asdict\\nfrom enum import Enum\\nfrom datetime import datetime\\nfrom openai import OpenAI\\nimport glob as glob_module\\nimport subprocess\\n\\n\\ndef _is_reasoning_model(model) -> bool:\\n \"\"\"True for models that emit reasoning_content and only accept temperature=1.\\n\\n On the live Moonshot endpoint the whole current Kimi family reasons:\\n kimi-k2.5 / kimi-k2.6 / kimi-k2.7* / kimi-k3. The legacy moonshot-v1-*\\n chat models do NOT reason (and also do not report cached_tokens).\"\"\"\\n m = str(model or \"\").lower().replace(\"/\", \"-\")\\n if \"gpt-5\" in m:\\n return True\\n return any(tag in m for tag in (\"kimi-k2.5\", \"kimi-k2.6\", \"kimi-k2.7\", \"kimi-k3\"))\\n\\n\\ndef _reasoning_safe_temperature(model, requested=1.0):\\n \"\"\"Reasoning models (Kimi K2.5/K2.6/K2.7/K3, GPT-5, ...) only accept\\n temperature=1. Return 1 for those; otherwise the requested value so\\n non-reasoning providers (moonshot-v1, Doubao, DeepSeek) are unchanged.\"\"\"\\n return 1 if _is_reasoning_model(model) else requested\\n\\n\\ndef _reasoning_safe_max_tokens(model, requested=2000):\\n \"\"\"Reasoning models spend completion budget on hidden reasoning tokens\\n before emitting content / tool calls. Give them enough headroom so a\\n tool call is not truncated away; leave non-reasoning models unchanged.\"\"\"\\n return max(requested, 4096) if _is_reasoning_model(model) else requested\\n\\n\\n# Configure logging\\nlogging.basicConfig(level=logging.INFO, format=\\'%(asctime)s - %(levelname)s - %(message)s\\')\\nlogger = logging.getLogger(__name__)\\n\\n\\nclass KVCacheMode(Enum):\\n \"\"\"Different KV cache optimization modes\"\"\"\\n CORRECT = \"correct\" # Correct implementation with stable context\\n DYNAMIC_SYSTEM = \"dynamic_system\" # Changing system prompt with timestamp\\n SHUFFLED_TOOLS = \"shuffled_tools\" # Shuffling tool order each request\\n DYNAMIC_PROFILE = \"dynamic_profile\" # Changing user profile with credits\\n SLIDING_WINDOW = \"sliding_window\" # Only keeping recent 5 messages\\n TEXT_FORMAT = \"text_format\" # Formatting messages as plain text\\n\\n\\n@dataclass\\nclass ToolCall:\\n \"\"\"Represents a single tool call\"\"\"\\n name: str\\n arguments: Dict[str, Any]\\n result: Any = None\\n error: Optional[str] = None\\n timestamp: float = field(default_factory=time.time)\\n\\n\\n@dataclass\\nclass AgentMetrics:\\n \"\"\"Metrics for agent performance\"\"\"\\n ttft: float = 0.0 # Time to first token (first iteration)\\n ttft_per_iteration: List[float] = field(default_factory=list) # TTFT for each iteration\\n total_time: float = 0.0\\n iterations: int = 0\\n tool_calls: int = 0\\n cache_hits: int = 0\\n cache_misses: int = 0\\n prompt_tokens: int = 0\\n completion_tokens: int = 0\\n cached_tokens: int = 0\\n\\n\\nclass LocalFileTools:\\n \"\"\"Local implementations of file system tools\"\"\"\\n \\n def __init__(self, root_dir: str = \".\"):\\n self.root_dir = os.path.abspath(root_dir)\\n logger.info(f\"File tools initialized with root: {self.root_dir}\")\\n \\n def read_file(self, file_path: str, offset: int = 0, size: int = None) -> Dict[str, Any]:\\n \"\"\"\\n Read contents of a file\\n \\n Args:\\n file_path: Path to the file relative to root directory\\n offset: Line number to start reading from (0-based, default: 0)\\n size: Number of lines to read (default: None, read all)\\n \\n Returns:\\n Dictionary with file contents or error\\n \"\"\"\\n try:\\n full_path = os.path.join(self.root_dir, file_path)\\n \\n # Security check - ensure path is within root_dir\\n real_path = os.path.realpath(full_path)\\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n with open(real_path, \\'r\\', encoding=\\'utf-8\\', errors=\\'ignore\\') as f:\\n lines = f.readlines()\\n \\n total_lines = len(lines)\\n \\n # Apply offset and size\\n if offset < 0:\\n offset = 0\\n if offset >= total_lines:\\n return {\\n \"path\": file_path,\\n \"content\": \"\",\\n \"total_lines\": total_lines,\\n \"lines_read\": 0,\\n \"offset\": offset,\\n \"success\": True,\\n \"message\": f\"Offset {offset} exceeds file length ({total_lines} lines)\"\\n }\\n \\n # Determine end line\\n if size is None:\\n end = total_lines\\n else:\\n end = min(offset + size, total_lines)\\n \\n # Get the requested lines\\n selected_lines = lines[offset:end]\\n content = \\'\\'.join(selected_lines)\\n \\n # Apply size limit for safety (10KB)\\n truncated = False\\n if len(content) > 10000:\\n content = content[:10000]\\n truncated = True\\n \\n return {\\n \"path\": file_path,\\n \"content\": content,\\n \"total_lines\": total_lines,\\n \"lines_read\": len(selected_lines),\\n \"offset\": offset,\\n \"end_line\": end,\\n \"truncated\": truncated,\\n \"success\": True\\n }\\n except FileNotFoundError:\\n return {\\n \"error\": f\"File not found: {file_path}\",\\n \"success\": False\\n }\\n except Exception as e:\\n return {\\n \"error\": f\"Error reading file: {str(e)}\",\\n \"success\": False\\n }\\n \\n def find(self, pattern: str = \"*\", directory: str = \".\") -> Dict[str, Any]:\\n \"\"\"\\n Find files matching a pattern (similar to Unix find command)\\n \\n Args:\\n pattern: File name pattern (supports wildcards, default: \"*\" for all files)\\n directory: Directory to search in (relative to root_dir)\\n \\n Returns:\\n Dictionary with list of matching files\\n \"\"\"\\n try:\\n # Handle directory path properly\\n if directory == \".\":\\n search_dir = self.root_dir\\n else:\\n # Remove leading/trailing slashes for consistency\\n directory = directory.strip(\\'/\\')\\n search_dir = os.path.join(self.root_dir, directory)\\n \\n # Security check\\n real_path = os.path.realpath(search_dir)\\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n # Check if directory exists\\n if not os.path.exists(real_path):\\n return {\\n \"error\": f\"Directory not found: {directory}\",\\n \"success\": False\\n }\\n \\n # Use glob to find matching files\\n matches = []\\n for root, dirs, files in os.walk(real_path):\\n # Filter hidden directories and __pycache__\\n dirs[:] = [d for d in dirs if not d.startswith(\\'.\\') and d != \\'__pycache__\\']\\n \\n for file in files:\\n # Skip hidden files and .pyc files\\n if file.startswith(\\'.\\') or file.endswith(\\'.pyc\\'):\\n continue\\n \\n if glob_module.fnmatch.fnmatch(file, pattern):\\n # Get path relative to root_dir (not search_dir)\\n full_path = os.path.join(root, file)\\n rel_path = os.path.relpath(full_path, self.root_dir)\\n matches.append(rel_path)\\n \\n # Sort for consistency\\n matches.sort()\\n \\n # Limit results for demonstration\\n if len(matches) > 100:\\n matches = matches[:100]\\n truncated = True\\n else:\\n truncated = False\\n \\n return {\\n \"pattern\": pattern,\\n \"directory\": directory,\\n \"matches\": matches,\\n \"count\": len(matches),\\n \"truncated\": truncated,\\n \"success\": True\\n }\\n except Exception as e:\\n return {\\n \"error\": f\"Error finding files: {str(e)}\",\\n \"success\": False\\n }\\n \\n def grep(self, pattern: str, file_path: str = None, directory: str = None) -> Dict[str, Any]:\\n \"\"\"\\n Search for pattern in files (similar to Unix grep command)\\n \\n Args:\\n pattern: Regular expression pattern to search for\\n file_path: Single file to search in (optional)\\n directory: Directory to search in (optional)\\n \\n Returns:\\n Dictionary with matching lines\\n \"\"\"\\n try:\\n matches = []\\n files_searched = []\\n \\n if file_path:\\n # Search in single file\\n full_path = os.path.join(self.root_dir, file_path)\\n real_path = os.path.realpath(full_path)\\n \\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n ', 'total_lines': 869, 'lines_read': 869, 'offset': 0, 'end_line': 869, 'truncated': True, 'success': True}, error=None, timestamp=1784337561.854856)"
],
"metrics": {
"ttft": 7.818356990814209,
"ttft_per_iteration": [
7.818356990814209,
5.210463047027588,
20.33803105354309
],
"total_time": 33.36925220489502,
"iterations": 3,
"tool_calls": 3,
"cache_hits": 1,
"cache_misses": 0,
"prompt_tokens": 7568,
"completion_tokens": 514,
"cached_tokens": 256
},
"mode": "shuffled_tools",
"model": "kimi-k2.6",
"task": "Find all Python files in this directory; read main.py and agent.py; summarize in 3 sentences."
}
result_sliding_window_20260718_kimi_k2_6.json¶
{
"success": false,
"iterations": 5,
"tool_calls": [
"ToolCall(name='find', arguments={'pattern': '*.py'}, result={'pattern': '*.py', 'directory': '.', 'matches': ['agent.py', 'demo_quick.py', 'main.py', 'test_api.py', 'test_cache_invalidation.py', 'test_cached_tokens.py', 'test_completion.py', 'test_error_handling.py', 'test_file_range.py', 'test_interactive.py', 'test_message_flow.py', 'test_tools.py', 'test_ttft.py'], 'count': 13, 'truncated': False, 'success': True}, error=None, timestamp=1784337602.4177501)",
"ToolCall(name='find', arguments={'pattern': '*.py'}, result={'pattern': '*.py', 'directory': '.', 'matches': ['agent.py', 'demo_quick.py', 'main.py', 'test_api.py', 'test_cache_invalidation.py', 'test_cached_tokens.py', 'test_completion.py', 'test_error_handling.py', 'test_file_range.py', 'test_interactive.py', 'test_message_flow.py', 'test_tools.py', 'test_ttft.py'], 'count': 13, 'truncated': False, 'success': True}, error=None, timestamp=1784337604.403392)",
"ToolCall(name='find', arguments={'pattern': '*.py'}, result={'pattern': '*.py', 'directory': '.', 'matches': ['agent.py', 'demo_quick.py', 'main.py', 'test_api.py', 'test_cache_invalidation.py', 'test_cached_tokens.py', 'test_completion.py', 'test_error_handling.py', 'test_file_range.py', 'test_interactive.py', 'test_message_flow.py', 'test_tools.py', 'test_ttft.py'], 'count': 13, 'truncated': False, 'success': True}, error=None, timestamp=1784337608.362566)",
"ToolCall(name='read_file', arguments={'file_path': 'main.py'}, result={'path': 'main.py', 'content': '\"\"\"\\nMain script to demonstrate KV cache importance\\nRuns the ReAct agent with different implementations and compares performance\\n\"\"\"\\n\\nimport os\\nimport sys\\nimport glob\\nimport json\\nimport argparse\\nimport logging\\nfrom typing import Dict, List, Any\\nfrom datetime import datetime\\nfrom dataclasses import asdict\\nfrom agent import KVCacheAgent, KVCacheMode, AgentMetrics, compare_implementations\\n\\n# Default model (Moonshot / Kimi). The whole current Kimi family (k2.5/k2.6/\\n# k2.7/k3) reports cached_tokens for automatic prefix caching AND reasons, so it\\n# only accepts temperature=1 (agent.py handles that automatically). kimi-k2.6 has\\n# the lightest reasoning footprint of the cache-reporting models, giving the\\n# cleanest TTFT while still exposing the prefix-cache hit metric this demo needs.\\n# (The non-reasoning moonshot-v1-* models do NOT report cached_tokens, so they\\n# cannot demonstrate the cache effect.)\\nDEFAULT_MODEL = \"kimi-k2.6\"\\n\\n# Configure logging\\nlogging.basicConfig(\\n level=logging.INFO,\\n format=\\'%(asctime)s - %(levelname)s - %(message)s\\',\\n handlers=[\\n logging.FileHandler(\\'kv_cache_demo.log\\'),\\n logging.StreamHandler()\\n ]\\n)\\nlogger = logging.getLogger(__name__)\\n\\n\\n# ---------------------------------------------------------------------------\\n# Metrics helpers (shared by live comparison and offline report)\\n# ---------------------------------------------------------------------------\\n\\ndef _coerce_metrics(metrics: Any) -> Dict[str, Any]:\\n \"\"\"Normalize a stored metrics value into a plain dict.\\n\\n Handles both formats found in result files:\\n - dict: produced by --compare (asdict) and by the fixed --mode path\\n - str : legacy single-mode files that stored repr(AgentMetrics(...))\\n because json.dump used default=str\\n \"\"\"\\n if isinstance(metrics, dict):\\n return metrics\\n if isinstance(metrics, str) and metrics.startswith(\"AgentMetrics(\"):\\n # Safe eval: only AgentMetrics is exposed, no builtins.\\n try:\\n obj = eval(metrics, {\"__builtins__\": {}}, {\"AgentMetrics\": AgentMetrics})\\n return asdict(obj)\\n except Exception as e: # pragma: no cover - defensive\\n logger.warning(f\"Could not parse legacy metrics string: {e}\")\\n return {}\\n\\n\\ndef _avg_ttft(m: Dict[str, Any]) -> float:\\n \"\"\"Average TTFT across iterations, falling back to first-iteration TTFT.\"\"\"\\n lst = m.get(\"ttft_per_iteration\") or []\\n return sum(lst) / len(lst) if lst else float(m.get(\"ttft\", 0.0) or 0.0)\\n\\n\\ndef _hit_rate(m: Dict[str, Any]) -> float:\\n total = (m.get(\"cache_hits\", 0) or 0) + (m.get(\"cache_misses\", 0) or 0)\\n return (m.get(\"cache_hits\", 0) or 0) / total * 100 if total else 0.0\\n\\n\\ndef _billable_tokens(m: Dict[str, Any], cache_price_ratio: float) -> float:\\n \"\"\"Illustrative billable prompt tokens under a prompt-cache discount.\\n\\n cached tokens are charged at cache_price_ratio of the normal price; the\\n rest at full price. This is a transparent function of the *measured*\\n token counts and a user-supplied ratio - it is not a fabricated\\n provider-specific price.\\n \"\"\"\\n prompt = m.get(\"prompt_tokens\", 0) or 0\\n cached = m.get(\"cached_tokens\", 0) or 0\\n cached = min(cached, prompt)\\n return (prompt - cached) + cached * cache_price_ratio\\n\\n\\ndef print_comparison_table(results: Dict[str, Any], cache_price_ratio: float = 0.1) -> None:\\n \"\"\"Render the cross-strategy comparison table (latency / cache / cost).\"\"\"\\n print(f\"\\\\n{\\'Mode\\':<16} {\\'Iters\\':<6} {\\'1st TTFT\\':<10} {\\'Avg TTFT\\':<10} \"\\n f\"{\\'Total(s)\\':<10} {\\'Prompt\\':<9} {\\'Cached\\':<9} {\\'Hit%\\':<7} \"\\n f\"{\\'Cache%\\':<8} {\\'Bill.Tok\\':<10} {\\'Save%\\':<7}\")\\n print(\"-\" * 112)\\n\\n for mode, data in results.items():\\n m = _coerce_metrics(data.get(\"metrics\", {}))\\n prompt = m.get(\"prompt_tokens\", 0) or 0\\n cached = m.get(\"cached_tokens\", 0) or 0\\n iters = data.get(\"iterations\", m.get(\"iterations\", 0)) or 0\\n cache_pct = cached / prompt * 100 if prompt else 0.0\\n billable = _billable_tokens(m, cache_price_ratio)\\n save_pct = (prompt - billable) / prompt * 100 if prompt else 0.0\\n\\n print(f\"{mode:<16} {iters:<6} {float(m.get(\\'ttft\\', 0.0) or 0.0):<10.3f} \"\\n f\"{_avg_ttft(m):<10.3f} {float(m.get(\\'total_time\\', 0.0) or 0.0):<10.3f} \"\\n f\"{prompt:<9,} {cached:<9,} {_hit_rate(m):<7.1f} \"\\n f\"{cache_pct:<8.1f} {billable:<10,.0f} {save_pct:<7.1f}\")\\n\\n print(\"-\" * 112)\\n print(f\"\u6ce8\uff1aBill.Tok / Save% \u5047\u8bbe\u7f13\u5b58 token \u6309\u6b63\u5e38\u4ef7\u7684 {cache_price_ratio:.0%} \u8ba1\u8d39\"\\n f\"\uff08\u53ef\u7528 --cache-price-ratio \u8c03\u6574\uff09\uff0c\u4ec5\u4e3a\u6210\u672c\u793a\u610f\uff0c\u975e\u67d0\u5bb6\u670d\u52a1\u5546\u5b9e\u9645\u62a5\u4ef7\u3002\")\\n\\n\\ndef load_result_files(paths: List[str]) -> Dict[str, Any]:\\n \"\"\"Load result_*.json files into a {mode: {...}} dict for offline reporting.\"\"\"\\n results: Dict[str, Any] = {}\\n for path in sorted(paths):\\n try:\\n with open(path, \\'r\\') as f:\\n data = json.load(f)\\n except Exception as e:\\n logger.warning(f\"Skipping {path}: {e}\")\\n continue\\n\\n # A comparison_*.json holds many modes; a result_*.json holds one.\\n if \"mode\" not in data and all(isinstance(v, dict) and \"metrics\" in v\\n for v in data.values()):\\n for mode, entry in data.items():\\n results[mode] = {\"metrics\": _coerce_metrics(entry.get(\"metrics\", {})),\\n \"iterations\": entry.get(\"iterations\"),\\n \"_source\": path}\\n else:\\n mode = data.get(\"mode\", os.path.splitext(os.path.basename(path))[0])\\n results[mode] = {\"metrics\": _coerce_metrics(data.get(\"metrics\", {})),\\n \"iterations\": data.get(\"iterations\"),\\n \"_source\": path}\\n return results\\n\\n\\ndef run_report(inputs: List[str] = None, cache_price_ratio: float = 0.1) -> None:\\n \"\"\"Offline: build the comparison table from existing result_*.json files.\\n\\n No API key required - reads previously saved runs so the final result is\\n legible in one command without re-hitting the model.\\n \"\"\"\\n if not inputs:\\n inputs = [\"result_*.json\", \"comparison_*.json\"]\\n\\n paths: List[str] = []\\n for item in inputs:\\n if os.path.isdir(item):\\n paths.extend(glob.glob(os.path.join(item, \"result_*.json\")))\\n paths.extend(glob.glob(os.path.join(item, \"comparison_*.json\")))\\n else:\\n paths.extend(glob.glob(item))\\n\\n paths = sorted(set(paths))\\n if not paths:\\n logger.error(\"\u672a\u627e\u5230\u4efb\u4f55 result_*.json / comparison_*.json \u7ed3\u679c\u6587\u4ef6\u3002\"\\n \"\u8bf7\u5148\u8fd0\u884c --mode \u6216 --compare \u751f\u6210\u7ed3\u679c\uff0c\u6216\u7528 --input \u6307\u5b9a\u8def\u5f84\u3002\")\\n sys.exit(1)\\n\\n results = load_result_files(paths)\\n\\n print(\"\\\\n\" + \"=\" * 112)\\n print(\"KV CACHE \u79bb\u7ebf\u5bf9\u6bd4\u62a5\u544a\uff08\u57fa\u4e8e\u5df2\u4fdd\u5b58\u7684\u5b9e\u6d4b\u7ed3\u679c\uff09\")\\n print(\"=\" * 112)\\n print(f\"\u6570\u636e\u6765\u6e90\uff08{len(paths)} \u4e2a\u6587\u4ef6\uff09:\")\\n for mode, data in results.items():\\n print(f\" \u2022 {mode:<16} \u2190 {os.path.basename(data.get(\\'_source\\', \\'?\\'))}\")\\n\\n print_comparison_table(results, cache_price_ratio)\\n\\n print(\"\\\\n\ud83d\udcdd \u8bf4\u660e\uff1a\u4e0d\u540c\u7ed3\u679c\u6587\u4ef6\u53ef\u80fd\u6765\u81ea\u4e0d\u540c\u4efb\u52a1/\u65f6\u95f4\uff0c\u7edd\u5bf9\u6570\u503c\u4ec5\u4f9b\u540c\u4e00\u6b21\u8fd0\u884c\u5185\u6a2a\u5411\u5bf9\u6bd4\uff1b\"\\n \"\u5982\u9700\u4e25\u683c\u5bf9\u7167\uff0c\u8bf7\u7528 --compare \u5728\u540c\u4e00\u4efb\u52a1\u4e0b\u4e00\u6b21\u6027\u751f\u6210\u5168\u90e8\u6a21\u5f0f\u7684\u6570\u636e\u3002\")\\n\\n\\ndef create_summary_task() -> str:\\n \"\"\"Create a task that requires reading multiple files\"\"\"\\n return \"\"\"Please analyze and summarize all the projects in the week1 and week2 directories.\\nFor each project:\\n1. Find all Python files\\n2. Read the main files and understand the functionality\\n3. Identify the key features and purpose\\n4. Provide a comprehensive summary\\n\\nStart with week1 projects, then move to week2. Be thorough in your analysis.\"\"\"\\n\\n\\ndef run_single_mode(api_key: str, mode: str, task: str = None, root_dir: str = \"../..\",\\n model: str = DEFAULT_MODEL, output: str = None):\\n \"\"\"\\n Run agent in a single mode\\n\\n Args:\\n api_key: API key for Kimi\\n mode: KV cache mode to use\\n task: Custom task (optional)\\n root_dir: Root directory for file operations (default: \"../..\" = /projects from kv-cache dir)\\n model: Model to use\\n output: Output path for the result JSON (optional; auto-named if omitted)\\n \"\"\"\\n # Parse mode\\n mode_map = {\\n \"correct\": KVCacheMode.CORRECT,\\n \"dynamic_system\": KVCacheMode.DYNAMIC_SYSTEM,\\n \"shuffled_tools\": KVCacheMode.SHUFFLED_TOOLS,\\n \"dynamic_profile\": KVCacheMode.DYNAMIC_PROFILE,\\n \"sliding_window\": KVCacheMode.SLIDING_WINDOW,\\n \"text_format\": KVCacheMode.TEXT_FORMAT\\n }\\n \\n if mode not in mode_map:\\n logger.error(f\"Invalid mode: {mode}\")\\n logger.info(f\"Valid modes: {\\', \\'.join(mode_map.keys())}\")\\n return\\n \\n # Use default task if not provided\\n if not task:\\n task = create_summary_task()\\n \\n logger.info(f\"Running in mode: {mode}\")\\n logger.info(f\"Task: {task}\")\\n logger.info(\"=\"*80)\\n \\n # Create agent and execute task\\n agent = KVCacheAgent(\\n api_key=api_key,\\n mode=mode_map[mode],\\n model=model,\\n root_dir=root_dir,\\n verbose=True\\n )\\n \\n result = agent.execute_task(task, max_iterations=30)\\n \\n # Print results\\n print(\"\\\\n\" + \"=\"*80)\\n print(f\"EXECUTION RESULTS - Mode: {mode}\")\\n print(\"=\"*80)\\n \\n metrics = result[\"metrics\"]\\n print(f\"\\\\n\ud83d\udcca Performance Metrics:\")\\n print(f\" \u2022 Time to First Token (TTFT): {metrics.ttft:.3f} seconds\")\\n \\n # Show TTFT progression\\n if metrics.ttft_per_iteration:\\n print(f\" \u2022 TTFT per iteration:\")\\n for i, ttft in enumerate(metrics.ttft_per_iteration, 1):\\n print(f\" Iteration {i}: {ttft:.3f}s\")\\n\\n # Show improvement\\n if len(metrics.ttft_per_iteration) > 1:\\n first_ttft = metrics.ttft_per_iteration[0]\\n last_ttft = metrics.ttft_per_iteration[-1]\\n avg_after_first = sum(metrics.ttft_per_iteration[1:]) / len(metrics.ttft_per_iteration[1:])\\n print(f\" \u2022 TTFT Analysis:\")\\n print(f\" First iteration: {first_ttft:.3f}s\")\\n print(f\" Last iteration: {', 'total_lines': 537, 'lines_read': 537, 'offset': 0, 'end_line': 537, 'truncated': True, 'success': True}, error=None, timestamp=1784337614.781182)",
"ToolCall(name='read_file', arguments={'file_path': 'agent.py'}, result={'path': 'agent.py', 'content': '\"\"\"\\nKV Cache Demonstration Agent with ReAct Pattern\\nDemonstrates the importance of KV cache through correct and incorrect implementations.\\nUses local file system tools to read and search through code files.\\n\"\"\"\\n\\nimport json\\nimport os\\nimport re\\nimport time\\nimport logging\\nimport random\\nfrom typing import List, Dict, Any, Optional, Tuple\\nfrom dataclasses import dataclass, field, asdict\\nfrom enum import Enum\\nfrom datetime import datetime\\nfrom openai import OpenAI\\nimport glob as glob_module\\nimport subprocess\\n\\n\\ndef _is_reasoning_model(model) -> bool:\\n \"\"\"True for models that emit reasoning_content and only accept temperature=1.\\n\\n On the live Moonshot endpoint the whole current Kimi family reasons:\\n kimi-k2.5 / kimi-k2.6 / kimi-k2.7* / kimi-k3. The legacy moonshot-v1-*\\n chat models do NOT reason (and also do not report cached_tokens).\"\"\"\\n m = str(model or \"\").lower().replace(\"/\", \"-\")\\n if \"gpt-5\" in m:\\n return True\\n return any(tag in m for tag in (\"kimi-k2.5\", \"kimi-k2.6\", \"kimi-k2.7\", \"kimi-k3\"))\\n\\n\\ndef _reasoning_safe_temperature(model, requested=1.0):\\n \"\"\"Reasoning models (Kimi K2.5/K2.6/K2.7/K3, GPT-5, ...) only accept\\n temperature=1. Return 1 for those; otherwise the requested value so\\n non-reasoning providers (moonshot-v1, Doubao, DeepSeek) are unchanged.\"\"\"\\n return 1 if _is_reasoning_model(model) else requested\\n\\n\\ndef _reasoning_safe_max_tokens(model, requested=2000):\\n \"\"\"Reasoning models spend completion budget on hidden reasoning tokens\\n before emitting content / tool calls. Give them enough headroom so a\\n tool call is not truncated away; leave non-reasoning models unchanged.\"\"\"\\n return max(requested, 4096) if _is_reasoning_model(model) else requested\\n\\n\\n# Configure logging\\nlogging.basicConfig(level=logging.INFO, format=\\'%(asctime)s - %(levelname)s - %(message)s\\')\\nlogger = logging.getLogger(__name__)\\n\\n\\nclass KVCacheMode(Enum):\\n \"\"\"Different KV cache optimization modes\"\"\"\\n CORRECT = \"correct\" # Correct implementation with stable context\\n DYNAMIC_SYSTEM = \"dynamic_system\" # Changing system prompt with timestamp\\n SHUFFLED_TOOLS = \"shuffled_tools\" # Shuffling tool order each request\\n DYNAMIC_PROFILE = \"dynamic_profile\" # Changing user profile with credits\\n SLIDING_WINDOW = \"sliding_window\" # Only keeping recent 5 messages\\n TEXT_FORMAT = \"text_format\" # Formatting messages as plain text\\n\\n\\n@dataclass\\nclass ToolCall:\\n \"\"\"Represents a single tool call\"\"\"\\n name: str\\n arguments: Dict[str, Any]\\n result: Any = None\\n error: Optional[str] = None\\n timestamp: float = field(default_factory=time.time)\\n\\n\\n@dataclass\\nclass AgentMetrics:\\n \"\"\"Metrics for agent performance\"\"\"\\n ttft: float = 0.0 # Time to first token (first iteration)\\n ttft_per_iteration: List[float] = field(default_factory=list) # TTFT for each iteration\\n total_time: float = 0.0\\n iterations: int = 0\\n tool_calls: int = 0\\n cache_hits: int = 0\\n cache_misses: int = 0\\n prompt_tokens: int = 0\\n completion_tokens: int = 0\\n cached_tokens: int = 0\\n\\n\\nclass LocalFileTools:\\n \"\"\"Local implementations of file system tools\"\"\"\\n \\n def __init__(self, root_dir: str = \".\"):\\n self.root_dir = os.path.abspath(root_dir)\\n logger.info(f\"File tools initialized with root: {self.root_dir}\")\\n \\n def read_file(self, file_path: str, offset: int = 0, size: int = None) -> Dict[str, Any]:\\n \"\"\"\\n Read contents of a file\\n \\n Args:\\n file_path: Path to the file relative to root directory\\n offset: Line number to start reading from (0-based, default: 0)\\n size: Number of lines to read (default: None, read all)\\n \\n Returns:\\n Dictionary with file contents or error\\n \"\"\"\\n try:\\n full_path = os.path.join(self.root_dir, file_path)\\n \\n # Security check - ensure path is within root_dir\\n real_path = os.path.realpath(full_path)\\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n with open(real_path, \\'r\\', encoding=\\'utf-8\\', errors=\\'ignore\\') as f:\\n lines = f.readlines()\\n \\n total_lines = len(lines)\\n \\n # Apply offset and size\\n if offset < 0:\\n offset = 0\\n if offset >= total_lines:\\n return {\\n \"path\": file_path,\\n \"content\": \"\",\\n \"total_lines\": total_lines,\\n \"lines_read\": 0,\\n \"offset\": offset,\\n \"success\": True,\\n \"message\": f\"Offset {offset} exceeds file length ({total_lines} lines)\"\\n }\\n \\n # Determine end line\\n if size is None:\\n end = total_lines\\n else:\\n end = min(offset + size, total_lines)\\n \\n # Get the requested lines\\n selected_lines = lines[offset:end]\\n content = \\'\\'.join(selected_lines)\\n \\n # Apply size limit for safety (10KB)\\n truncated = False\\n if len(content) > 10000:\\n content = content[:10000]\\n truncated = True\\n \\n return {\\n \"path\": file_path,\\n \"content\": content,\\n \"total_lines\": total_lines,\\n \"lines_read\": len(selected_lines),\\n \"offset\": offset,\\n \"end_line\": end,\\n \"truncated\": truncated,\\n \"success\": True\\n }\\n except FileNotFoundError:\\n return {\\n \"error\": f\"File not found: {file_path}\",\\n \"success\": False\\n }\\n except Exception as e:\\n return {\\n \"error\": f\"Error reading file: {str(e)}\",\\n \"success\": False\\n }\\n \\n def find(self, pattern: str = \"*\", directory: str = \".\") -> Dict[str, Any]:\\n \"\"\"\\n Find files matching a pattern (similar to Unix find command)\\n \\n Args:\\n pattern: File name pattern (supports wildcards, default: \"*\" for all files)\\n directory: Directory to search in (relative to root_dir)\\n \\n Returns:\\n Dictionary with list of matching files\\n \"\"\"\\n try:\\n # Handle directory path properly\\n if directory == \".\":\\n search_dir = self.root_dir\\n else:\\n # Remove leading/trailing slashes for consistency\\n directory = directory.strip(\\'/\\')\\n search_dir = os.path.join(self.root_dir, directory)\\n \\n # Security check\\n real_path = os.path.realpath(search_dir)\\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n # Check if directory exists\\n if not os.path.exists(real_path):\\n return {\\n \"error\": f\"Directory not found: {directory}\",\\n \"success\": False\\n }\\n \\n # Use glob to find matching files\\n matches = []\\n for root, dirs, files in os.walk(real_path):\\n # Filter hidden directories and __pycache__\\n dirs[:] = [d for d in dirs if not d.startswith(\\'.\\') and d != \\'__pycache__\\']\\n \\n for file in files:\\n # Skip hidden files and .pyc files\\n if file.startswith(\\'.\\') or file.endswith(\\'.pyc\\'):\\n continue\\n \\n if glob_module.fnmatch.fnmatch(file, pattern):\\n # Get path relative to root_dir (not search_dir)\\n full_path = os.path.join(root, file)\\n rel_path = os.path.relpath(full_path, self.root_dir)\\n matches.append(rel_path)\\n \\n # Sort for consistency\\n matches.sort()\\n \\n # Limit results for demonstration\\n if len(matches) > 100:\\n matches = matches[:100]\\n truncated = True\\n else:\\n truncated = False\\n \\n return {\\n \"pattern\": pattern,\\n \"directory\": directory,\\n \"matches\": matches,\\n \"count\": len(matches),\\n \"truncated\": truncated,\\n \"success\": True\\n }\\n except Exception as e:\\n return {\\n \"error\": f\"Error finding files: {str(e)}\",\\n \"success\": False\\n }\\n \\n def grep(self, pattern: str, file_path: str = None, directory: str = None) -> Dict[str, Any]:\\n \"\"\"\\n Search for pattern in files (similar to Unix grep command)\\n \\n Args:\\n pattern: Regular expression pattern to search for\\n file_path: Single file to search in (optional)\\n directory: Directory to search in (optional)\\n \\n Returns:\\n Dictionary with matching lines\\n \"\"\"\\n try:\\n matches = []\\n files_searched = []\\n \\n if file_path:\\n # Search in single file\\n full_path = os.path.join(self.root_dir, file_path)\\n real_path = os.path.realpath(full_path)\\n \\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n ', 'total_lines': 869, 'lines_read': 869, 'offset': 0, 'end_line': 869, 'truncated': True, 'success': True}, error=None, timestamp=1784337614.781894)"
],
"metrics": {
"ttft": 2.2344162464141846,
"ttft_per_iteration": [
2.2344162464141846,
1.985018014907837,
3.9584708213806152,
6.417687892913818
],
"total_time": 14.704215049743652,
"iterations": 5,
"tool_calls": 5,
"cache_hits": 4,
"cache_misses": 0,
"prompt_tokens": 2224,
"completion_tokens": 241,
"cached_tokens": 1510
},
"mode": "sliding_window",
"model": "kimi-k2.6",
"task": "Find all Python files in this directory; read main.py and agent.py; summarize in 3 sentences."
}
result_text_format_20260718_kimi_k2_6.json¶
{
"success": true,
"iterations": 3,
"tool_calls": [
"ToolCall(name='find', arguments={'pattern': '*.py'}, result={'pattern': '*.py', 'directory': '.', 'matches': ['agent.py', 'demo_quick.py', 'main.py', 'test_api.py', 'test_cache_invalidation.py', 'test_cached_tokens.py', 'test_completion.py', 'test_error_handling.py', 'test_file_range.py', 'test_interactive.py', 'test_message_flow.py', 'test_tools.py', 'test_ttft.py'], 'count': 13, 'truncated': False, 'success': True}, error=None, timestamp=1784337621.09088)",
"ToolCall(name='read_file', arguments={'file_path': 'main.py'}, result={'path': 'main.py', 'content': '\"\"\"\\nMain script to demonstrate KV cache importance\\nRuns the ReAct agent with different implementations and compares performance\\n\"\"\"\\n\\nimport os\\nimport sys\\nimport glob\\nimport json\\nimport argparse\\nimport logging\\nfrom typing import Dict, List, Any\\nfrom datetime import datetime\\nfrom dataclasses import asdict\\nfrom agent import KVCacheAgent, KVCacheMode, AgentMetrics, compare_implementations\\n\\n# Default model (Moonshot / Kimi). The whole current Kimi family (k2.5/k2.6/\\n# k2.7/k3) reports cached_tokens for automatic prefix caching AND reasons, so it\\n# only accepts temperature=1 (agent.py handles that automatically). kimi-k2.6 has\\n# the lightest reasoning footprint of the cache-reporting models, giving the\\n# cleanest TTFT while still exposing the prefix-cache hit metric this demo needs.\\n# (The non-reasoning moonshot-v1-* models do NOT report cached_tokens, so they\\n# cannot demonstrate the cache effect.)\\nDEFAULT_MODEL = \"kimi-k2.6\"\\n\\n# Configure logging\\nlogging.basicConfig(\\n level=logging.INFO,\\n format=\\'%(asctime)s - %(levelname)s - %(message)s\\',\\n handlers=[\\n logging.FileHandler(\\'kv_cache_demo.log\\'),\\n logging.StreamHandler()\\n ]\\n)\\nlogger = logging.getLogger(__name__)\\n\\n\\n# ---------------------------------------------------------------------------\\n# Metrics helpers (shared by live comparison and offline report)\\n# ---------------------------------------------------------------------------\\n\\ndef _coerce_metrics(metrics: Any) -> Dict[str, Any]:\\n \"\"\"Normalize a stored metrics value into a plain dict.\\n\\n Handles both formats found in result files:\\n - dict: produced by --compare (asdict) and by the fixed --mode path\\n - str : legacy single-mode files that stored repr(AgentMetrics(...))\\n because json.dump used default=str\\n \"\"\"\\n if isinstance(metrics, dict):\\n return metrics\\n if isinstance(metrics, str) and metrics.startswith(\"AgentMetrics(\"):\\n # Safe eval: only AgentMetrics is exposed, no builtins.\\n try:\\n obj = eval(metrics, {\"__builtins__\": {}}, {\"AgentMetrics\": AgentMetrics})\\n return asdict(obj)\\n except Exception as e: # pragma: no cover - defensive\\n logger.warning(f\"Could not parse legacy metrics string: {e}\")\\n return {}\\n\\n\\ndef _avg_ttft(m: Dict[str, Any]) -> float:\\n \"\"\"Average TTFT across iterations, falling back to first-iteration TTFT.\"\"\"\\n lst = m.get(\"ttft_per_iteration\") or []\\n return sum(lst) / len(lst) if lst else float(m.get(\"ttft\", 0.0) or 0.0)\\n\\n\\ndef _hit_rate(m: Dict[str, Any]) -> float:\\n total = (m.get(\"cache_hits\", 0) or 0) + (m.get(\"cache_misses\", 0) or 0)\\n return (m.get(\"cache_hits\", 0) or 0) / total * 100 if total else 0.0\\n\\n\\ndef _billable_tokens(m: Dict[str, Any], cache_price_ratio: float) -> float:\\n \"\"\"Illustrative billable prompt tokens under a prompt-cache discount.\\n\\n cached tokens are charged at cache_price_ratio of the normal price; the\\n rest at full price. This is a transparent function of the *measured*\\n token counts and a user-supplied ratio - it is not a fabricated\\n provider-specific price.\\n \"\"\"\\n prompt = m.get(\"prompt_tokens\", 0) or 0\\n cached = m.get(\"cached_tokens\", 0) or 0\\n cached = min(cached, prompt)\\n return (prompt - cached) + cached * cache_price_ratio\\n\\n\\ndef print_comparison_table(results: Dict[str, Any], cache_price_ratio: float = 0.1) -> None:\\n \"\"\"Render the cross-strategy comparison table (latency / cache / cost).\"\"\"\\n print(f\"\\\\n{\\'Mode\\':<16} {\\'Iters\\':<6} {\\'1st TTFT\\':<10} {\\'Avg TTFT\\':<10} \"\\n f\"{\\'Total(s)\\':<10} {\\'Prompt\\':<9} {\\'Cached\\':<9} {\\'Hit%\\':<7} \"\\n f\"{\\'Cache%\\':<8} {\\'Bill.Tok\\':<10} {\\'Save%\\':<7}\")\\n print(\"-\" * 112)\\n\\n for mode, data in results.items():\\n m = _coerce_metrics(data.get(\"metrics\", {}))\\n prompt = m.get(\"prompt_tokens\", 0) or 0\\n cached = m.get(\"cached_tokens\", 0) or 0\\n iters = data.get(\"iterations\", m.get(\"iterations\", 0)) or 0\\n cache_pct = cached / prompt * 100 if prompt else 0.0\\n billable = _billable_tokens(m, cache_price_ratio)\\n save_pct = (prompt - billable) / prompt * 100 if prompt else 0.0\\n\\n print(f\"{mode:<16} {iters:<6} {float(m.get(\\'ttft\\', 0.0) or 0.0):<10.3f} \"\\n f\"{_avg_ttft(m):<10.3f} {float(m.get(\\'total_time\\', 0.0) or 0.0):<10.3f} \"\\n f\"{prompt:<9,} {cached:<9,} {_hit_rate(m):<7.1f} \"\\n f\"{cache_pct:<8.1f} {billable:<10,.0f} {save_pct:<7.1f}\")\\n\\n print(\"-\" * 112)\\n print(f\"\u6ce8\uff1aBill.Tok / Save% \u5047\u8bbe\u7f13\u5b58 token \u6309\u6b63\u5e38\u4ef7\u7684 {cache_price_ratio:.0%} \u8ba1\u8d39\"\\n f\"\uff08\u53ef\u7528 --cache-price-ratio \u8c03\u6574\uff09\uff0c\u4ec5\u4e3a\u6210\u672c\u793a\u610f\uff0c\u975e\u67d0\u5bb6\u670d\u52a1\u5546\u5b9e\u9645\u62a5\u4ef7\u3002\")\\n\\n\\ndef load_result_files(paths: List[str]) -> Dict[str, Any]:\\n \"\"\"Load result_*.json files into a {mode: {...}} dict for offline reporting.\"\"\"\\n results: Dict[str, Any] = {}\\n for path in sorted(paths):\\n try:\\n with open(path, \\'r\\') as f:\\n data = json.load(f)\\n except Exception as e:\\n logger.warning(f\"Skipping {path}: {e}\")\\n continue\\n\\n # A comparison_*.json holds many modes; a result_*.json holds one.\\n if \"mode\" not in data and all(isinstance(v, dict) and \"metrics\" in v\\n for v in data.values()):\\n for mode, entry in data.items():\\n results[mode] = {\"metrics\": _coerce_metrics(entry.get(\"metrics\", {})),\\n \"iterations\": entry.get(\"iterations\"),\\n \"_source\": path}\\n else:\\n mode = data.get(\"mode\", os.path.splitext(os.path.basename(path))[0])\\n results[mode] = {\"metrics\": _coerce_metrics(data.get(\"metrics\", {})),\\n \"iterations\": data.get(\"iterations\"),\\n \"_source\": path}\\n return results\\n\\n\\ndef run_report(inputs: List[str] = None, cache_price_ratio: float = 0.1) -> None:\\n \"\"\"Offline: build the comparison table from existing result_*.json files.\\n\\n No API key required - reads previously saved runs so the final result is\\n legible in one command without re-hitting the model.\\n \"\"\"\\n if not inputs:\\n inputs = [\"result_*.json\", \"comparison_*.json\"]\\n\\n paths: List[str] = []\\n for item in inputs:\\n if os.path.isdir(item):\\n paths.extend(glob.glob(os.path.join(item, \"result_*.json\")))\\n paths.extend(glob.glob(os.path.join(item, \"comparison_*.json\")))\\n else:\\n paths.extend(glob.glob(item))\\n\\n paths = sorted(set(paths))\\n if not paths:\\n logger.error(\"\u672a\u627e\u5230\u4efb\u4f55 result_*.json / comparison_*.json \u7ed3\u679c\u6587\u4ef6\u3002\"\\n \"\u8bf7\u5148\u8fd0\u884c --mode \u6216 --compare \u751f\u6210\u7ed3\u679c\uff0c\u6216\u7528 --input \u6307\u5b9a\u8def\u5f84\u3002\")\\n sys.exit(1)\\n\\n results = load_result_files(paths)\\n\\n print(\"\\\\n\" + \"=\" * 112)\\n print(\"KV CACHE \u79bb\u7ebf\u5bf9\u6bd4\u62a5\u544a\uff08\u57fa\u4e8e\u5df2\u4fdd\u5b58\u7684\u5b9e\u6d4b\u7ed3\u679c\uff09\")\\n print(\"=\" * 112)\\n print(f\"\u6570\u636e\u6765\u6e90\uff08{len(paths)} \u4e2a\u6587\u4ef6\uff09:\")\\n for mode, data in results.items():\\n print(f\" \u2022 {mode:<16} \u2190 {os.path.basename(data.get(\\'_source\\', \\'?\\'))}\")\\n\\n print_comparison_table(results, cache_price_ratio)\\n\\n print(\"\\\\n\ud83d\udcdd \u8bf4\u660e\uff1a\u4e0d\u540c\u7ed3\u679c\u6587\u4ef6\u53ef\u80fd\u6765\u81ea\u4e0d\u540c\u4efb\u52a1/\u65f6\u95f4\uff0c\u7edd\u5bf9\u6570\u503c\u4ec5\u4f9b\u540c\u4e00\u6b21\u8fd0\u884c\u5185\u6a2a\u5411\u5bf9\u6bd4\uff1b\"\\n \"\u5982\u9700\u4e25\u683c\u5bf9\u7167\uff0c\u8bf7\u7528 --compare \u5728\u540c\u4e00\u4efb\u52a1\u4e0b\u4e00\u6b21\u6027\u751f\u6210\u5168\u90e8\u6a21\u5f0f\u7684\u6570\u636e\u3002\")\\n\\n\\ndef create_summary_task() -> str:\\n \"\"\"Create a task that requires reading multiple files\"\"\"\\n return \"\"\"Please analyze and summarize all the projects in the week1 and week2 directories.\\nFor each project:\\n1. Find all Python files\\n2. Read the main files and understand the functionality\\n3. Identify the key features and purpose\\n4. Provide a comprehensive summary\\n\\nStart with week1 projects, then move to week2. Be thorough in your analysis.\"\"\"\\n\\n\\ndef run_single_mode(api_key: str, mode: str, task: str = None, root_dir: str = \"../..\",\\n model: str = DEFAULT_MODEL, output: str = None):\\n \"\"\"\\n Run agent in a single mode\\n\\n Args:\\n api_key: API key for Kimi\\n mode: KV cache mode to use\\n task: Custom task (optional)\\n root_dir: Root directory for file operations (default: \"../..\" = /projects from kv-cache dir)\\n model: Model to use\\n output: Output path for the result JSON (optional; auto-named if omitted)\\n \"\"\"\\n # Parse mode\\n mode_map = {\\n \"correct\": KVCacheMode.CORRECT,\\n \"dynamic_system\": KVCacheMode.DYNAMIC_SYSTEM,\\n \"shuffled_tools\": KVCacheMode.SHUFFLED_TOOLS,\\n \"dynamic_profile\": KVCacheMode.DYNAMIC_PROFILE,\\n \"sliding_window\": KVCacheMode.SLIDING_WINDOW,\\n \"text_format\": KVCacheMode.TEXT_FORMAT\\n }\\n \\n if mode not in mode_map:\\n logger.error(f\"Invalid mode: {mode}\")\\n logger.info(f\"Valid modes: {\\', \\'.join(mode_map.keys())}\")\\n return\\n \\n # Use default task if not provided\\n if not task:\\n task = create_summary_task()\\n \\n logger.info(f\"Running in mode: {mode}\")\\n logger.info(f\"Task: {task}\")\\n logger.info(\"=\"*80)\\n \\n # Create agent and execute task\\n agent = KVCacheAgent(\\n api_key=api_key,\\n mode=mode_map[mode],\\n model=model,\\n root_dir=root_dir,\\n verbose=True\\n )\\n \\n result = agent.execute_task(task, max_iterations=30)\\n \\n # Print results\\n print(\"\\\\n\" + \"=\"*80)\\n print(f\"EXECUTION RESULTS - Mode: {mode}\")\\n print(\"=\"*80)\\n \\n metrics = result[\"metrics\"]\\n print(f\"\\\\n\ud83d\udcca Performance Metrics:\")\\n print(f\" \u2022 Time to First Token (TTFT): {metrics.ttft:.3f} seconds\")\\n \\n # Show TTFT progression\\n if metrics.ttft_per_iteration:\\n print(f\" \u2022 TTFT per iteration:\")\\n for i, ttft in enumerate(metrics.ttft_per_iteration, 1):\\n print(f\" Iteration {i}: {ttft:.3f}s\")\\n\\n # Show improvement\\n if len(metrics.ttft_per_iteration) > 1:\\n first_ttft = metrics.ttft_per_iteration[0]\\n last_ttft = metrics.ttft_per_iteration[-1]\\n avg_after_first = sum(metrics.ttft_per_iteration[1:]) / len(metrics.ttft_per_iteration[1:])\\n print(f\" \u2022 TTFT Analysis:\")\\n print(f\" First iteration: {first_ttft:.3f}s\")\\n print(f\" Last iteration: {', 'total_lines': 537, 'lines_read': 537, 'offset': 0, 'end_line': 537, 'truncated': True, 'success': True}, error=None, timestamp=1784337629.322158)",
"ToolCall(name='read_file', arguments={'file_path': 'agent.py'}, result={'path': 'agent.py', 'content': '\"\"\"\\nKV Cache Demonstration Agent with ReAct Pattern\\nDemonstrates the importance of KV cache through correct and incorrect implementations.\\nUses local file system tools to read and search through code files.\\n\"\"\"\\n\\nimport json\\nimport os\\nimport re\\nimport time\\nimport logging\\nimport random\\nfrom typing import List, Dict, Any, Optional, Tuple\\nfrom dataclasses import dataclass, field, asdict\\nfrom enum import Enum\\nfrom datetime import datetime\\nfrom openai import OpenAI\\nimport glob as glob_module\\nimport subprocess\\n\\n\\ndef _is_reasoning_model(model) -> bool:\\n \"\"\"True for models that emit reasoning_content and only accept temperature=1.\\n\\n On the live Moonshot endpoint the whole current Kimi family reasons:\\n kimi-k2.5 / kimi-k2.6 / kimi-k2.7* / kimi-k3. The legacy moonshot-v1-*\\n chat models do NOT reason (and also do not report cached_tokens).\"\"\"\\n m = str(model or \"\").lower().replace(\"/\", \"-\")\\n if \"gpt-5\" in m:\\n return True\\n return any(tag in m for tag in (\"kimi-k2.5\", \"kimi-k2.6\", \"kimi-k2.7\", \"kimi-k3\"))\\n\\n\\ndef _reasoning_safe_temperature(model, requested=1.0):\\n \"\"\"Reasoning models (Kimi K2.5/K2.6/K2.7/K3, GPT-5, ...) only accept\\n temperature=1. Return 1 for those; otherwise the requested value so\\n non-reasoning providers (moonshot-v1, Doubao, DeepSeek) are unchanged.\"\"\"\\n return 1 if _is_reasoning_model(model) else requested\\n\\n\\ndef _reasoning_safe_max_tokens(model, requested=2000):\\n \"\"\"Reasoning models spend completion budget on hidden reasoning tokens\\n before emitting content / tool calls. Give them enough headroom so a\\n tool call is not truncated away; leave non-reasoning models unchanged.\"\"\"\\n return max(requested, 4096) if _is_reasoning_model(model) else requested\\n\\n\\n# Configure logging\\nlogging.basicConfig(level=logging.INFO, format=\\'%(asctime)s - %(levelname)s - %(message)s\\')\\nlogger = logging.getLogger(__name__)\\n\\n\\nclass KVCacheMode(Enum):\\n \"\"\"Different KV cache optimization modes\"\"\"\\n CORRECT = \"correct\" # Correct implementation with stable context\\n DYNAMIC_SYSTEM = \"dynamic_system\" # Changing system prompt with timestamp\\n SHUFFLED_TOOLS = \"shuffled_tools\" # Shuffling tool order each request\\n DYNAMIC_PROFILE = \"dynamic_profile\" # Changing user profile with credits\\n SLIDING_WINDOW = \"sliding_window\" # Only keeping recent 5 messages\\n TEXT_FORMAT = \"text_format\" # Formatting messages as plain text\\n\\n\\n@dataclass\\nclass ToolCall:\\n \"\"\"Represents a single tool call\"\"\"\\n name: str\\n arguments: Dict[str, Any]\\n result: Any = None\\n error: Optional[str] = None\\n timestamp: float = field(default_factory=time.time)\\n\\n\\n@dataclass\\nclass AgentMetrics:\\n \"\"\"Metrics for agent performance\"\"\"\\n ttft: float = 0.0 # Time to first token (first iteration)\\n ttft_per_iteration: List[float] = field(default_factory=list) # TTFT for each iteration\\n total_time: float = 0.0\\n iterations: int = 0\\n tool_calls: int = 0\\n cache_hits: int = 0\\n cache_misses: int = 0\\n prompt_tokens: int = 0\\n completion_tokens: int = 0\\n cached_tokens: int = 0\\n\\n\\nclass LocalFileTools:\\n \"\"\"Local implementations of file system tools\"\"\"\\n \\n def __init__(self, root_dir: str = \".\"):\\n self.root_dir = os.path.abspath(root_dir)\\n logger.info(f\"File tools initialized with root: {self.root_dir}\")\\n \\n def read_file(self, file_path: str, offset: int = 0, size: int = None) -> Dict[str, Any]:\\n \"\"\"\\n Read contents of a file\\n \\n Args:\\n file_path: Path to the file relative to root directory\\n offset: Line number to start reading from (0-based, default: 0)\\n size: Number of lines to read (default: None, read all)\\n \\n Returns:\\n Dictionary with file contents or error\\n \"\"\"\\n try:\\n full_path = os.path.join(self.root_dir, file_path)\\n \\n # Security check - ensure path is within root_dir\\n real_path = os.path.realpath(full_path)\\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n with open(real_path, \\'r\\', encoding=\\'utf-8\\', errors=\\'ignore\\') as f:\\n lines = f.readlines()\\n \\n total_lines = len(lines)\\n \\n # Apply offset and size\\n if offset < 0:\\n offset = 0\\n if offset >= total_lines:\\n return {\\n \"path\": file_path,\\n \"content\": \"\",\\n \"total_lines\": total_lines,\\n \"lines_read\": 0,\\n \"offset\": offset,\\n \"success\": True,\\n \"message\": f\"Offset {offset} exceeds file length ({total_lines} lines)\"\\n }\\n \\n # Determine end line\\n if size is None:\\n end = total_lines\\n else:\\n end = min(offset + size, total_lines)\\n \\n # Get the requested lines\\n selected_lines = lines[offset:end]\\n content = \\'\\'.join(selected_lines)\\n \\n # Apply size limit for safety (10KB)\\n truncated = False\\n if len(content) > 10000:\\n content = content[:10000]\\n truncated = True\\n \\n return {\\n \"path\": file_path,\\n \"content\": content,\\n \"total_lines\": total_lines,\\n \"lines_read\": len(selected_lines),\\n \"offset\": offset,\\n \"end_line\": end,\\n \"truncated\": truncated,\\n \"success\": True\\n }\\n except FileNotFoundError:\\n return {\\n \"error\": f\"File not found: {file_path}\",\\n \"success\": False\\n }\\n except Exception as e:\\n return {\\n \"error\": f\"Error reading file: {str(e)}\",\\n \"success\": False\\n }\\n \\n def find(self, pattern: str = \"*\", directory: str = \".\") -> Dict[str, Any]:\\n \"\"\"\\n Find files matching a pattern (similar to Unix find command)\\n \\n Args:\\n pattern: File name pattern (supports wildcards, default: \"*\" for all files)\\n directory: Directory to search in (relative to root_dir)\\n \\n Returns:\\n Dictionary with list of matching files\\n \"\"\"\\n try:\\n # Handle directory path properly\\n if directory == \".\":\\n search_dir = self.root_dir\\n else:\\n # Remove leading/trailing slashes for consistency\\n directory = directory.strip(\\'/\\')\\n search_dir = os.path.join(self.root_dir, directory)\\n \\n # Security check\\n real_path = os.path.realpath(search_dir)\\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n # Check if directory exists\\n if not os.path.exists(real_path):\\n return {\\n \"error\": f\"Directory not found: {directory}\",\\n \"success\": False\\n }\\n \\n # Use glob to find matching files\\n matches = []\\n for root, dirs, files in os.walk(real_path):\\n # Filter hidden directories and __pycache__\\n dirs[:] = [d for d in dirs if not d.startswith(\\'.\\') and d != \\'__pycache__\\']\\n \\n for file in files:\\n # Skip hidden files and .pyc files\\n if file.startswith(\\'.\\') or file.endswith(\\'.pyc\\'):\\n continue\\n \\n if glob_module.fnmatch.fnmatch(file, pattern):\\n # Get path relative to root_dir (not search_dir)\\n full_path = os.path.join(root, file)\\n rel_path = os.path.relpath(full_path, self.root_dir)\\n matches.append(rel_path)\\n \\n # Sort for consistency\\n matches.sort()\\n \\n # Limit results for demonstration\\n if len(matches) > 100:\\n matches = matches[:100]\\n truncated = True\\n else:\\n truncated = False\\n \\n return {\\n \"pattern\": pattern,\\n \"directory\": directory,\\n \"matches\": matches,\\n \"count\": len(matches),\\n \"truncated\": truncated,\\n \"success\": True\\n }\\n except Exception as e:\\n return {\\n \"error\": f\"Error finding files: {str(e)}\",\\n \"success\": False\\n }\\n \\n def grep(self, pattern: str, file_path: str = None, directory: str = None) -> Dict[str, Any]:\\n \"\"\"\\n Search for pattern in files (similar to Unix grep command)\\n \\n Args:\\n pattern: Regular expression pattern to search for\\n file_path: Single file to search in (optional)\\n directory: Directory to search in (optional)\\n \\n Returns:\\n Dictionary with matching lines\\n \"\"\"\\n try:\\n matches = []\\n files_searched = []\\n \\n if file_path:\\n # Search in single file\\n full_path = os.path.join(self.root_dir, file_path)\\n real_path = os.path.realpath(full_path)\\n \\n if not real_path.startswith(self.root_dir):\\n return {\\n \"error\": f\"Access denied: Path outside root directory\",\\n \"success\": False\\n }\\n \\n ', 'total_lines': 869, 'lines_read': 869, 'offset': 0, 'end_line': 869, 'truncated': True, 'success': True}, error=None, timestamp=1784337629.322432)"
],
"metrics": {
"ttft": 6.189475774765015,
"ttft_per_iteration": [
6.189475774765015,
8.230858087539673,
28.87564492225647
],
"total_time": 43.29711675643921,
"iterations": 3,
"tool_calls": 3,
"cache_hits": 2,
"cache_misses": 0,
"prompt_tokens": 7430,
"completion_tokens": 907,
"cached_tokens": 674
},
"mode": "text_format",
"model": "kimi-k2.6",
"task": "Find all Python files in this directory; read main.py and agent.py; summarize in 3 sentences."
}