Initial commit with translated description

This commit is contained in:
2026-03-29 13:22:59 +08:00
commit 2d011d7093
18 changed files with 3162 additions and 0 deletions

69
CHANGELOG.md Normal file
View File

@@ -0,0 +1,69 @@
# Changelog
## [1.0.18] - 2026-02-22
### Security Improvements
- **Removed `"target": "slack"` from heartbeat config** - The optimizer no longer sets a default notification target. Previously, enabling heartbeat could cause unintended Slack messages if the user had webhooks configured.
- **`optimize` command now defaults to dry-run** - `python cli.py optimize` shows a preview. Use `--apply` to write changes. This matches the standalone `optimizer.py` behavior.
- **`setup-heartbeat` command now defaults to dry-run** - `python cli.py setup-heartbeat` shows a preview. Use `--apply` to write changes.
### Documentation
- **Added "What This Tool Modifies" section** to SKILL.md and README.md, listing all paths under `~/.openclaw/` that may be written.
- Updated all CLI examples to reflect the new `--apply` flag workflow.
## [1.0.17] - 2026-02-21
### Security Improvements
- **Removed all subprocess calls** - Replaced `subprocess.run` with `shutil.which` and HTTP health checks. No shell execution in the entire codebase.
- **Removed non-utility files from repo** - Deleted marketing materials, install scripts, competitor analysis, and promotional content from the repository.
- **Removed Unicode symbols** - Replaced non-ASCII characters in test output with ASCII equivalents.
- **Excluded Python cache from publish** - Added `__pycache__/` and `*.pyc` to `.clawhubignore`.
### Changed
- Ollama model setup now provides manual instructions instead of auto-downloading.
- Provider reachability checks use HTTP endpoints instead of CLI commands.
- Cleaned up documentation references to removed files.
## [1.0.8] - 2026-02-12
### New Features
- **Configurable heartbeat providers** - Support for `ollama`, `lmstudio`, `groq`, and `none`. Configure via `setup-heartbeat --provider <name>`.
- **Rollback command** - List and restore config backups with `rollback --list` and `rollback --to <file>`.
- **Health check command** - Quick system status with `health` (config, JSON validity, provider reachable, workspace size, budgets).
- **Diff preview in dry-run** - `optimize --dry-run` now shows a colored unified diff instead of dumping the full config.
- **`--no-color` flag** - Disable colored output globally with `--no-color` or `NO_COLOR` env var.
### Improvements
- **Shared colors module** - Deduplicated color code from 3 files into `src/colors.py`.
- **Version single source of truth** - All files read version from `src/__init__.py`. No more hardcoded version strings.
- **Extended triggers** - Added 10 new search keywords for better search matching.
- **Provider-aware verification** - `verify` checks the configured heartbeat provider instead of only Ollama.
### Fixes
- **License consistency** - Fixed setup.py classifier from "Proprietary" to MIT, README from "Commercial" to MIT.
- **URLs** - setup.py now points to correct smartpeopleconnected GitHub URLs.
- **Version sync** - All 7 files that showed "1.0.0" now correctly show 1.0.8.
## [1.0.7] - 2026-02-08
### Security Improvements
- **Cleaned up SKILL.md** - Removed unnecessary HTML comment from SKILL.md.
- **Dry-run is now the default** - running `optimizer.py` without flags shows a preview only. Use `--apply` to make actual changes. This prevents accidental config modifications.
- **User confirmation before downloads** - `ollama pull` now asks for confirmation before downloading ~2GB model data.
- **Existing files are no longer overwritten** - files in `~/.openclaw/prompts/` are skipped if they already exist, preserving user customizations.
### New Features
- **7-day savings report** - after 7 days of usage, `verify.py` shows your accumulated cost savings with a weekly breakdown. This report appears once every 7 days.
### Changed
- `--dry-run` flag replaced by `--apply` flag (dry-run is now the default behavior)
- Documentation updated to reflect new `--apply` workflow
## [1.0.6] - Initial ClawHub release
- Model routing (Haiku default)
- Ollama heartbeats (free local LLM)
- Session management optimization
- Prompt caching
- Budget controls and rate limits
- Verification tool

271
README.md Normal file
View File

@@ -0,0 +1,271 @@
# Token Optimizer for OpenClaw
**Reduce your AI costs by 97% - From $1,500+/month to under $50/month**
[![Version](https://img.shields.io/badge/version-1.0.18-blue.svg)](https://github.com/smartpeopleconnected/openclaw-token-optimizer)
[![License](https://img.shields.io/badge/license-MIT-green.svg)](LICENSE)
[![OpenClaw](https://img.shields.io/badge/OpenClaw-Compatible-purple.svg)](https://openclaw.ai)
[![Cost Savings](https://img.shields.io/badge/savings-97%25-brightgreen.svg)](https://github.com/smartpeopleconnected/openclaw-token-optimizer)
[![Ko-fi](https://img.shields.io/badge/Ko--fi-Support%20this%20project-FF5E5B?logo=ko-fi&logoColor=white)](https://ko-fi.com/smartpeopleconnected)
---
## The Problem
If you've been running OpenClaw and watching your API bills climb, you're not alone. The default configuration prioritizes capability over cost, which means you're probably burning through tokens on routine tasks that don't need expensive models.
**Common issues:**
- Loading 50KB of history on every message (2-3M wasted tokens/session)
- Using Sonnet/Opus for simple tasks that Haiku handles perfectly
- Paying for API heartbeats that could run on a free local LLM
- No rate limits leading to runaway automation costs
## The Solution
Token Optimizer applies four key optimizations that work together to slash your costs:
| Optimization | Before | After | Savings |
|--------------|--------|-------|---------|
| Session Management | 50KB context | 8KB context | 80% |
| Model Routing | Sonnet for everything | Haiku default | 92% |
| Heartbeat to Ollama | Paid API | Free local LLM | 100% |
| Prompt Caching | No caching | 90% cache hits | 90% |
**Combined result: 97% cost reduction**
## Cost Comparison
| Time Period | Before | After |
|-------------|--------|-------|
| Daily | $2-3 | **$0.10** |
| Monthly | $70-90 | **$3-5** |
| Yearly | $800+ | **$40-60** |
## What This Tool Modifies
All changes are written under `~/.openclaw/`. A backup is created before any modification.
| Path | Purpose |
|------|---------|
| `~/.openclaw/openclaw.json` | Main OpenClaw config (model routing, heartbeat, budgets) |
| `~/.openclaw/backups/` | Timestamped config backups (created automatically) |
| `~/.openclaw/workspace/` | Template files (SOUL.md, USER.md, IDENTITY.md) |
| `~/.openclaw/prompts/` | Agent prompt optimization rules |
| `~/.openclaw/token-optimizer-stats.json` | Usage stats for savings reports |
**Safe by default** - All commands run in dry-run (preview) mode. Pass `--apply` to write changes.
## Quick Start
### Installation
```bash
# Preview changes (dry-run by default)
python cli.py optimize
# Apply changes
python cli.py optimize --apply
# Quick health check
python cli.py health
```
### Verify Setup
```bash
python cli.py verify
```
## Features
### 1. Intelligent Model Routing
Sets Haiku as the default model with easy aliases for switching:
- `haiku` - Fast, cheap, perfect for 80% of tasks
- `sonnet` - Complex reasoning, architecture decisions
- `opus` - Mission-critical only
### 2. Free Heartbeats via Ollama
Routes heartbeat checks to a local LLM (llama3.2:3b) instead of paid API:
- Zero API calls for status checks
- No impact on rate limits
- Saves $5-15/month automatically
### 3. Lean Session Management
Optimized context loading rules that reduce startup context from 50KB to 8KB:
- Load only essential files (SOUL.md, USER.md)
- On-demand history retrieval
- Daily memory notes instead of history bloat
### 4. Prompt Caching
Automatic 90% discount on repeated content:
- Agent prompts cached and reused
- 5-minute TTL for optimal cache hits
- Per-model cache configuration
### 5. Budget Controls
Built-in rate limits and budget warnings:
- Daily/monthly budget caps
- Warning at 75% threshold
- Rate limiting between API calls
## Usage
### Analyze Current Setup
```bash
python cli.py analyze
```
Shows current configuration status, workspace file sizes, optimization opportunities, and estimated monthly savings.
### Preview Changes (Dry Run - Default)
```bash
python cli.py optimize
```
Shows a colored unified diff of what would change, without modifying anything.
### Apply Full Optimization
```bash
python cli.py optimize --apply
```
Applies all optimizations: model routing, heartbeat, caching, rate limits, workspace templates, and agent prompts.
### Apply Specific Optimizations
```bash
python cli.py optimize --apply --mode routing # Model routing only
python cli.py optimize --apply --mode heartbeat # Heartbeat only
python cli.py optimize --apply --mode caching # Prompt caching only
python cli.py optimize --apply --mode limits # Rate limits only
```
### Quick Health Check
```bash
python cli.py health
```
Checks config exists, valid JSON, provider reachable, workspace lean, and budget active.
### Configure Heartbeat Provider
```bash
# Preview (dry-run by default)
python cli.py setup-heartbeat --provider ollama
# Apply changes
python cli.py setup-heartbeat --provider ollama --apply
python cli.py setup-heartbeat --provider lmstudio --apply
python cli.py setup-heartbeat --provider groq --apply
python cli.py setup-heartbeat --provider none --apply
python cli.py setup-heartbeat --provider groq --fallback ollama --apply
```
### Rollback Configuration
```bash
python cli.py rollback --list # List available backups
python cli.py rollback --to <filename> # Restore a specific backup
```
### Verify Setup
```bash
python cli.py verify
```
### Disable Colors
```bash
python cli.py --no-color optimize
# or
NO_COLOR=1 python cli.py optimize
```
## Configuration
After installation, edit these files:
### `~/.openclaw/workspace/SOUL.md`
Agent principles and operating rules. Includes:
- Model selection rules
- Session initialization rules
- Rate limit rules
### `~/.openclaw/workspace/USER.md`
Your context: name, role, mission, success metrics.
### `~/.openclaw/prompts/OPTIMIZATION-RULES.md`
Copy these rules into your agent prompt.
## Requirements
- Python 3.8+
- OpenClaw installed and configured
- Ollama (optional, for free heartbeats)
### Installing Ollama (Optional)
Ollama is only needed if you want free local heartbeats. Download from [https://ollama.ai](https://ollama.ai), then:
```bash
ollama pull llama3.2:3b
ollama serve
```
Or use the CLI to configure a different provider:
```bash
python cli.py setup-heartbeat --provider lmstudio
python cli.py setup-heartbeat --provider none # disable heartbeat
```
## File Structure
```
token-optimizer/
+-- skill.json # Skill manifest
+-- README.md # This file
+-- src/
| +-- __init__.py # Version (single source of truth)
| +-- colors.py # Shared ANSI colors
| +-- analyzer.py # Analyzes current config
| +-- optimizer.py # Applies optimizations
| +-- verify.py # Verifies setup
+-- templates/
| +-- openclaw-config-optimized.json
| +-- SOUL.md
| +-- USER.md
| +-- OPTIMIZATION-RULES.md
+-- test/
+-- simulation_test.py # Simulation tests
```
## Troubleshooting
### Context size still large
- Ensure SESSION INITIALIZATION RULE is in your agent prompt
- Check that SOUL.md and USER.md are lean (<15KB total)
### Still using Sonnet for everything
- Verify `~/.openclaw/openclaw.json` has correct model configuration
- Ensure MODEL SELECTION RULE is in agent prompt
### Heartbeat errors
- Make sure Ollama is running: `ollama serve`
- Verify model is installed: `ollama list`
### Costs haven't dropped
- Run `python src/verify.py` to check all optimizations
- Ensure agent prompt includes all optimization rules
## Support
- **Issues:** [GitHub Issues](https://github.com/smartpeopleconnected/openclaw-token-optimizer/issues)
**If this tool saved you money, consider supporting development:**
[![Ko-fi](https://ko-fi.com/img/githubbutton_sm.svg)](https://ko-fi.com/smartpeopleconnected)
## License
MIT License. See [LICENSE](LICENSE) for details.
---
**Built with care by Smart People Connected**
*Stop burning tokens. Start building things.*

158
SKILL.md Normal file
View File

@@ -0,0 +1,158 @@
---
name: token-optimizer
description: "将OpenClaw AI成本降低97%。"
homepage: https://github.com/smartpeopleconnected/openclaw-token-optimizer
triggers:
- too expensive
- costs too much
- burning tokens
- high token usage
- reduce costs
- save money
- optimize tokens
- budget exceeded
- token optimization
- cut api costs
- lower ai spend
- cheaper model
- cost savings
- api bill
- spending too much
- waste tokens
- token budget
- reduce token usage
---
# Token Optimizer for OpenClaw
Slash your AI costs from $1,500+/month to under $50/month.
## The Problem
OpenClaw defaults prioritize capability over cost. You're burning expensive Sonnet/Opus tokens on tasks Haiku handles perfectly, paying for API heartbeats that could run free locally, and loading 50KB of context when 8KB suffices.
## The Solution
Four core optimizations plus powerful tooling:
### Model Routing (92% savings)
Haiku by default, Sonnet/Opus only when needed
### Multi-Provider Heartbeats (100% savings)
Route heartbeats to Ollama, LM Studio, Groq, or disable entirely. Not locked to one provider.
### Session Management (80% savings)
Load 8KB instead of 50KB context
### Caching (90% savings)
Reuse prompts at 10% cost
### New in v1.0.8
- **Rollback** - List and restore config backups instantly
- **Health Check** - Quick system status in one command
- **Diff Preview** - See exactly what changes before applying
- **--no-color** - CI/pipeline friendly output
## Cost Comparison
| Period | Before | After |
|--------|--------|-------|
| Daily | $2-3 | $0.10 |
| Monthly | $70-90 | $3-5 |
| Yearly | $800+ | $40-60 |
## What's Included
- One-command optimizer with diff preview
- Multi-provider heartbeat (Ollama, LM Studio, Groq)
- Config rollback and health check commands
- Ready-to-use config templates
- SOUL.md & USER.md templates
- Optimization rules for agent prompts
- Verification and savings reports
## What This Tool Modifies
All changes are written under `~/.openclaw/`. A backup is created before any modification.
| Path | Purpose |
|------|---------|
| `~/.openclaw/openclaw.json` | Main OpenClaw config (model routing, heartbeat, budgets) |
| `~/.openclaw/backups/` | Timestamped config backups (created automatically) |
| `~/.openclaw/workspace/` | Template files (SOUL.md, USER.md, IDENTITY.md) |
| `~/.openclaw/prompts/` | Agent prompt optimization rules |
| `~/.openclaw/token-optimizer-stats.json` | Usage stats for savings reports |
**Safe by default** - All commands run in dry-run (preview) mode. Pass `--apply` to write changes.
## Quick Start
```bash
# Install
clawhub install token-optimizer
# Analyze current setup
python cli.py analyze
# Preview changes (dry-run by default)
python cli.py optimize
# Apply all optimizations
python cli.py optimize --apply
# Verify setup
python cli.py verify
# Quick health check
python cli.py health
# Configure heartbeat provider (preview)
python cli.py setup-heartbeat --provider ollama
# Configure heartbeat provider (apply)
python cli.py setup-heartbeat --provider ollama --apply
# List and restore backups
python cli.py rollback --list
python cli.py rollback --to <backup-file>
```
## Configuration Generated
```json
{
"agents": {
"defaults": {
"model": { "primary": "anthropic/claude-haiku-4-5" },
"cache": { "enabled": true, "ttl": "5m" }
}
},
"heartbeat": {
"provider": "ollama",
"model": "ollama/llama3.2:3b"
},
"budgets": {
"daily": 5.00,
"monthly": 200.00
}
}
```
## Links
- **GitHub**: https://github.com/smartpeopleconnected/openclaw-token-optimizer
- **Issues**: https://github.com/smartpeopleconnected/openclaw-token-optimizer/issues
## Author
**Smart People Connected**
- GitHub: [@smartpeopleconnected](https://github.com/smartpeopleconnected)
- Email: smartpeopleconnected@gmail.com
## License
MIT License - Free to use, modify, and distribute.
---
*5 minutes to setup. 97% cost reduction. Stop burning tokens. Start building.*

6
_meta.json Normal file
View File

@@ -0,0 +1,6 @@
{
"ownerId": "kn7fazk300m4881qd1hgkqhhr180gzvw",
"slug": "token-optimizer",
"version": "1.0.18",
"publishedAt": 1771671237453
}

277
cli.py Normal file
View File

@@ -0,0 +1,277 @@
#!/usr/bin/env python3
"""
Token Optimizer CLI
Command-line interface for OpenClaw token optimization.
"""
import sys
import json
import argparse
from pathlib import Path
from src import __version__
def main():
parser = argparse.ArgumentParser(
prog='token-optimizer',
description='Reduce OpenClaw AI costs by 97%',
epilog='For more info: https://github.com/smartpeopleconnected/openclaw-token-optimizer'
)
parser.add_argument(
'--no-color',
action='store_true',
help='Disable colored output'
)
subparsers = parser.add_subparsers(dest='command', help='Available commands')
# Analyze command
subparsers.add_parser(
'analyze',
help='Analyze current configuration and show optimization opportunities'
)
# Optimize command
optimize_parser = subparsers.add_parser(
'optimize',
help='Apply token optimizations'
)
optimize_parser.add_argument(
'--mode',
choices=['full', 'routing', 'heartbeat', 'caching', 'limits'],
default='full',
help='Optimization mode (default: full)'
)
optimize_parser.add_argument(
'--apply',
action='store_true',
help='Apply changes (default is dry-run preview)'
)
# Verify command
subparsers.add_parser(
'verify',
help='Verify optimization setup and show estimated savings'
)
# Setup heartbeat command
heartbeat_parser = subparsers.add_parser(
'setup-heartbeat',
help='Configure heartbeat provider (ollama, lmstudio, groq, none)'
)
heartbeat_parser.add_argument(
'--provider',
choices=['ollama', 'lmstudio', 'groq', 'none'],
default='ollama',
help='Heartbeat provider (default: ollama)'
)
heartbeat_parser.add_argument(
'--model',
default=None,
help='Model name for heartbeat (default: provider-specific)'
)
heartbeat_parser.add_argument(
'--fallback',
choices=['ollama', 'lmstudio', 'groq', 'none'],
default=None,
help='Fallback provider if primary is unavailable'
)
heartbeat_parser.add_argument(
'--apply',
action='store_true',
help='Apply changes (default is dry-run preview)'
)
# Rollback command
rollback_parser = subparsers.add_parser(
'rollback',
help='Restore a previous configuration backup'
)
rollback_parser.add_argument(
'--list',
action='store_true',
dest='list_backups',
help='List available backups'
)
rollback_parser.add_argument(
'--to',
dest='backup_file',
default=None,
help='Restore a specific backup file'
)
# Health command
subparsers.add_parser(
'health',
help='Quick system health check'
)
# Version command
subparsers.add_parser(
'version',
help='Show version information'
)
args = parser.parse_args()
# Apply --no-color globally
if args.no_color:
import src.colors
src.colors.NO_COLOR = True
if args.command == 'analyze':
from src.analyzer import main as analyze_main
return analyze_main()
elif args.command == 'optimize':
from src.optimizer import TokenOptimizer
dry_run = not args.apply
if dry_run:
from src.colors import Colors, colorize
print(colorize("[DRY-RUN] Preview mode. Use --apply to make changes.\n", Colors.YELLOW))
optimizer = TokenOptimizer(dry_run=dry_run)
optimizer.optimize_mode(args.mode)
return 0
elif args.command == 'verify':
from src.verify import main as verify_main
return verify_main()
elif args.command == 'setup-heartbeat':
from src.optimizer import TokenOptimizer
from src.colors import Colors, colorize
dry_run = not args.apply
if dry_run:
print(colorize("[DRY-RUN] Preview mode. Use --apply to make changes.\n", Colors.YELLOW))
optimizer = TokenOptimizer(dry_run=dry_run)
optimizer.setup_heartbeat_provider(
provider=args.provider,
model=args.model,
fallback=args.fallback
)
config = optimizer.load_config()
config = optimizer.apply_heartbeat(
config,
provider=args.provider,
model=args.model,
fallback=args.fallback
)
optimizer.save_config(config)
return 0
elif args.command == 'rollback':
from src.optimizer import TokenOptimizer
from src.colors import Colors, colorize
optimizer = TokenOptimizer()
if args.list_backups:
backups = optimizer.list_backups()
if not backups:
print(colorize("[INFO] No backups found", Colors.YELLOW))
else:
print(colorize(f"[INFO] {len(backups)} backup(s) found:\n", Colors.CYAN))
for b in backups:
size_kb = b.stat().st_size / 1024
print(f" {b.name} ({size_kb:.1f} KB)")
print(colorize(f"\nRestore with: token-optimizer rollback --to <filename>", Colors.CYAN))
return 0
if args.backup_file:
backup_path = optimizer.backup_dir / args.backup_file
if not backup_path.exists():
# Try as absolute path
backup_path = Path(args.backup_file)
success = optimizer.restore_backup(backup_path)
return 0 if success else 1
print(colorize("[ERROR] Use --list to see backups or --to <file> to restore", Colors.RED))
return 1
elif args.command == 'health':
from src.colors import Colors, colorize
from src.optimizer import resolve_heartbeat_provider, check_heartbeat_provider
print(colorize("\n=== Token Optimizer - Health Check ===\n", Colors.BOLD + Colors.CYAN))
openclaw_dir = Path.home() / '.openclaw'
config_path = openclaw_dir / 'openclaw.json'
checks_passed = 0
checks_total = 0
# 1. Config exists
checks_total += 1
if config_path.exists():
print(colorize("[PASS] Config file exists", Colors.GREEN))
checks_passed += 1
else:
print(colorize("[FAIL] Config file not found", Colors.RED))
print(colorize(" Run: token-optimizer optimize", Colors.CYAN))
# 2. Valid JSON
config = {}
checks_total += 1
if config_path.exists():
try:
with open(config_path, 'r') as f:
config = json.load(f)
print(colorize("[PASS] Config is valid JSON", Colors.GREEN))
checks_passed += 1
except (json.JSONDecodeError, IOError):
print(colorize("[FAIL] Config is not valid JSON", Colors.RED))
else:
print(colorize("[SKIP] Config not found, skipping JSON check", Colors.YELLOW))
# 3. Provider reachable
checks_total += 1
provider = resolve_heartbeat_provider(config)
if provider == "none":
print(colorize("[SKIP] Heartbeat disabled (provider: none)", Colors.YELLOW))
checks_passed += 1
elif check_heartbeat_provider(provider):
print(colorize(f"[PASS] Heartbeat provider '{provider}' is reachable", Colors.GREEN))
checks_passed += 1
else:
print(colorize(f"[FAIL] Heartbeat provider '{provider}' is not reachable", Colors.RED))
# 4. Workspace lean
checks_total += 1
workspace_dir = openclaw_dir / 'workspace'
if workspace_dir.exists():
total_size = sum(f.stat().st_size for f in workspace_dir.iterdir() if f.is_file())
size_kb = total_size / 1024
if size_kb < 15:
print(colorize(f"[PASS] Workspace is lean ({size_kb:.1f} KB)", Colors.GREEN))
checks_passed += 1
else:
print(colorize(f"[WARN] Workspace is large ({size_kb:.1f} KB, target <15 KB)", Colors.YELLOW))
else:
print(colorize("[SKIP] No workspace directory found", Colors.YELLOW))
checks_passed += 1
# 5. Budget active
checks_total += 1
budgets = config.get('budgets', {})
if budgets.get('daily') or budgets.get('monthly'):
print(colorize(f"[PASS] Budget controls active (daily: ${budgets.get('daily', 'n/a')}, monthly: ${budgets.get('monthly', 'n/a')})", Colors.GREEN))
checks_passed += 1
else:
print(colorize("[FAIL] No budget controls configured", Colors.RED))
# Summary
print(colorize(f"\n{checks_passed}/{checks_total} checks passed", Colors.BOLD))
return 0 if checks_passed == checks_total else 1
elif args.command == 'version':
print(f"Token Optimizer v{__version__}")
print("Reduce OpenClaw AI costs by 97%")
return 0
else:
parser.print_help()
return 0
if __name__ == '__main__':
sys.exit(main())

8
requirements.txt Normal file
View File

@@ -0,0 +1,8 @@
# Token Optimizer Requirements
# No external dependencies required for core functionality
# Optional development dependencies
# pytest>=7.0
# pytest-cov>=4.0
# black>=23.0
# mypy>=1.0

76
setup.py Normal file
View File

@@ -0,0 +1,76 @@
#!/usr/bin/env python3
"""
Token Optimizer Setup
Package configuration for pip installation.
"""
from setuptools import setup, find_packages
from pathlib import Path
import re
# Read version from src/__init__.py (single source of truth)
init_path = Path(__file__).parent / "src" / "__init__.py"
version = re.search(r'__version__\s*=\s*"([^"]+)"', init_path.read_text()).group(1)
# Read README for long description
readme_path = Path(__file__).parent / "README.md"
long_description = readme_path.read_text(encoding="utf-8") if readme_path.exists() else ""
setup(
name="token-optimizer",
version=version,
author="Smart People Connected",
author_email="smartpeopleconnected@gmail.com",
description="Reduce OpenClaw AI costs by 97% - From $1,500+/month to under $50/month",
long_description=long_description,
long_description_content_type="text/markdown",
url="https://github.com/smartpeopleconnected/openclaw-token-optimizer",
project_urls={
"Homepage": "https://github.com/smartpeopleconnected/openclaw-token-optimizer",
"Bug Tracker": "https://github.com/smartpeopleconnected/openclaw-token-optimizer/issues",
"Ko-fi": "https://ko-fi.com/smartpeopleconnected",
},
classifiers=[
"Development Status :: 5 - Production/Stable",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries :: Python Modules",
"Topic :: Office/Business :: Financial",
],
packages=find_packages(),
package_data={
"": ["templates/*", "templates/**/*"],
},
include_package_data=True,
python_requires=">=3.8",
install_requires=[],
extras_require={
"dev": [
"pytest>=7.0",
"pytest-cov>=4.0",
"black>=23.0",
"mypy>=1.0",
],
},
entry_points={
"console_scripts": [
"token-optimizer=cli:main",
],
},
keywords=[
"openclaw",
"token-optimization",
"cost-reduction",
"ai-efficiency",
"claude",
"anthropic",
"llm",
],
)

64
skill.json Normal file
View File

@@ -0,0 +1,64 @@
{
"name": "token-optimizer",
"version": "1.0.18",
"summary": "Reduce OpenClaw AI costs by 97%. Haiku routing, free Ollama heartbeats, prompt caching, budget controls. $1,500/month to $50/month in 5 minutes.",
"description": "Stop burning tokens! Slash your AI costs from $1,500+/month to under $50/month. Model routing (Haiku default), free Ollama heartbeats, lean session management, prompt caching, and budget controls. Works in 5 minutes.",
"author": "smartpeopleconnected",
"license": "MIT",
"keywords": [
"token-optimizer",
"cost-reduction",
"save-money",
"reduce-costs",
"expensive",
"burning-tokens",
"token-burn",
"high-costs",
"api-costs",
"budget",
"haiku",
"sonnet",
"opus",
"model-routing",
"model-selection",
"ollama",
"heartbeat",
"free-heartbeat",
"prompt-caching",
"cache",
"context-optimization",
"session-management",
"rate-limit",
"openclaw",
"clawdbot",
"efficiency",
"optimization",
"97-percent",
"savings"
],
"homepage": "https://github.com/smartpeopleconnected/openclaw-token-optimizer",
"repository": {
"type": "git",
"url": "https://github.com/smartpeopleconnected/openclaw-token-optimizer.git"
},
"triggers": [
"too expensive",
"costs too much",
"burning tokens",
"high token usage",
"reduce costs",
"save money",
"optimize tokens",
"budget exceeded",
"token optimization",
"cut api costs",
"lower ai spend",
"cheaper model",
"cost savings",
"api bill",
"spending too much",
"waste tokens",
"token budget",
"reduce token usage"
]
}

13
src/__init__.py Normal file
View File

@@ -0,0 +1,13 @@
"""
Token Optimizer for OpenClaw
Reduce AI costs by 97% through intelligent optimization.
"""
__version__ = "1.0.18"
__author__ = "TokenOptimizer"
from .analyzer import OpenClawAnalyzer
from .optimizer import TokenOptimizer
from .verify import OptimizationVerifier
__all__ = ['OpenClawAnalyzer', 'TokenOptimizer', 'OptimizationVerifier']

401
src/analyzer.py Normal file
View File

@@ -0,0 +1,401 @@
#!/usr/bin/env python3
"""
Token Optimizer - Analyzer Module
Analyzes OpenClaw configuration and estimates token usage & savings.
"""
import json
import os
import sys
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Optional, Tuple
try:
from src.colors import Colors, colorize
except ImportError:
from colors import Colors, colorize
class OpenClawAnalyzer:
"""Analyzes OpenClaw configuration for token optimization opportunities."""
# Cost per 1K tokens (approximate)
COSTS = {
'sonnet': 0.003,
'haiku': 0.00025,
'opus': 0.015,
'ollama': 0.0
}
# Average token estimates
ESTIMATES = {
'large_context': 50000, # 50KB unoptimized context
'lean_context': 8000, # 8KB optimized context
'heartbeat_tokens': 500, # Tokens per heartbeat
'heartbeats_per_day': 24, # Hourly heartbeats
'avg_messages_per_day': 100, # Average API calls
}
def __init__(self):
self.config_path = self._find_config()
self.config = self._load_config()
self.workspace_files = self._scan_workspace()
self.issues: List[Dict] = []
self.optimizations: List[Dict] = []
def _find_config(self) -> Optional[Path]:
"""Find OpenClaw configuration file."""
possible_paths = [
Path.home() / '.openclaw' / 'openclaw.json',
Path.home() / '.openclaw' / 'openclaw-config.json',
Path.home() / '.openclaw' / 'config.json',
Path.cwd() / '.openclaw.json',
Path.cwd() / 'openclaw.json',
]
for path in possible_paths:
if path.exists():
return path
return None
def _load_config(self) -> Dict:
"""Load OpenClaw configuration."""
if self.config_path and self.config_path.exists():
try:
with open(self.config_path, 'r') as f:
return json.load(f)
except json.JSONDecodeError:
return {}
return {}
def _scan_workspace(self) -> Dict[str, int]:
"""Scan workspace files and their sizes."""
workspace_files = {}
workspace_paths = [
Path.cwd(),
Path.home() / '.openclaw' / 'workspace',
]
target_files = ['SOUL.md', 'USER.md', 'IDENTITY.md', 'MEMORY.md',
'TOOLS.md', 'REFERENCE.md', 'CONTEXT.md']
for base_path in workspace_paths:
if base_path.exists():
for file_name in target_files:
file_path = base_path / file_name
if file_path.exists():
workspace_files[file_name] = file_path.stat().st_size
return workspace_files
def analyze_model_routing(self) -> Dict:
"""Analyze model routing configuration."""
result = {
'status': 'not_configured',
'default_model': 'unknown',
'has_haiku': False,
'has_sonnet': False,
'has_aliases': False,
'monthly_savings': 0
}
agents_config = self.config.get('agents', {})
defaults = agents_config.get('defaults', {})
model_config = defaults.get('model', {})
models = defaults.get('models', {})
# Check primary model
primary = model_config.get('primary', '')
if 'haiku' in primary.lower():
result['default_model'] = 'haiku'
result['status'] = 'optimized'
elif 'sonnet' in primary.lower():
result['default_model'] = 'sonnet'
result['status'] = 'needs_optimization'
elif 'opus' in primary.lower():
result['default_model'] = 'opus'
result['status'] = 'needs_optimization'
# Check available models
for model_name in models:
if 'haiku' in model_name.lower():
result['has_haiku'] = True
if 'sonnet' in model_name.lower():
result['has_sonnet'] = True
if models[model_name].get('alias'):
result['has_aliases'] = True
# Calculate potential savings
if result['status'] == 'needs_optimization':
daily_calls = self.ESTIMATES['avg_messages_per_day']
avg_tokens = 2000 # per call
if result['default_model'] == 'sonnet':
current_cost = (daily_calls * avg_tokens / 1000) * self.COSTS['sonnet']
else: # opus
current_cost = (daily_calls * avg_tokens / 1000) * self.COSTS['opus']
optimized_cost = (daily_calls * avg_tokens / 1000) * self.COSTS['haiku']
result['monthly_savings'] = (current_cost - optimized_cost) * 30
return result
def analyze_heartbeat(self) -> Dict:
"""Analyze heartbeat configuration."""
result = {
'status': 'not_configured',
'provider': 'api',
'interval': 3600,
'monthly_cost': 0,
'monthly_savings': 0
}
heartbeat_config = self.config.get('heartbeat', {})
if heartbeat_config:
result['interval'] = heartbeat_config.get('every', '1h')
model = heartbeat_config.get('model', '')
if 'ollama' in model.lower() or 'local' in model.lower():
result['provider'] = 'ollama'
result['status'] = 'optimized'
result['monthly_cost'] = 0
else:
result['provider'] = 'api'
result['status'] = 'needs_optimization'
# Calculate cost
heartbeats_per_day = self.ESTIMATES['heartbeats_per_day']
tokens_per_heartbeat = self.ESTIMATES['heartbeat_tokens']
cost_per_1k = self.COSTS['haiku'] # assume haiku at minimum
daily_cost = (heartbeats_per_day * tokens_per_heartbeat / 1000) * cost_per_1k
result['monthly_cost'] = daily_cost * 30
result['monthly_savings'] = result['monthly_cost']
else:
result['status'] = 'not_configured'
return result
def analyze_session_management(self) -> Dict:
"""Analyze session initialization and context management."""
result = {
'status': 'unknown',
'estimated_context_size': 0,
'optimized_context_size': 8000,
'monthly_savings': 0
}
# Calculate current context size from workspace files
total_size = sum(self.workspace_files.values())
result['estimated_context_size'] = total_size if total_size > 0 else self.ESTIMATES['large_context']
# Check if files are reasonably sized
if total_size > 20000: # > 20KB
result['status'] = 'needs_optimization'
elif total_size > 0:
result['status'] = 'optimized'
else:
result['status'] = 'no_workspace_files'
# Calculate savings
if result['status'] == 'needs_optimization':
daily_calls = self.ESTIMATES['avg_messages_per_day']
excess_tokens = (result['estimated_context_size'] - result['optimized_context_size'])
cost_per_1k = self.COSTS['haiku']
daily_waste = (excess_tokens / 1000) * cost_per_1k * daily_calls
result['monthly_savings'] = daily_waste * 30
return result
def analyze_caching(self) -> Dict:
"""Analyze prompt caching configuration."""
result = {
'status': 'not_configured',
'enabled': False,
'ttl': '5m',
'monthly_savings': 0
}
agents_config = self.config.get('agents', {})
defaults = agents_config.get('defaults', {})
cache_config = defaults.get('cache', {})
if cache_config.get('enabled'):
result['enabled'] = True
result['status'] = 'optimized'
result['ttl'] = cache_config.get('ttl', '5m')
else:
result['status'] = 'needs_optimization'
# Calculate potential savings (90% on cached content)
daily_calls = self.ESTIMATES['avg_messages_per_day']
prompt_size = 5000 # 5KB typical
cost_per_1k = self.COSTS['sonnet'] # caching matters most for sonnet
uncached_cost = (prompt_size / 1000) * cost_per_1k * daily_calls
cached_cost = uncached_cost * 0.1 # 90% discount
result['monthly_savings'] = (uncached_cost - cached_cost) * 30
return result
def analyze_rate_limits(self) -> Dict:
"""Check if rate limits are configured."""
result = {
'status': 'unknown',
'has_api_limit': False,
'has_budget': False,
'daily_budget': None,
'monthly_budget': None
}
# Check for rate limit configuration
rate_limits = self.config.get('rate_limits', {})
budgets = self.config.get('budgets', {})
if rate_limits:
result['has_api_limit'] = True
if budgets:
result['has_budget'] = True
result['daily_budget'] = budgets.get('daily')
result['monthly_budget'] = budgets.get('monthly')
if result['has_api_limit'] or result['has_budget']:
result['status'] = 'configured'
else:
result['status'] = 'not_configured'
return result
def run_full_analysis(self) -> Dict:
"""Run complete analysis and return results."""
print(colorize("\n=== OpenClaw Token Optimizer - Analysis ===\n", Colors.BOLD + Colors.CYAN))
results = {
'timestamp': datetime.now().isoformat(),
'config_found': self.config_path is not None,
'config_path': str(self.config_path) if self.config_path else None,
'workspace_files': self.workspace_files,
'model_routing': self.analyze_model_routing(),
'heartbeat': self.analyze_heartbeat(),
'session_management': self.analyze_session_management(),
'caching': self.analyze_caching(),
'rate_limits': self.analyze_rate_limits(),
}
# Calculate total potential savings
total_savings = (
results['model_routing']['monthly_savings'] +
results['heartbeat']['monthly_savings'] +
results['session_management']['monthly_savings'] +
results['caching']['monthly_savings']
)
results['total_monthly_savings'] = total_savings
self._print_results(results)
return results
def _print_results(self, results: Dict):
"""Print formatted analysis results."""
# Config status
if results['config_found']:
print(colorize(f"[FOUND] Config: {results['config_path']}", Colors.GREEN))
else:
print(colorize("[NOT FOUND] No OpenClaw config file detected", Colors.YELLOW))
print(" Run 'optimize' to create optimized configuration\n")
# Workspace files
print(colorize("\n--- Workspace Files ---", Colors.BOLD))
if results['workspace_files']:
for name, size in results['workspace_files'].items():
size_kb = size / 1024
color = Colors.GREEN if size_kb < 5 else Colors.YELLOW if size_kb < 15 else Colors.RED
print(f" {name}: {colorize(f'{size_kb:.1f}KB', color)}")
else:
print(colorize(" No workspace files found", Colors.YELLOW))
# Model Routing
print(colorize("\n--- Model Routing ---", Colors.BOLD))
mr = results['model_routing']
status_color = Colors.GREEN if mr['status'] == 'optimized' else Colors.RED
print(f" Status: {colorize(mr['status'].upper(), status_color)}")
print(f" Default Model: {mr['default_model']}")
if mr['monthly_savings'] > 0:
print(colorize(f" Potential Savings: ${mr['monthly_savings']:.2f}/month", Colors.GREEN))
# Heartbeat
print(colorize("\n--- Heartbeat Configuration ---", Colors.BOLD))
hb = results['heartbeat']
status_color = Colors.GREEN if hb['status'] == 'optimized' else Colors.YELLOW if hb['status'] == 'not_configured' else Colors.RED
print(f" Status: {colorize(hb['status'].upper(), status_color)}")
print(f" Provider: {hb['provider']}")
if hb['monthly_savings'] > 0:
print(colorize(f" Potential Savings: ${hb['monthly_savings']:.2f}/month", Colors.GREEN))
# Session Management
print(colorize("\n--- Session Management ---", Colors.BOLD))
sm = results['session_management']
status_color = Colors.GREEN if sm['status'] == 'optimized' else Colors.YELLOW if sm['status'] == 'no_workspace_files' else Colors.RED
print(f" Status: {colorize(sm['status'].upper(), status_color)}")
print(f" Estimated Context: {sm['estimated_context_size'] / 1024:.1f}KB")
if sm['monthly_savings'] > 0:
print(colorize(f" Potential Savings: ${sm['monthly_savings']:.2f}/month", Colors.GREEN))
# Caching
print(colorize("\n--- Prompt Caching ---", Colors.BOLD))
cache = results['caching']
status_color = Colors.GREEN if cache['status'] == 'optimized' else Colors.RED
print(f" Status: {colorize(cache['status'].upper(), status_color)}")
print(f" Enabled: {cache['enabled']}")
if cache['monthly_savings'] > 0:
print(colorize(f" Potential Savings: ${cache['monthly_savings']:.2f}/month", Colors.GREEN))
# Rate Limits
print(colorize("\n--- Rate Limits & Budgets ---", Colors.BOLD))
rl = results['rate_limits']
status_color = Colors.GREEN if rl['status'] == 'configured' else Colors.YELLOW
print(f" Status: {colorize(rl['status'].upper(), status_color)}")
print(f" API Limits: {'Yes' if rl['has_api_limit'] else 'No'}")
print(f" Budgets: {'Yes' if rl['has_budget'] else 'No'}")
# Total Savings
print(colorize("\n========================================", Colors.BOLD + Colors.CYAN))
print(colorize(f"TOTAL POTENTIAL SAVINGS: ${results['total_monthly_savings']:.2f}/month", Colors.BOLD + Colors.GREEN))
print(colorize(f" ${results['total_monthly_savings'] * 12:.2f}/year", Colors.GREEN))
print(colorize("========================================\n", Colors.BOLD + Colors.CYAN))
# Recommendations
if results['total_monthly_savings'] > 0:
print(colorize("RECOMMENDATIONS:", Colors.BOLD + Colors.YELLOW))
if mr['status'] == 'needs_optimization':
print(" 1. Switch default model to Haiku")
if hb['status'] != 'optimized':
print(" 2. Route heartbeats to Ollama (free)")
if sm['status'] == 'needs_optimization':
print(" 3. Implement session initialization rules")
if cache['status'] != 'optimized':
print(" 4. Enable prompt caching")
if rl['status'] != 'configured':
print(" 5. Add rate limits and budgets")
print(colorize("\nRun 'token-optimizer optimize' to apply all optimizations", Colors.CYAN))
def main():
"""Main entry point."""
analyzer = OpenClawAnalyzer()
results = analyzer.run_full_analysis()
# Save results to file
output_path = Path.cwd() / '.token-optimizer-analysis.json'
with open(output_path, 'w') as f:
json.dump(results, f, indent=2, default=str)
print(f"\nDetailed results saved to: {output_path}")
print(colorize("\nRun 'python src/verify.py' to see your accumulated savings report.", Colors.CYAN))
return 0 if results['total_monthly_savings'] == 0 else 1
if __name__ == '__main__':
sys.exit(main())

30
src/colors.py Normal file
View File

@@ -0,0 +1,30 @@
"""
Shared ANSI color codes and colorize helper for Token Optimizer.
"""
import os
import sys
# Global flag - set to True to disable all color output
NO_COLOR = os.environ.get("NO_COLOR", "") != ""
class Colors:
RED = '\033[91m'
GREEN = '\033[92m'
YELLOW = '\033[93m'
BLUE = '\033[94m'
MAGENTA = '\033[95m'
CYAN = '\033[96m'
WHITE = '\033[97m'
BOLD = '\033[1m'
END = '\033[0m'
def colorize(text: str, color: str) -> str:
"""Apply color to text if terminal supports it and NO_COLOR is not set."""
if NO_COLOR:
return text
if sys.stdout.isatty():
return f"{color}{text}{Colors.END}"
return text

741
src/optimizer.py Normal file
View File

@@ -0,0 +1,741 @@
#!/usr/bin/env python3
"""
Token Optimizer - Main Optimization Module
Applies token optimization configurations to OpenClaw.
"""
import json
import os
import sys
import shutil
import difflib
import urllib.request
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Optional
import argparse
try:
from src.colors import Colors, colorize
from src import __version__
except ImportError:
# Standalone execution fallback
from colors import Colors, colorize
__version__ = "1.0.8"
HEARTBEAT_PROVIDERS = {
"ollama": {
"endpoint": "http://localhost:11434",
"default_model": "llama3.2:3b",
"model_prefix": "ollama/",
"cli_name": "ollama",
},
"lmstudio": {
"endpoint": "http://localhost:1234",
"default_model": "llama3.2:3b",
"model_prefix": "lmstudio/",
"cli_name": None,
},
"groq": {
"endpoint": "https://api.groq.com",
"default_model": "llama-3.2-3b-preview",
"model_prefix": "groq/",
"cli_name": None,
},
"none": {
"endpoint": None,
"default_model": None,
"model_prefix": "",
"cli_name": None,
},
}
def resolve_heartbeat_provider(config: Dict) -> str:
"""Resolve heartbeat provider from config, with auto-detect fallback."""
heartbeat = config.get("heartbeat", {})
# Explicit provider field takes priority
provider = heartbeat.get("provider")
if provider and provider in HEARTBEAT_PROVIDERS:
return provider
# Auto-detect from model string
model = heartbeat.get("model", "")
for name in HEARTBEAT_PROVIDERS:
if name != "none" and name in model.lower():
return name
return "ollama" # default
def check_heartbeat_provider(provider: str) -> bool:
"""Check if a heartbeat provider is reachable."""
if provider == "none":
return True
info = HEARTBEAT_PROVIDERS.get(provider)
if not info:
return False
# Check if CLI tool is installed (e.g. ollama)
cli_name = info.get("cli_name")
if cli_name:
if shutil.which(cli_name) is None:
return False
# CLI found, try reaching the HTTP endpoint too
endpoint = info.get("endpoint")
if endpoint:
try:
req = urllib.request.Request(endpoint, method="GET")
urllib.request.urlopen(req, timeout=5)
return True
except Exception:
return False
return True
# Try HTTP endpoint
endpoint = info.get("endpoint")
if endpoint:
try:
req = urllib.request.Request(endpoint, method="GET")
urllib.request.urlopen(req, timeout=5)
return True
except Exception:
return False
return False
class TokenOptimizer:
"""Applies token optimizations to OpenClaw configuration."""
def __init__(self, dry_run: bool = False):
self.dry_run = dry_run
self.openclaw_dir = Path.home() / '.openclaw'
self.config_path = self.openclaw_dir / 'openclaw.json'
self.backup_dir = self.openclaw_dir / 'backups'
self.templates_dir = Path(__file__).parent.parent / 'templates'
def backup_config(self) -> Optional[Path]:
"""Create backup of existing configuration."""
if not self.config_path.exists():
return None
self.backup_dir.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
backup_path = self.backup_dir / f'openclaw_{timestamp}.json'
if not self.dry_run:
shutil.copy(self.config_path, backup_path)
print(colorize(f"[BACKUP] Config backed up to: {backup_path}", Colors.BLUE))
else:
print(colorize(f"[DRY-RUN] Would backup config to: {backup_path}", Colors.YELLOW))
return backup_path
def load_config(self) -> Dict:
"""Load existing config or return empty dict."""
if self.config_path.exists():
try:
with open(self.config_path, 'r') as f:
return json.load(f)
except json.JSONDecodeError:
print(colorize("[WARNING] Existing config is invalid JSON, starting fresh", Colors.YELLOW))
return {}
def save_config(self, config: Dict):
"""Save configuration to file. In dry-run mode, show a diff preview."""
self.openclaw_dir.mkdir(parents=True, exist_ok=True)
if self.dry_run:
print(colorize("\n[DRY-RUN] Changes preview:", Colors.YELLOW))
existing = self.load_config()
self._show_diff(existing, config)
else:
with open(self.config_path, 'w') as f:
json.dump(config, f, indent=2)
print(colorize(f"[SAVED] Config written to: {self.config_path}", Colors.GREEN))
def generate_optimized_config(self) -> Dict:
"""Generate fully optimized OpenClaw configuration."""
return {
"agents": {
"defaults": {
"model": {
"primary": "anthropic/claude-haiku-4-5"
},
"cache": {
"enabled": True,
"ttl": "5m",
"priority": "high"
},
"models": {
"anthropic/claude-sonnet-4-5": {
"alias": "sonnet",
"cache": True
},
"anthropic/claude-haiku-4-5": {
"alias": "haiku",
"cache": False
},
"anthropic/claude-opus-4-5": {
"alias": "opus",
"cache": True
}
}
}
},
"heartbeat": {
"every": "1h",
"model": "ollama/llama3.2:3b",
"session": "main",
"prompt": "Check: Any blockers, opportunities, or progress updates needed?"
},
"rate_limits": {
"api_calls": {
"min_interval_seconds": 5,
"web_search_interval_seconds": 10,
"max_searches_per_batch": 5,
"batch_cooldown_seconds": 120
}
},
"budgets": {
"daily": 5.00,
"monthly": 200.00,
"warning_threshold": 0.75
},
"_meta": {
"optimized_by": "token-optimizer",
"version": __version__,
"optimized_at": datetime.now().isoformat()
}
}
def merge_config(self, existing: Dict, optimized: Dict) -> Dict:
"""Merge optimized settings into existing config, preserving user customizations."""
def deep_merge(base: Dict, override: Dict) -> Dict:
result = base.copy()
for key, value in override.items():
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
result[key] = deep_merge(result[key], value)
else:
result[key] = value
return result
return deep_merge(existing, optimized)
def apply_model_routing(self, config: Dict) -> Dict:
"""Apply model routing optimization only."""
optimized = self.generate_optimized_config()
if 'agents' not in config:
config['agents'] = {}
if 'defaults' not in config['agents']:
config['agents']['defaults'] = {}
config['agents']['defaults']['model'] = optimized['agents']['defaults']['model']
config['agents']['defaults']['models'] = optimized['agents']['defaults']['models']
print(colorize("[APPLIED] Model routing: Haiku default, Sonnet/Opus aliases", Colors.GREEN))
return config
def apply_heartbeat(self, config: Dict, provider: str = None, model: str = None, fallback: str = None) -> Dict:
"""Apply heartbeat optimization with configurable provider."""
optimized = self.generate_optimized_config()
if provider is None:
provider = resolve_heartbeat_provider(config)
info = HEARTBEAT_PROVIDERS.get(provider, HEARTBEAT_PROVIDERS["ollama"])
if provider == "none":
config.pop("heartbeat", None)
print(colorize("[APPLIED] Heartbeat: disabled", Colors.YELLOW))
return config
heartbeat = optimized['heartbeat']
heartbeat['provider'] = provider
if model:
heartbeat['model'] = f"{info['model_prefix']}{model}"
else:
heartbeat['model'] = f"{info['model_prefix']}{info['default_model']}"
if info.get('endpoint'):
heartbeat['endpoint'] = info['endpoint']
if fallback and fallback in HEARTBEAT_PROVIDERS:
heartbeat['fallback'] = fallback
config['heartbeat'] = heartbeat
print(colorize(f"[APPLIED] Heartbeat: {provider} {heartbeat['model']}", Colors.GREEN))
return config
def apply_caching(self, config: Dict) -> Dict:
"""Apply prompt caching optimization only."""
optimized = self.generate_optimized_config()
if 'agents' not in config:
config['agents'] = {}
if 'defaults' not in config['agents']:
config['agents']['defaults'] = {}
config['agents']['defaults']['cache'] = optimized['agents']['defaults']['cache']
print(colorize("[APPLIED] Prompt caching: Enabled with 5m TTL", Colors.GREEN))
return config
def apply_rate_limits(self, config: Dict) -> Dict:
"""Apply rate limits and budgets."""
optimized = self.generate_optimized_config()
config['rate_limits'] = optimized['rate_limits']
config['budgets'] = optimized['budgets']
print(colorize("[APPLIED] Rate limits and budget controls", Colors.GREEN))
return config
def check_ollama(self) -> bool:
"""Check if Ollama is installed and running."""
if shutil.which('ollama') is None:
return False
try:
req = urllib.request.Request("http://localhost:11434", method="GET")
urllib.request.urlopen(req, timeout=5)
return True
except Exception:
return False
def setup_heartbeat_provider(self, provider: str = "ollama", model: str = None, fallback: str = None) -> bool:
"""Set up heartbeat provider."""
print(colorize(f"\n--- Setting up {provider} for Heartbeat ---", Colors.BOLD))
if provider == "none":
print(colorize("[OK] Heartbeat disabled", Colors.YELLOW))
return True
if provider not in HEARTBEAT_PROVIDERS:
print(colorize(f"[ERROR] Unknown provider: {provider}. Choose from: {', '.join(HEARTBEAT_PROVIDERS.keys())}", Colors.RED))
return False
reachable = check_heartbeat_provider(provider)
if not reachable:
info = HEARTBEAT_PROVIDERS[provider]
endpoint = info.get("endpoint", "")
print(colorize(f"[WARNING] {provider} not reachable at {endpoint}", Colors.YELLOW))
if provider == "ollama":
print(" Install Ollama from: https://ollama.ai")
print(" Then run: ollama pull llama3.2:3b")
elif provider == "lmstudio":
print(" Start LM Studio and enable the local server on port 1234")
elif provider == "groq":
print(" Set GROQ_API_KEY environment variable")
if fallback and fallback != provider:
print(colorize(f"[FALLBACK] Trying fallback provider: {fallback}", Colors.CYAN))
return self.setup_heartbeat_provider(fallback, model)
return False
# Ollama-specific: check/pull model
if provider == "ollama":
return self._setup_ollama_model(model)
print(colorize(f"[OK] {provider} is reachable", Colors.GREEN))
return True
def _setup_ollama_model(self, model: str = None) -> bool:
"""Check Ollama model availability and provide instructions."""
target_model = model or "llama3.2:3b"
print(colorize(f"[OK] Ollama is installed and reachable", Colors.GREEN))
print(colorize(f"[INFO] Make sure the model is available:", Colors.CYAN))
print(f" ollama pull {target_model}")
print(f" ollama serve")
return True
def setup_ollama_heartbeat(self) -> bool:
"""Attempt to set up Ollama for heartbeat (legacy compatibility)."""
return self.setup_heartbeat_provider("ollama")
def list_backups(self) -> List[Path]:
"""List available config backups."""
if not self.backup_dir.exists():
return []
backups = sorted(self.backup_dir.glob('openclaw_*.json'), reverse=True)
return backups
def restore_backup(self, backup_path: Path) -> bool:
"""Restore a config backup."""
if not backup_path.exists():
print(colorize(f"[ERROR] Backup not found: {backup_path}", Colors.RED))
return False
try:
with open(backup_path, 'r') as f:
json.load(f) # validate JSON
except json.JSONDecodeError:
print(colorize(f"[ERROR] Backup is not valid JSON: {backup_path}", Colors.RED))
return False
if self.dry_run:
print(colorize(f"[DRY-RUN] Would restore config from: {backup_path}", Colors.YELLOW))
return True
# Backup current config before restoring
self.backup_config()
shutil.copy(backup_path, self.config_path)
print(colorize(f"[RESTORED] Config restored from: {backup_path}", Colors.GREEN))
return True
def _show_diff(self, old_config: Dict, new_config: Dict):
"""Show colored unified diff between old and new config."""
old_lines = json.dumps(old_config, indent=2).splitlines(keepends=True)
new_lines = json.dumps(new_config, indent=2).splitlines(keepends=True)
diff = list(difflib.unified_diff(
old_lines, new_lines,
fromfile="current config",
tofile="optimized config",
lineterm=""
))
if not diff:
print(colorize(" (no changes)", Colors.YELLOW))
return
for line in diff:
line = line.rstrip('\n')
if line.startswith('+++') or line.startswith('---'):
print(colorize(line, Colors.BOLD))
elif line.startswith('+'):
print(colorize(line, Colors.GREEN))
elif line.startswith('-'):
print(colorize(line, Colors.RED))
elif line.startswith('@@'):
print(colorize(line, Colors.CYAN))
else:
print(line)
def init_stats(self):
"""Initialize or update stats tracking file for benefit reports."""
stats_path = self.openclaw_dir / 'token-optimizer-stats.json'
if stats_path.exists():
try:
with open(stats_path, 'r') as f:
stats = json.load(f)
except json.JSONDecodeError:
stats = {}
else:
stats = {}
if 'installed_at' not in stats:
stats['installed_at'] = datetime.now().isoformat()
stats['last_optimized'] = datetime.now().isoformat()
stats.setdefault('last_benefit_report', None)
stats.setdefault('verify_count', 0)
if not self.dry_run:
with open(stats_path, 'w') as f:
json.dump(stats, f, indent=2)
print(colorize("[STATS] Tracking initialized for savings reports", Colors.BLUE))
def optimize_full(self):
"""Apply all optimizations."""
print(colorize("\n=== Token Optimizer - Full Optimization ===\n", Colors.BOLD + Colors.CYAN))
# Backup existing config
self.backup_config()
# Load existing config
existing = self.load_config()
# Generate and merge optimized config
optimized = self.generate_optimized_config()
final_config = self.merge_config(existing, optimized)
# Apply all optimizations
print(colorize("\nApplying optimizations:", Colors.BOLD))
print(colorize(" [1/4] Model routing (Haiku default)", Colors.GREEN))
print(colorize(" [2/4] Heartbeat to Ollama (free)", Colors.GREEN))
print(colorize(" [3/4] Prompt caching (90% savings)", Colors.GREEN))
print(colorize(" [4/4] Rate limits & budgets", Colors.GREEN))
# Save config
self.save_config(final_config)
# Setup Ollama
self.setup_ollama_heartbeat()
# Generate workspace templates
self.generate_workspace_templates()
# Generate agent prompt additions
self.generate_agent_prompts()
# Initialize stats tracking
self.init_stats()
print(colorize("\n=== Optimization Complete ===", Colors.BOLD + Colors.GREEN))
print("\nNext steps:")
print(" 1. Review generated files in ~/.openclaw/")
print(" 2. Add agent prompt rules from ~/.openclaw/prompts/")
print(" 3. Start Ollama: ollama serve")
print(" 4. Verify with: token-optimizer verify")
def optimize_mode(self, mode: str):
"""Apply specific optimization mode."""
self.backup_config()
config = self.load_config()
if mode == 'routing':
config = self.apply_model_routing(config)
elif mode == 'heartbeat':
config = self.apply_heartbeat(config)
self.setup_ollama_heartbeat()
elif mode == 'caching':
config = self.apply_caching(config)
elif mode == 'limits':
config = self.apply_rate_limits(config)
elif mode == 'full':
self.optimize_full()
return
else:
print(colorize(f"[ERROR] Unknown mode: {mode}", Colors.RED))
return
self.save_config(config)
def generate_workspace_templates(self):
"""Generate optimized workspace file templates."""
workspace_dir = self.openclaw_dir / 'workspace'
workspace_dir.mkdir(parents=True, exist_ok=True)
# SOUL.md template
soul_content = """# SOUL.md - Agent Core Principles
## Identity
[YOUR AGENT NAME/ROLE]
## Core Principles
1. Efficiency first - minimize token usage
2. Quality over quantity - precise responses
3. Proactive communication - surface blockers early
## How to Operate
- Default to Haiku for routine tasks
- Switch to Sonnet only for: architecture, security, complex reasoning
- Batch similar operations together
- Use memory_search() on demand, not auto-load
## Model Selection Rule
```
Default: Always use Haiku
Switch to Sonnet ONLY when:
- Architecture decisions
- Production code review
- Security analysis
- Complex debugging/reasoning
- Strategic multi-project decisions
When in doubt: Try Haiku first.
```
## Rate Limits
- 5s between API calls
- 10s between searches
- Max 5 searches/batch, then 2min break
"""
# USER.md template
user_content = """# USER.md - User Context
## Profile
- **Name:** [YOUR NAME]
- **Timezone:** [YOUR TIMEZONE]
- **Working Hours:** [YOUR HOURS]
## Mission
[WHAT YOU'RE BUILDING]
## Success Metrics
1. [METRIC 1]
2. [METRIC 2]
3. [METRIC 3]
## Communication Preferences
- Brief, actionable updates
- Surface blockers immediately
- Daily summary at end of session
"""
# IDENTITY.md template
identity_content = """# IDENTITY.md - Agent Identity
## Role
[AGENT ROLE - e.g., "Technical Lead", "Research Assistant"]
## Expertise
- [DOMAIN 1]
- [DOMAIN 2]
- [DOMAIN 3]
## Constraints
- Stay within defined budgets
- Follow rate limits strictly
- Escalate uncertainty early
"""
templates = {
'SOUL.md': soul_content,
'USER.md': user_content,
'IDENTITY.md': identity_content
}
print(colorize("\n--- Generating Workspace Templates ---", Colors.BOLD))
for filename, content in templates.items():
filepath = workspace_dir / filename
if filepath.exists() and not self.dry_run:
print(colorize(f" [SKIP] {filename} already exists", Colors.YELLOW))
else:
if not self.dry_run:
with open(filepath, 'w') as f:
f.write(content.strip())
print(colorize(f" [CREATED] {filepath}", Colors.GREEN))
def generate_agent_prompts(self):
"""Generate agent prompt additions for optimization."""
prompts_dir = self.openclaw_dir / 'prompts'
prompts_dir.mkdir(parents=True, exist_ok=True)
# Session initialization rule
session_init = """## SESSION INITIALIZATION RULE
On every session start:
1. Load ONLY these files:
- SOUL.md
- USER.md
- IDENTITY.md
- memory/YYYY-MM-DD.md (if it exists)
2. DO NOT auto-load:
- MEMORY.md
- Session history
- Prior messages
- Previous tool outputs
3. When user asks about prior context:
- Use memory_search() on demand
- Pull only the relevant snippet with memory_get()
- Don't load the whole file
4. Update memory/YYYY-MM-DD.md at end of session with:
- What you worked on
- Decisions made
- Leads generated
- Blockers
- Next steps
This saves 80% on context overhead.
"""
# Model selection rule
model_selection = """## MODEL SELECTION RULE
Default: Always use Haiku
Switch to Sonnet ONLY when:
- Architecture decisions
- Production code review
- Security analysis
- Complex debugging/reasoning
- Strategic multi-project decisions
When in doubt: Try Haiku first.
"""
# Rate limits rule
rate_limits = """## RATE LIMITS
- 5 seconds minimum between API calls
- 10 seconds between web searches
- Max 5 searches per batch, then 2-minute break
- Batch similar work (one request for 10 leads, not 10 requests)
- If you hit 429 error: STOP, wait 5 minutes, retry
## DAILY BUDGET: $5 (warning at 75%)
## MONTHLY BUDGET: $200 (warning at 75%)
"""
# Combined optimization prompt
combined = f"""# TOKEN OPTIMIZATION RULES
Add these rules to your agent prompt:
---
{session_init}
---
{model_selection}
---
{rate_limits}
---
## IMPORTANT
These rules work together to reduce costs by 97%.
Do not remove or modify unless you understand the cost implications.
"""
prompts = {
'session-init.md': session_init,
'model-selection.md': model_selection,
'rate-limits.md': rate_limits,
'OPTIMIZATION-RULES.md': combined
}
print(colorize("\n--- Generating Agent Prompts ---", Colors.BOLD))
for filename, content in prompts.items():
filepath = prompts_dir / filename
if filepath.exists() and not self.dry_run:
print(colorize(f" [SKIP] {filename} already exists", Colors.YELLOW))
else:
if not self.dry_run:
with open(filepath, 'w') as f:
f.write(content.strip())
print(colorize(f" [CREATED] {filepath}", Colors.GREEN))
print(colorize(f"\n[INFO] Add contents of {prompts_dir / 'OPTIMIZATION-RULES.md'} to your agent prompt", Colors.CYAN))
def main():
parser = argparse.ArgumentParser(description='Token Optimizer for OpenClaw')
parser.add_argument('--mode', choices=['full', 'routing', 'heartbeat', 'caching', 'limits'],
default='full', help='Optimization mode')
parser.add_argument('--apply', action='store_true',
help='Apply changes (default is dry-run for safety)')
args = parser.parse_args()
dry_run = not args.apply
if dry_run:
print(colorize("[DRY-RUN] Preview mode. Use --apply to make changes.\n", Colors.YELLOW))
optimizer = TokenOptimizer(dry_run=dry_run)
optimizer.optimize_mode(args.mode)
return 0
if __name__ == '__main__':
sys.exit(main())

387
src/verify.py Normal file
View File

@@ -0,0 +1,387 @@
#!/usr/bin/env python3
"""
Token Optimizer - Verification Module
Verifies optimization setup and estimates savings.
"""
import json
import os
import sys
import shutil
import urllib.request
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Tuple
try:
from src.colors import Colors, colorize
except ImportError:
from colors import Colors, colorize
class OptimizationVerifier:
"""Verifies token optimization setup."""
def __init__(self):
self.openclaw_dir = Path.home() / '.openclaw'
self.config_path = self.openclaw_dir / 'openclaw.json'
self.checks: List[Tuple[str, bool, str]] = []
def load_config(self) -> Dict:
"""Load OpenClaw configuration."""
if self.config_path.exists():
try:
with open(self.config_path, 'r') as f:
return json.load(f)
except json.JSONDecodeError:
return {}
return {}
def check_config_exists(self) -> bool:
"""Check if config file exists."""
exists = self.config_path.exists()
self.checks.append(("Config file exists", exists, str(self.config_path)))
return exists
def check_model_routing(self, config: Dict) -> bool:
"""Check model routing is optimized."""
try:
primary = config.get('agents', {}).get('defaults', {}).get('model', {}).get('primary', '')
is_haiku = 'haiku' in primary.lower()
self.checks.append(("Default model is Haiku", is_haiku, primary or "not set"))
return is_haiku
except Exception:
self.checks.append(("Default model is Haiku", False, "config error"))
return False
def check_model_aliases(self, config: Dict) -> bool:
"""Check model aliases are configured."""
try:
models = config.get('agents', {}).get('defaults', {}).get('models', {})
has_aliases = any('alias' in m for m in models.values())
alias_list = [m.get('alias', '') for m in models.values() if m.get('alias')]
self.checks.append(("Model aliases configured", has_aliases, ', '.join(alias_list) or "none"))
return has_aliases
except Exception:
self.checks.append(("Model aliases configured", False, "config error"))
return False
def check_heartbeat_provider(self, config: Dict) -> bool:
"""Check heartbeat provider is configured and reachable."""
try:
heartbeat = config.get('heartbeat', {})
if not heartbeat:
self.checks.append(("Heartbeat provider configured", False, "not configured"))
return False
provider = heartbeat.get('provider', '')
model = heartbeat.get('model', '')
# Auto-detect provider from model string if not explicit
if not provider:
for name in ('ollama', 'lmstudio', 'groq'):
if name in model.lower():
provider = name
break
if not provider:
provider = 'ollama'
is_free = provider in ('ollama', 'lmstudio', 'none')
label = f"{provider} ({model})" if model else provider
self.checks.append(("Heartbeat provider configured", True, label))
if provider != 'none':
reachable = self.check_provider_reachable(provider, heartbeat.get('endpoint'))
self.checks.append(("Heartbeat provider reachable", reachable,
f"{provider} {'OK' if reachable else 'not reachable'}"))
return True
except Exception:
self.checks.append(("Heartbeat provider configured", False, "config error"))
return False
def check_provider_reachable(self, provider: str, endpoint: str = None) -> bool:
"""Check if a heartbeat provider is reachable."""
if provider == "ollama":
if shutil.which("ollama") is None:
return False
try:
req = urllib.request.Request("http://localhost:11434", method="GET")
urllib.request.urlopen(req, timeout=5)
return True
except Exception:
return False
elif provider in ("lmstudio", "groq"):
url = endpoint or ("http://localhost:1234" if provider == "lmstudio" else "https://api.groq.com")
try:
req = urllib.request.Request(url, method="GET")
urllib.request.urlopen(req, timeout=5)
return True
except Exception:
return False
return False
def check_caching_enabled(self, config: Dict) -> bool:
"""Check prompt caching is enabled."""
try:
cache = config.get('agents', {}).get('defaults', {}).get('cache', {})
enabled = cache.get('enabled', False)
ttl = cache.get('ttl', 'not set')
self.checks.append(("Prompt caching enabled", enabled, f"TTL: {ttl}"))
return enabled
except Exception:
self.checks.append(("Prompt caching enabled", False, "config error"))
return False
def check_rate_limits(self, config: Dict) -> bool:
"""Check rate limits are configured."""
try:
rate_limits = config.get('rate_limits', {})
has_limits = bool(rate_limits)
details = "configured" if has_limits else "not configured"
self.checks.append(("Rate limits configured", has_limits, details))
return has_limits
except Exception:
self.checks.append(("Rate limits configured", False, "config error"))
return False
def check_budgets(self, config: Dict) -> bool:
"""Check budgets are configured."""
try:
budgets = config.get('budgets', {})
daily = budgets.get('daily')
monthly = budgets.get('monthly')
has_budgets = daily is not None or monthly is not None
details = f"daily: ${daily}, monthly: ${monthly}" if has_budgets else "not configured"
self.checks.append(("Budget limits configured", has_budgets, details))
return has_budgets
except Exception:
self.checks.append(("Budget limits configured", False, "config error"))
return False
def check_workspace_files(self) -> bool:
"""Check workspace files exist and are optimized."""
workspace_dir = self.openclaw_dir / 'workspace'
required_files = ['SOUL.md', 'USER.md']
found = []
total_size = 0
for filename in required_files:
filepath = workspace_dir / filename
if filepath.exists():
found.append(filename)
total_size += filepath.stat().st_size
all_found = len(found) == len(required_files)
size_kb = total_size / 1024
is_lean = size_kb < 15 # Less than 15KB is considered lean
self.checks.append(("Workspace files exist", all_found, ', '.join(found) or "none found"))
self.checks.append(("Workspace files are lean", is_lean, f"{size_kb:.1f}KB total"))
return all_found and is_lean
def check_prompts_exist(self) -> bool:
"""Check agent prompt files exist."""
prompts_dir = self.openclaw_dir / 'prompts'
optimization_rules = prompts_dir / 'OPTIMIZATION-RULES.md'
exists = optimization_rules.exists()
self.checks.append(("Optimization prompts generated", exists, str(optimization_rules) if exists else "not found"))
return exists
def calculate_savings(self, config: Dict) -> Dict:
"""Calculate estimated monthly savings."""
savings = {
'model_routing': 0,
'heartbeat': 0,
'caching': 0,
'session': 0,
'total': 0
}
# Model routing savings (Sonnet -> Haiku)
primary = config.get('agents', {}).get('defaults', {}).get('model', {}).get('primary', '')
if 'haiku' in primary.lower():
# Assume 100 calls/day, 2000 tokens each
# Sonnet: 0.003 * 200 = $0.60/day
# Haiku: 0.00025 * 200 = $0.05/day
savings['model_routing'] = (0.60 - 0.05) * 30 # ~$16.50/month
# Heartbeat savings
heartbeat = config.get('heartbeat', {})
if 'ollama' in heartbeat.get('model', '').lower():
# 24 heartbeats/day * 500 tokens * $0.00025/1K
savings['heartbeat'] = 24 * 0.5 * 0.00025 * 30 # ~$0.09/month (small but free)
# Caching savings (90% on repeated content)
cache = config.get('agents', {}).get('defaults', {}).get('cache', {})
if cache.get('enabled'):
# 5KB agent prompt * 100 calls/day * 0.003/1K * 0.9 savings
savings['caching'] = 5 * 100 * 0.003 * 0.9 * 30 # ~$40.50/month
# Session management (estimated from lean context)
savings['session'] = 12.00 # Estimated from guide
savings['total'] = sum(v for k, v in savings.items() if k != 'total')
return savings
def check_benefit_report(self, savings: Dict):
"""Show benefit report every 7 days with donation CTA."""
stats_path = self.openclaw_dir / 'token-optimizer-stats.json'
if not stats_path.exists():
return
try:
with open(stats_path, 'r') as f:
stats = json.load(f)
except (json.JSONDecodeError, IOError):
return
installed_at = stats.get('installed_at')
if not installed_at:
return
try:
install_date = datetime.fromisoformat(installed_at)
except ValueError:
return
now = datetime.now()
days_active = (now - install_date).days
if days_active < 7:
return
# Check if 7 days since last report
last_report = stats.get('last_benefit_report')
if last_report:
try:
last_report_date = datetime.fromisoformat(last_report)
days_since_report = (now - last_report_date).days
if days_since_report < 7:
return
except ValueError:
pass
# Calculate accumulated savings
weekly_savings = savings['total'] / 4.33 # monthly to weekly
total_savings = (savings['total'] / 30) * days_active
yearly_projection = savings['total'] * 12
# Show benefit report
print(colorize("\n +--------------------------------------------------+", Colors.BOLD + Colors.GREEN))
print(colorize(" | Your Savings Report |", Colors.BOLD + Colors.GREEN))
print(colorize(" +--------------------------------------------------+", Colors.GREEN))
print(colorize(f" | Active for: {days_active} days ", Colors.GREEN))
print(colorize(f" | ", Colors.GREEN))
print(colorize(f" | Savings this week: ~${weekly_savings:>8.2f} ", Colors.GREEN))
print(colorize(f" | Savings since install: ~${total_savings:>8.2f} ", Colors.GREEN))
print(colorize(f" | Projected yearly: ~${yearly_projection:>8.2f} ", Colors.GREEN))
print(colorize(f" | ", Colors.GREEN))
print(colorize(f" | Token Optimizer is saving you real money. ", Colors.GREEN))
print(colorize(f" | If it helps, consider a small thank-you: ", Colors.GREEN))
print(colorize(f" | ", Colors.GREEN))
print(colorize(f" | -> https://ko-fi.com/smartpeopleconnected ", Colors.CYAN + Colors.BOLD))
print(colorize(f" | ", Colors.GREEN))
print(colorize(" +--------------------------------------------------+", Colors.GREEN))
# Update last report timestamp
stats['last_benefit_report'] = now.isoformat()
stats['verify_count'] = stats.get('verify_count', 0) + 1
try:
with open(stats_path, 'w') as f:
json.dump(stats, f, indent=2)
except IOError:
pass
def run_verification(self):
"""Run all verification checks."""
print(colorize("\n=== Token Optimizer - Verification ===\n", Colors.BOLD + Colors.CYAN))
# Load config
config = self.load_config()
# Run checks
self.check_config_exists()
self.check_model_routing(config)
self.check_model_aliases(config)
self.check_heartbeat_provider(config)
self.check_caching_enabled(config)
self.check_rate_limits(config)
self.check_budgets(config)
self.check_workspace_files()
self.check_prompts_exist()
# Print results
print(colorize("VERIFICATION RESULTS:", Colors.BOLD))
print("-" * 60)
passed = 0
failed = 0
for name, status, details in self.checks:
if status:
icon = colorize("[PASS]", Colors.GREEN)
passed += 1
else:
icon = colorize("[FAIL]", Colors.RED)
failed += 1
print(f" {icon} {name}")
print(colorize(f" {details}", Colors.BLUE))
print("-" * 60)
# Summary
total = passed + failed
score = (passed / total) * 100 if total > 0 else 0
if score == 100:
status_color = Colors.GREEN
status_text = "FULLY OPTIMIZED"
elif score >= 70:
status_color = Colors.YELLOW
status_text = "PARTIALLY OPTIMIZED"
else:
status_color = Colors.RED
status_text = "NEEDS OPTIMIZATION"
print(colorize(f"\nStatus: {status_text}", Colors.BOLD + status_color))
print(f"Score: {passed}/{total} checks passed ({score:.0f}%)")
# Calculate and show savings
savings = self.calculate_savings(config)
print(colorize("\n--- ESTIMATED MONTHLY SAVINGS ---", Colors.BOLD))
print(f" Model Routing: ${savings['model_routing']:.2f}")
print(f" Heartbeat: ${savings['heartbeat']:.2f}")
print(f" Prompt Caching: ${savings['caching']:.2f}")
print(f" Session Mgmt: ${savings['session']:.2f}")
print(colorize(f" TOTAL: ${savings['total']:.2f}/month", Colors.BOLD + Colors.GREEN))
print(colorize(f" YEARLY: ${savings['total'] * 12:.2f}/year", Colors.GREEN))
# Recommendations
if failed > 0:
print(colorize("\n--- RECOMMENDATIONS ---", Colors.BOLD + Colors.YELLOW))
for name, status, details in self.checks:
if not status:
print(f" - Fix: {name}")
print(colorize("\nRun 'token-optimizer optimize' to apply missing optimizations", Colors.CYAN))
# Show benefit report (every 7 days)
self.check_benefit_report(savings)
return failed == 0
def main():
verifier = OptimizationVerifier()
success = verifier.run_verification()
return 0 if success else 1
if __name__ == '__main__':
sys.exit(main())

View File

@@ -0,0 +1,88 @@
# TOKEN OPTIMIZATION RULES
Add these rules to your agent prompt for 97% cost reduction.
---
## SESSION INITIALIZATION RULE
On every session start:
1. Load ONLY these files:
- SOUL.md
- USER.md
- IDENTITY.md
- memory/YYYY-MM-DD.md (if it exists)
2. DO NOT auto-load:
- MEMORY.md
- Session history
- Prior messages
- Previous tool outputs
3. When user asks about prior context:
- Use memory_search() on demand
- Pull only the relevant snippet with memory_get()
- Don't load the whole file
4. Update memory/YYYY-MM-DD.md at end of session with:
- What you worked on
- Decisions made
- Leads generated
- Blockers
- Next steps
**This saves 80% on context overhead.**
---
## MODEL SELECTION RULE
Default: Always use Haiku
Switch to Sonnet ONLY when:
- Architecture decisions
- Production code review
- Security analysis
- Complex debugging/reasoning
- Strategic multi-project decisions
When in doubt: Try Haiku first.
---
## RATE LIMITS
- 5 seconds minimum between API calls
- 10 seconds between web searches
- Max 5 searches per batch, then 2-minute break
- Batch similar work (one request for 10 leads, not 10 requests)
- If you hit 429 error: STOP, wait 5 minutes, retry
## DAILY BUDGET: $5 (warning at 75%)
## MONTHLY BUDGET: $200 (warning at 75%)
---
## COST AWARENESS
Before any operation, consider:
1. Can this be batched with similar operations?
2. Is this the minimum model needed for this task?
3. Am I loading only necessary context?
4. Will this push me over budget limits?
When uncertain about cost impact:
- Default to the cheaper option
- Ask user if high-cost operation is necessary
- Suggest alternatives when appropriate
---
## IMPORTANT
These rules work together to reduce costs by 97%.
Do not remove or modify unless you understand the cost implications.
Expected cost reduction:
- Before: $1,500+/month
- After: $30-50/month

84
templates/SOUL.md Normal file
View File

@@ -0,0 +1,84 @@
# SOUL.md - Agent Core Principles
## Identity
[YOUR AGENT NAME/ROLE - e.g., "Technical Assistant", "Research Agent"]
## Core Principles
1. **Efficiency first** - minimize token usage without sacrificing quality
2. **Precision over verbosity** - concise, actionable responses
3. **Proactive communication** - surface blockers and decisions early
4. **Batching mindset** - group similar operations together
## Operating Rules
### Model Selection Rule
```
DEFAULT: Always use Haiku
SWITCH TO SONNET only when:
- Architecture decisions affecting multiple systems
- Production code review (security implications)
- Security analysis or vulnerability assessment
- Complex debugging requiring deep reasoning
- Strategic decisions spanning multiple projects
SWITCH TO OPUS only when:
- Mission-critical decisions with high stakes
- Novel problems with no established patterns
- User explicitly requests highest capability
WHEN IN DOUBT: Try Haiku first. Escalate if results insufficient.
```
### Session Initialization Rule
```
ON EVERY SESSION START:
1. Load ONLY these files:
- SOUL.md (this file)
- USER.md (user context)
- IDENTITY.md (if exists)
- memory/YYYY-MM-DD.md (today's notes, if exists)
2. DO NOT auto-load:
- MEMORY.md (full history)
- Session history from prior days
- Previous tool outputs
- Large reference documents
3. When user asks about prior context:
- Use memory_search() on demand
- Pull only relevant snippet with memory_get()
- Never load entire files preemptively
4. At session end, update memory/YYYY-MM-DD.md with:
- Work completed
- Decisions made
- Open blockers
- Next steps
```
### Rate Limits
```
- 5 seconds minimum between API calls
- 10 seconds between web searches
- Maximum 5 searches per batch, then 2-minute cooldown
- Batch similar operations (one request for 10 items, not 10 requests)
- On 429 error: STOP, wait 5 minutes, then retry
```
### Budget Awareness
```
DAILY BUDGET: $5 (alert at 75%)
MONTHLY BUDGET: $200 (alert at 75%)
If approaching limits:
1. Notify user immediately
2. Suggest deferring non-urgent work
3. Switch to lower-cost model if appropriate
```
## Quality Standards
- Verify before acting (read files before editing)
- Test changes when possible
- Document decisions for future reference
- Ask clarifying questions rather than assume

45
templates/USER.md Normal file
View File

@@ -0,0 +1,45 @@
# USER.md - User Context
## Profile
- **Name:** [YOUR NAME]
- **Role:** [YOUR ROLE - e.g., "Founder", "Developer", "Researcher"]
- **Timezone:** [YOUR TIMEZONE - e.g., "America/New_York", "UTC"]
- **Working Hours:** [YOUR HOURS - e.g., "9am-6pm EST"]
## Mission
[Brief description of what you're building or working toward]
Example: "Building an AI-powered sales automation platform that helps B2B companies generate qualified leads."
## Current Focus
[What you're working on this week/sprint]
Example:
- Launch MVP by end of month
- Integrate with CRM systems
- Optimize lead scoring algorithm
## Success Metrics
1. [METRIC 1 - e.g., "Generate 100 qualified leads per week"]
2. [METRIC 2 - e.g., "Reduce response time to under 5 minutes"]
3. [METRIC 3 - e.g., "Achieve 90% customer satisfaction"]
## Communication Preferences
- **Updates:** [e.g., "Brief, bullet-pointed summaries"]
- **Questions:** [e.g., "Ask immediately, don't assume"]
- **Blockers:** [e.g., "Escalate within 30 minutes if stuck"]
- **Tone:** [e.g., "Direct and professional"]
## Tools & Stack
- **Primary Language:** [e.g., "Python", "TypeScript"]
- **Framework:** [e.g., "FastAPI", "Next.js"]
- **Database:** [e.g., "PostgreSQL", "MongoDB"]
- **Deployment:** [e.g., "AWS", "Vercel"]
## Important Context
[Any critical information the agent should always keep in mind]
Example:
- We're pre-revenue, optimize for speed over perfection
- Main competitor is X, avoid their approach to Y
- Legal review required for any customer-facing copy

View File

@@ -0,0 +1,64 @@
{
"$schema": "https://openclaw.ai/schemas/config.json",
"$comment": "Optimized OpenClaw configuration - saves 97% on token costs",
"agents": {
"defaults": {
"model": {
"primary": "anthropic/claude-haiku-4-5",
"$comment": "Haiku as default - 12x cheaper than Sonnet, sufficient for 80% of tasks"
},
"cache": {
"enabled": true,
"ttl": "5m",
"priority": "high",
"$comment": "90% discount on cached tokens within 5-minute window"
},
"models": {
"anthropic/claude-sonnet-4-5": {
"alias": "sonnet",
"cache": true,
"$comment": "Use for: architecture, security analysis, complex reasoning"
},
"anthropic/claude-haiku-4-5": {
"alias": "haiku",
"cache": false,
"$comment": "Default for routine tasks - fast and cheap"
},
"anthropic/claude-opus-4-5": {
"alias": "opus",
"cache": true,
"$comment": "Reserve for mission-critical decisions only"
}
}
}
},
"heartbeat": {
"every": "1h",
"provider": "ollama",
"model": "ollama/llama3.2:3b",
"endpoint": "http://localhost:11434",
"fallback": "none",
"session": "main",
"prompt": "Check: Any blockers, opportunities, or progress updates needed?",
"$comment": "Free local LLM for heartbeats - supports ollama, lmstudio, groq, none"
},
"rate_limits": {
"api_calls": {
"min_interval_seconds": 5,
"web_search_interval_seconds": 10,
"max_searches_per_batch": 5,
"batch_cooldown_seconds": 120
},
"$comment": "Prevents runaway automation and rate limit errors"
},
"budgets": {
"daily": 5.00,
"monthly": 200.00,
"warning_threshold": 0.75,
"$comment": "Hard limits prevent surprise bills"
}
}

380
test/simulation_test.py Normal file
View File

@@ -0,0 +1,380 @@
#!/usr/bin/env python3
"""
Token Optimizer - Simulation Test
Demonstrates before/after performance comparison with mock data.
"""
import json
import os
import sys
import shutil
from pathlib import Path
from datetime import datetime
# Fix Windows encoding
if sys.platform == 'win32':
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
# Add src to path
sys.path.insert(0, str(Path(__file__).parent.parent / 'src'))
from analyzer import OpenClawAnalyzer
from optimizer import TokenOptimizer
# ANSI colors
class Colors:
RED = '\033[91m'
GREEN = '\033[92m'
YELLOW = '\033[93m'
BLUE = '\033[94m'
CYAN = '\033[96m'
BOLD = '\033[1m'
END = '\033[0m'
def colorize(text, color):
if sys.stdout.isatty():
return f"{color}{text}{Colors.END}"
return text
# Cost constants
COSTS = {
'opus': {'input': 0.015, 'output': 0.075},
'sonnet': {'input': 0.003, 'output': 0.015},
'haiku': {'input': 0.00025, 'output': 0.00125},
'ollama': {'input': 0.0, 'output': 0.0}
}
def create_mock_environment(test_dir: Path):
"""Create a mock OpenClaw environment for testing."""
print(colorize("\n=== CREATING MOCK OPENCLAW ENVIRONMENT ===\n", Colors.BOLD + Colors.CYAN))
# Create directory structure
openclaw_dir = test_dir / '.openclaw'
workspace_dir = openclaw_dir / 'workspace'
workspace_dir.mkdir(parents=True, exist_ok=True)
# Create UNOPTIMIZED config (typical default setup)
unoptimized_config = {
"agents": {
"defaults": {
"model": {
"primary": "anthropic/claude-sonnet-4-5" # Expensive default!
}
}
},
"heartbeat": {
"every": "1h",
"model": "anthropic/claude-sonnet-4-5", # Paid API for heartbeats!
"prompt": "Status check"
}
# No caching, no budgets, no rate limits
}
config_path = openclaw_dir / 'openclaw.json'
with open(config_path, 'w') as f:
json.dump(unoptimized_config, f, indent=2)
# Create BLOATED workspace files (typical unoptimized setup)
# Large SOUL.md (15KB - too big!)
soul_content = """# SOUL.md - Agent Configuration
## Identity
You are an AI assistant...
## Detailed History
""" + "\n".join([f"- Historical entry {i}: Lorem ipsum dolor sit amet, consectetur adipiscing elit. Sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris." for i in range(100)])
with open(workspace_dir / 'SOUL.md', 'w') as f:
f.write(soul_content)
# Large MEMORY.md (25KB - loaded every time!)
memory_content = """# MEMORY.md - Full History
## All Previous Sessions
""" + "\n".join([f"### Session {i}\nUser asked about topic {i}. Assistant responded with detailed explanation about subject {i}. This conversation covered multiple aspects including technical details, examples, and follow-up questions. The user was satisfied with the response.\n" for i in range(200)])
with open(workspace_dir / 'MEMORY.md', 'w') as f:
f.write(memory_content)
# Large USER.md (10KB)
user_content = """# USER.md - User Profile
## Complete User History
""" + "\n".join([f"- Preference {i}: User likes detailed explanations with examples and code snippets when relevant." for i in range(150)])
with open(workspace_dir / 'USER.md', 'w') as f:
f.write(user_content)
print(f" Created mock OpenClaw directory: {openclaw_dir}")
print(f" Config: Sonnet default, paid heartbeats, no caching")
print(f" Workspace files: ~50KB total (bloated)")
return openclaw_dir
def calculate_costs(config: dict, workspace_size_kb: float, daily_messages: int = 100, daily_heartbeats: int = 24):
"""Calculate estimated daily/monthly costs."""
# Determine model
model_name = config.get('agents', {}).get('defaults', {}).get('model', {}).get('primary', '')
if 'haiku' in model_name.lower():
model = 'haiku'
elif 'opus' in model_name.lower():
model = 'opus'
else:
model = 'sonnet'
# Heartbeat model
hb_model_name = config.get('heartbeat', {}).get('model', '')
if 'ollama' in hb_model_name.lower():
hb_model = 'ollama'
elif 'haiku' in hb_model_name.lower():
hb_model = 'haiku'
else:
hb_model = 'sonnet'
# Caching
cache_enabled = config.get('agents', {}).get('defaults', {}).get('cache', {}).get('enabled', False)
cache_discount = 0.9 if cache_enabled else 0.0
# Calculate costs
costs = COSTS[model]
hb_costs = COSTS[hb_model]
# Context tokens (workspace loaded each message)
context_tokens = workspace_size_kb * 250 # ~250 tokens per KB
# Message costs (context + average 500 token response)
avg_output_tokens = 500
# Per-message cost
input_cost = (context_tokens / 1000) * costs['input']
output_cost = (avg_output_tokens / 1000) * costs['output']
# Apply cache discount to input (agent prompt)
if cache_enabled:
# First message full price, subsequent 90% off
cached_input_cost = input_cost * (1 - cache_discount * 0.8) # 80% of messages cached
else:
cached_input_cost = input_cost
message_cost = cached_input_cost + output_cost
# Heartbeat cost
hb_tokens = 500 # tokens per heartbeat
hb_cost = (hb_tokens / 1000) * (hb_costs['input'] + hb_costs['output'])
# Daily costs
daily_message_cost = message_cost * daily_messages
daily_hb_cost = hb_cost * daily_heartbeats
daily_total = daily_message_cost + daily_hb_cost
return {
'model': model,
'heartbeat_model': hb_model,
'cache_enabled': cache_enabled,
'context_tokens': context_tokens,
'per_message_cost': message_cost,
'daily_message_cost': daily_message_cost,
'daily_heartbeat_cost': daily_hb_cost,
'daily_total': daily_total,
'monthly_total': daily_total * 30,
'yearly_total': daily_total * 365
}
def print_cost_report(title: str, costs: dict, color: str):
"""Print a formatted cost report."""
print(colorize(f"\n{'='*60}", color))
print(colorize(f" {title}", Colors.BOLD + color))
print(colorize(f"{'='*60}", color))
print(f"\n Configuration:")
print(f" Model: {costs['model'].upper()}")
print(f" Heartbeat: {costs['heartbeat_model'].upper()}")
print(f" Caching: {'Enabled' if costs['cache_enabled'] else 'Disabled'}")
print(f" Context: {costs['context_tokens']:,.0f} tokens ({costs['context_tokens']/250:.1f}KB)")
print(f"\n Per-Message Cost: ${costs['per_message_cost']:.4f}")
print(f"\n Daily Costs:")
print(f" Messages (100): ${costs['daily_message_cost']:.2f}")
print(f" Heartbeats: ${costs['daily_heartbeat_cost']:.2f}")
print(colorize(f" TOTAL: ${costs['daily_total']:.2f}", Colors.BOLD))
print(f"\n Projected Costs:")
print(colorize(f" Monthly: ${costs['monthly_total']:.2f}", Colors.BOLD))
print(colorize(f" Yearly: ${costs['yearly_total']:.2f}", Colors.BOLD))
def run_simulation():
"""Run the full simulation test."""
print(colorize("""
+---------------------------------------------------------------+
| |
| TOKEN OPTIMIZER - SIMULATION TEST |
| |
| Demonstrates before/after performance comparison |
| |
+---------------------------------------------------------------+
""", Colors.BOLD + Colors.CYAN))
# Setup test directory
test_dir = Path(__file__).parent / 'mock_environment'
if test_dir.exists():
shutil.rmtree(test_dir)
test_dir.mkdir(parents=True)
# Create mock environment
openclaw_dir = create_mock_environment(test_dir)
# ========== BEFORE OPTIMIZATION ==========
print(colorize("\n\n" + "="*60, Colors.RED))
print(colorize(" PHASE 1: BEFORE OPTIMIZATION (Typical Default Setup)", Colors.BOLD + Colors.RED))
print(colorize("="*60, Colors.RED))
# Load unoptimized config
with open(openclaw_dir / 'openclaw.json') as f:
before_config = json.load(f)
# Calculate workspace size
workspace_dir = openclaw_dir / 'workspace'
before_workspace_size = sum(f.stat().st_size for f in workspace_dir.iterdir() if f.is_file()) / 1024
print(f"\n Workspace size: {before_workspace_size:.1f}KB")
print(f" Files loaded every message: SOUL.md, MEMORY.md, USER.md")
before_costs = calculate_costs(before_config, before_workspace_size)
print_cost_report("BEFORE OPTIMIZATION - COST ANALYSIS", before_costs, Colors.RED)
# ========== APPLY OPTIMIZATION ==========
print(colorize("\n\n" + "="*60, Colors.YELLOW))
print(colorize(" PHASE 2: APPLYING TOKEN OPTIMIZER", Colors.BOLD + Colors.YELLOW))
print(colorize("="*60, Colors.YELLOW))
# Create optimized config
optimized_config = {
"agents": {
"defaults": {
"model": {
"primary": "anthropic/claude-haiku-4-5" # Cheap default
},
"cache": {
"enabled": True,
"ttl": "5m",
"priority": "high"
},
"models": {
"anthropic/claude-sonnet-4-5": {"alias": "sonnet", "cache": True},
"anthropic/claude-haiku-4-5": {"alias": "haiku", "cache": False},
"anthropic/claude-opus-4-5": {"alias": "opus", "cache": True}
}
}
},
"heartbeat": {
"every": "1h",
"model": "ollama/llama3.2:3b", # FREE local LLM
"prompt": "Status check"
},
"rate_limits": {
"api_calls": {"min_interval_seconds": 5}
},
"budgets": {
"daily": 5.00,
"monthly": 150.00,
"warning_threshold": 0.75
}
}
# Save optimized config
with open(openclaw_dir / 'openclaw.json', 'w') as f:
json.dump(optimized_config, f, indent=2)
print("\n Optimizations applied:")
print(colorize(" [OK] Model routing: Haiku default (92% cheaper)", Colors.GREEN))
print(colorize(" [OK] Heartbeat: Ollama local (100% free)", Colors.GREEN))
print(colorize(" [OK] Prompt caching: Enabled (90% discount)", Colors.GREEN))
print(colorize(" [OK] Budget controls: $5/day, $150/month", Colors.GREEN))
# Create lean workspace files
lean_soul = """# SOUL.md
## Core Principles
- Efficiency first
- Use Haiku for routine tasks
- Sonnet only for complex reasoning
"""
lean_user = """# USER.md
- Name: User
- Preference: Concise responses
"""
# Remove bloated files, create lean ones
(workspace_dir / 'MEMORY.md').unlink() # Don't auto-load
with open(workspace_dir / 'SOUL.md', 'w') as f:
f.write(lean_soul)
with open(workspace_dir / 'USER.md', 'w') as f:
f.write(lean_user)
print(colorize(" [OK] Workspace: Reduced from 50KB to 2KB", Colors.GREEN))
print(colorize(" [OK] Memory: On-demand loading only", Colors.GREEN))
# ========== AFTER OPTIMIZATION ==========
print(colorize("\n\n" + "="*60, Colors.GREEN))
print(colorize(" PHASE 3: AFTER OPTIMIZATION", Colors.BOLD + Colors.GREEN))
print(colorize("="*60, Colors.GREEN))
# Calculate new workspace size
after_workspace_size = sum(f.stat().st_size for f in workspace_dir.iterdir() if f.is_file()) / 1024
print(f"\n Workspace size: {after_workspace_size:.1f}KB")
print(f" Files loaded: SOUL.md, USER.md only (lean)")
after_costs = calculate_costs(optimized_config, after_workspace_size)
print_cost_report("AFTER OPTIMIZATION - COST ANALYSIS", after_costs, Colors.GREEN)
# ========== COMPARISON ==========
print(colorize("\n\n" + "="*60, Colors.BOLD + Colors.CYAN))
print(colorize(" SAVINGS SUMMARY", Colors.BOLD + Colors.CYAN))
print(colorize("="*60, Colors.BOLD + Colors.CYAN))
daily_savings = before_costs['daily_total'] - after_costs['daily_total']
monthly_savings = before_costs['monthly_total'] - after_costs['monthly_total']
yearly_savings = before_costs['yearly_total'] - after_costs['yearly_total']
savings_percent = (1 - after_costs['monthly_total'] / before_costs['monthly_total']) * 100
print(f"\n {'Metric':<20} {'Before':>12} {'After':>12} {'Savings':>12}")
print(f" {'-'*56}")
print(f" {'Daily Cost':<20} ${before_costs['daily_total']:>10.2f} ${after_costs['daily_total']:>10.2f} {colorize(f'${daily_savings:>10.2f}', Colors.GREEN)}")
print(f" {'Monthly Cost':<20} ${before_costs['monthly_total']:>10.2f} ${after_costs['monthly_total']:>10.2f} {colorize(f'${monthly_savings:>10.2f}', Colors.GREEN)}")
print(f" {'Yearly Cost':<20} ${before_costs['yearly_total']:>10.2f} ${after_costs['yearly_total']:>10.2f} {colorize(f'${yearly_savings:>10.2f}', Colors.GREEN)}")
print(colorize(f"\n +-------------------------------------------------------+", Colors.BOLD + Colors.GREEN))
print(colorize(f" | TOTAL SAVINGS: {savings_percent:.0f}% |", Colors.BOLD + Colors.GREEN))
print(colorize(f" | ${monthly_savings:.2f}/month = ${yearly_savings:.2f}/year |", Colors.BOLD + Colors.GREEN))
print(colorize(f" +-------------------------------------------------------+", Colors.BOLD + Colors.GREEN))
# Breakdown
print(colorize("\n Savings Breakdown:", Colors.BOLD))
model_savings = (COSTS['sonnet']['input'] - COSTS['haiku']['input']) / COSTS['sonnet']['input'] * 100
print(f" - Model Routing (Sonnet->Haiku): {model_savings:.0f}% per token")
print(f" - Heartbeat (Paid->Ollama): 100% (free)")
print(f" - Context Reduction (50KB->2KB): 96% less tokens")
print(f" - Prompt Caching: 90% on repeated content")
# Cleanup
shutil.rmtree(test_dir)
print(colorize("\n\n[OK] Simulation complete! Mock environment cleaned up.", Colors.GREEN))
print(colorize("\nTo apply these optimizations to your real OpenClaw setup:", Colors.CYAN))
print(" python src/optimizer.py --mode full\n")
if __name__ == '__main__':
run_simulation()