feat: usage tracking + knowledge import script
- usage_tracker.py: daily token/call buckets per user (home/{user}/usage.json)
- Hook into local backend (OpenAI usage field) and Gemini API (usage_metadata)
- Claude/Gemini CLI backends produce no structured token data and are not tracked
- Fix CLAUDE.md stale tool count (27 → 39) and refresh tool list
- scripts/import_knowledge.py: walk markdown dirs, chunk by H2, call local LLM
for summaries, create AE journal entries with path-derived tags; resumable via
state file; --dry-run and --limit flags for safe testing
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
75
cortex/usage_tracker.py
Normal file
75
cortex/usage_tracker.py
Normal file
@@ -0,0 +1,75 @@
|
||||
"""
|
||||
API usage and token tracking.
|
||||
|
||||
Writes daily buckets to home/{username}/usage.json:
|
||||
|
||||
{
|
||||
"2026-05-01": {
|
||||
"gemini_api/gemini-2.0-flash": {"calls": 3, "prompt_tokens": 8400, "completion_tokens": 520},
|
||||
"local/llama3.2:latest": {"calls": 2, "prompt_tokens": 1200, "completion_tokens": 310}
|
||||
}
|
||||
}
|
||||
|
||||
Claude CLI and Gemini CLI backends produce no structured token data and are not tracked.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
from config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_LOCK = asyncio.Lock()
|
||||
|
||||
|
||||
def _usage_path(username: str) -> Path:
|
||||
return settings.home_root() / username / "usage.json"
|
||||
|
||||
|
||||
async def record(
|
||||
username: str,
|
||||
backend: str,
|
||||
model_name: str,
|
||||
prompt_tokens: int,
|
||||
completion_tokens: int,
|
||||
) -> None:
|
||||
"""Append one call's token counts to the daily usage log for this user.
|
||||
|
||||
backend — "gemini_api" | "local"
|
||||
model_name — the exact model string (e.g. "gemini-2.0-flash", "llama3.2:latest")
|
||||
"""
|
||||
path = _usage_path(username)
|
||||
today = date.today().isoformat()
|
||||
key = f"{backend}/{model_name}"
|
||||
|
||||
async with _LOCK:
|
||||
try:
|
||||
data: dict = json.loads(path.read_text()) if path.exists() else {}
|
||||
except Exception:
|
||||
data = {}
|
||||
|
||||
entry = data.setdefault(today, {}).setdefault(
|
||||
key, {"calls": 0, "prompt_tokens": 0, "completion_tokens": 0}
|
||||
)
|
||||
entry["calls"] += 1
|
||||
entry["prompt_tokens"] += prompt_tokens
|
||||
entry["completion_tokens"] += completion_tokens
|
||||
|
||||
try:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(data, indent=2))
|
||||
except Exception as e:
|
||||
logger.warning("Failed to write usage data to %s: %s", path, e)
|
||||
|
||||
|
||||
def read_usage(username: str) -> dict:
|
||||
"""Return the full usage dict for this user. Empty dict if no file yet."""
|
||||
path = _usage_path(username)
|
||||
try:
|
||||
return json.loads(path.read_text()) if path.exists() else {}
|
||||
except Exception:
|
||||
return {}
|
||||
Reference in New Issue
Block a user