Tokeny wejściowe
- Prompty i instrukcje
- Kontekst z plików
- Historia poprzednich rozmów
- Wiadomości systemowe
Zarządzanie kosztami Claude Code na skalę przedsiębiorstwa wymaga strategicznego planowania, monitorowania i optymalizacji. Ten przewodnik zawiera kompleksowe strategie kontroli wydatków przy jednoczesnej maksymalizacji wartości.
Tokeny wejściowe
Tokeny wyjściowe
Ukryte koszty
Skonfiguruj śledzenie użycia
# Utwórz skrypt monitorującycat > monitor-usage.sh << 'EOF'#!/bin/bash
# KonfiguracjaCOST_THRESHOLD=100 # Dzienny limit w USDALERT_EMAIL="tech-team@company.com"
# Śledź użycietrack_usage() { local user=$1 local tokens=$2 local cost=$3
# Zapisz do bazy danych echo "INSERT INTO usage_logs (user, tokens, cost, timestamp) VALUES ('$user', $tokens, $cost, NOW());" | mysql usage_db
# Sprawdź progi DAILY_TOTAL=$(echo "SELECT SUM(cost) FROM usage_logs WHERE DATE(timestamp) = CURDATE();" | mysql usage_db)
if (( $(echo "$DAILY_TOTAL > $COST_THRESHOLD" | bc -l) )); then send_alert "Przekroczono dzienny próg kosztów: \$$DAILY_TOTAL" fi}
# Monitoruj użycie Claude Codeclaude monitor --format json | while read -r line; do USER=$(echo $line | jq -r '.user') TOKENS=$(echo $line | jq -r '.tokens') COST=$(echo $line | jq -r '.cost') track_usage "$USER" "$TOKENS" "$COST"doneEOF
chmod +x monitor-usage.sh
Utwórz dashboard kosztów
import pandas as pdimport plotly.graph_objects as gofrom datetime import datetime, timedeltaimport mysql.connector
class CostDashboard: def __init__(self, db_config): self.conn = mysql.connector.connect(**db_config)
def get_usage_data(self, days=30): query = """ SELECT DATE(timestamp) as date, user, SUM(tokens) as total_tokens, SUM(cost) as total_cost FROM usage_logs WHERE timestamp >= DATE_SUB(NOW(), INTERVAL %s DAY) GROUP BY DATE(timestamp), user ORDER BY date DESC """ df = pd.read_sql(query, self.conn, params=[days]) return df
def create_cost_chart(self): df = self.get_usage_data()
fig = go.Figure()
# Dodaj ślady dla każdego użytkownika for user in df['user'].unique(): user_data = df[df['user'] == user] fig.add_trace(go.Scatter( x=user_data['date'], y=user_data['total_cost'], mode='lines+markers', name=user, stackgroup='one' ))
fig.update_layout( title='Koszty użycia Claude Code według użytkownika', xaxis_title='Data', yaxis_title='Koszt (USD)', hovermode='x unified' )
return fig
def generate_report(self): df = self.get_usage_data()
# Oblicz statystyki total_cost = df['total_cost'].sum() avg_daily_cost = df.groupby('date')['total_cost'].sum().mean() top_users = df.groupby('user')['total_cost'].sum().nlargest(5)
report = f""" # Raport użycia Claude Code
## Podsumowanie (ostatnie 30 dni) - Całkowity koszt: ${total_cost:.2f} - Średni koszt dzienny: ${avg_daily_cost:.2f} - Łącznie tokenów: {df['total_tokens'].sum():,}
## Top użytkownicy według kosztów {top_users.to_string()}
## Trendy kosztów Zobacz załączoną wizualizację """
return report
Implementuj alerty
groups: - name: claude_code_costs interval: 5m rules: - alert: HighTokenUsage expr: | sum(rate(claude_code_tokens_total[5m])) by (user) > 10000 for: 10m labels: severity: warning annotations: summary: "Wysokie użycie tokenów dla użytkownika {{ $labels.user }}" description: "{{ $labels.user }} używa {{ $value }} tokenów/minutę"
- alert: CostThresholdExceeded expr: | sum(claude_code_cost_total) by (department) > 1000 for: 5m labels: severity: critical annotations: summary: "Dział {{ $labels.department }} przekroczył budżet" description: "Aktualny koszt: ${{ $value }}"
System alokacji budżetu
from datetime import datetimefrom enum import Enumimport json
class BudgetPeriod(Enum): DAILY = "daily" WEEKLY = "weekly" MONTHLY = "monthly"
class BudgetManager: def __init__(self, config_file="budgets.json"): with open(config_file) as f: self.config = json.load(f) self.usage = {}
def check_budget(self, department, user, estimated_tokens): """Sprawdź, czy żądanie mieści się w budżecie""" dept_config = self.config.get(department, {})
# Sprawdź budżet działu dept_limit = dept_config.get('limit', float('inf')) dept_usage = self.get_usage(department, dept_config['period'])
if dept_usage + self.estimate_cost(estimated_tokens) > dept_limit: return False, f"Przekroczono budżet działu: ${dept_usage:.2f}/${dept_limit:.2f}"
# Sprawdź budżet użytkownika user_limit = dept_config.get('user_limits', {}).get(user, float('inf')) user_usage = self.get_usage(f"{department}:{user}", dept_config['period'])
if user_usage + self.estimate_cost(estimated_tokens) > user_limit: return False, f"Przekroczono budżet użytkownika: ${user_usage:.2f}/${user_limit:.2f}"
return True, "Sprawdzenie budżetu pomyślne"
def record_usage(self, department, user, tokens, cost): """Zapisz rzeczywiste użycie""" timestamp = datetime.now()
# Zapisz użycie działu if department not in self.usage: self.usage[department] = [] self.usage[department].append({ 'timestamp': timestamp, 'tokens': tokens, 'cost': cost })
# Zapisz użycie użytkownika user_key = f"{department}:{user}" if user_key not in self.usage: self.usage[user_key] = [] self.usage[user_key].append({ 'timestamp': timestamp, 'tokens': tokens, 'cost': cost })
def get_usage(self, key, period): """Pobierz użycie dla bieżącego okresu""" if key not in self.usage: return 0.0
# Oblicz początek okresu now = datetime.now() if period == BudgetPeriod.DAILY: period_start = now.replace(hour=0, minute=0, second=0) elif period == BudgetPeriod.WEEKLY: period_start = now - timedelta(days=now.weekday()) else: # Monthly period_start = now.replace(day=1, hour=0, minute=0, second=0)
# Zsumuj użycie w okresie total = sum( entry['cost'] for entry in self.usage[key] if entry['timestamp'] >= period_start )
return total
def estimate_cost(self, tokens): """Oszacuj koszt dla liczby tokenów""" # Ceny Claude-3 (przykład) return tokens * 0.000015 # $15 za milion tokenów
{ "engineering": { "limit": 500, "period": "daily", "user_limits": { "senior_dev_1": 100, "senior_dev_2": 100, "junior_dev_1": 50, "intern_1": 25 }, "model_restrictions": { "claude-4.1-opus": ["senior_dev_1", "senior_dev_2"], "claude-4-sonnet": "*", "claude-instant": "*" } }, "marketing": { "limit": 200, "period": "weekly", "user_limits": { "content_lead": 150, "writer_1": 50 }, "model_restrictions": { "claude-4.1-opus": ["content_lead"], "claude-4-sonnet": "*" } }, "support": { "limit": 1000, "period": "monthly", "user_limits": {}, "model_restrictions": { "claude-instant": "*" } }}
departments: engineering: limit: 500 period: daily users: - name: senior_dev_1 limit: 100 models: [claude-4.1-opus, claude-4-sonnet] - name: senior_dev_2 limit: 100 models: [claude-4.1-opus, claude-4-sonnet] - name: junior_dev_1 limit: 50 models: [claude-4-sonnet, claude-instant]
marketing: limit: 200 period: weekly users: - name: content_lead limit: 150 models: [claude-4.1-opus, claude-4-sonnet] - name: writer_1 limit: 50 models: [claude-4-sonnet]
Efektywne promptowanie
class PromptOptimizer: def __init__(self): self.token_counter = TokenCounter()
def optimize_prompt(self, prompt, max_tokens=2000): """Optymalizuj prompt w celu redukcji tokenów"""
# Usuń niepotrzebne białe znaki prompt = ' '.join(prompt.split())
# Skompresuj powtarzające się instrukcje prompt = self.compress_instructions(prompt)
# Usuń redundantny kontekst prompt = self.remove_redundancy(prompt)
# Skróć jeśli potrzeba if self.token_counter.count(prompt) > max_tokens: prompt = self.smart_truncate(prompt, max_tokens)
return prompt
def compress_instructions(self, prompt): """Zamień rozwlekłe instrukcje na zwięzłe""" replacements = { "Proszę przeanalizuj następujący kod i podaj": "Przeanalizuj:", "Czy możesz pomóc mi zrozumieć": "Wyjaśnij:", "Chciałbym żebyś": "", "Czy mógłbyś proszę": "" }
for verbose, concise in replacements.items(): prompt = prompt.replace(verbose, concise)
return prompt
def batch_similar_requests(self, requests): """Grupuj podobne żądania razem""" batched = {}
for req in requests: # Grupuj według typu żądania req_type = self.classify_request(req) if req_type not in batched: batched[req_type] = [] batched[req_type].append(req)
# Utwórz prompty grupowe batch_prompts = [] for req_type, reqs in batched.items(): if len(reqs) > 1: batch_prompt = f"Przetwórz te {len(reqs)} żądania typu {req_type}:\n" for i, req in enumerate(reqs, 1): batch_prompt += f"\n{i}. {req}" batch_prompts.append(batch_prompt) else: batch_prompts.extend(reqs)
return batch_prompts
Zarządzanie kontekstem
class ContextManager: def __init__(self, max_context_tokens=4000): self.max_tokens = max_context_tokens self.context_cache = {}
def prepare_context(self, files, focus_file=None): """Przygotuj zoptymalizowany kontekst z plików""" context_parts = [] token_budget = self.max_tokens
# Priorytet 1: Plik główny if focus_file and focus_file in files: content = self.summarize_file(files[focus_file]) tokens = self.count_tokens(content) if tokens <= token_budget * 0.5: # Maks 50% dla pliku głównego context_parts.append(f"=== {focus_file} ===\n{content}") token_budget -= tokens
# Priorytet 2: Powiązane pliki related = self.find_related_files(focus_file, files) for file in related: if token_budget <= 0: break
summary = self.create_summary(files[file]) tokens = self.count_tokens(summary)
if tokens <= token_budget: context_parts.append(f"=== {file} (podsumowanie) ===\n{summary}") token_budget -= tokens
return "\n\n".join(context_parts)
def create_summary(self, content): """Utwórz efektywne tokenowo podsumowanie""" lines = content.split('\n')
# Wydobądź kluczowe elementy imports = [l for l in lines if l.strip().startswith('import')] functions = [l for l in lines if 'def ' in l or 'function ' in l] classes = [l for l in lines if 'class ' in l]
summary = [] if imports: summary.append("Importy: " + ", ".join(imports[:5])) if functions: summary.append("Funkcje: " + ", ".join(functions[:10])) if classes: summary.append("Klasy: " + ", ".join(classes[:5]))
return "\n".join(summary)
Implementuj inteligentne routowanie
class ModelRouter: def __init__(self): self.models = { 'claude-4.1-opus': { 'cost_per_1k': 0.015, 'capabilities': ['complex_code', 'architecture', 'analysis'] }, 'claude-4-sonnet': { 'cost_per_1k': 0.003, 'capabilities': ['general_code', 'refactoring', 'testing'] }, 'claude-instant': { 'cost_per_1k': 0.0008, 'capabilities': ['simple_code', 'formatting', 'comments'] } }
def select_model(self, task_type, complexity, budget_remaining): """Wybierz optymalny model na podstawie zadania i budżetu"""
# Mapuj zadanie na wymagane możliwości required_caps = self.get_required_capabilities(task_type)
# Filtruj zdolne modele capable_models = [ model for model, info in self.models.items() if any(cap in info['capabilities'] for cap in required_caps) ]
# Sortuj według kosztów capable_models.sort(key=lambda m: self.models[m]['cost_per_1k'])
# Sprawdź budżet for model in capable_models: estimated_cost = self.estimate_task_cost(model, complexity) if estimated_cost <= budget_remaining: return model
return None # Brak modelu w budżecie
def estimate_task_cost(self, model, complexity): """Oszacuj koszt zadania""" base_tokens = { 'simple': 1000, 'moderate': 5000, 'complex': 15000 }
tokens = base_tokens.get(complexity, 5000) cost_per_token = self.models[model]['cost_per_1k'] / 1000
return tokens * cost_per_token
Utwórz reguły routowania
rules: - pattern: "napraw.*błąd|debug|error" model: claude-4-sonnet max_tokens: 2000
- pattern: "zaimplementuj.*funkcję|utwórz.*od zera" model: claude-4.1-opus max_tokens: 8000
- pattern: "formatuj|lint|komentarz" model: claude-instant max_tokens: 1000
- pattern: "test|test jednostkowy|test integracyjny" model: claude-4-sonnet max_tokens: 3000
- pattern: "architektura|design|refaktor.*duży" model: claude-4.1-opus max_tokens: 10000
default: model: claude-4-sonnet max_tokens: 4000
Inteligentny system cache
import hashlibimport jsonimport redisfrom datetime import timedelta
class ResponseCache: def __init__(self, redis_host='localhost', ttl_hours=24): self.redis = redis.Redis(host=redis_host, decode_responses=True) self.ttl = timedelta(hours=ttl_hours)
def get_cache_key(self, prompt, model, params): """Generuj deterministyczny klucz cache""" cache_data = { 'prompt': prompt, 'model': model, 'params': params }
# Utwórz hash żądania cache_string = json.dumps(cache_data, sort_keys=True) return f"claude:cache:{hashlib.sha256(cache_string.encode()).hexdigest()}"
def get(self, prompt, model, params): """Pobierz odpowiedź z cache jeśli dostępna""" key = self.get_cache_key(prompt, model, params) cached = self.redis.get(key)
if cached: # Zaktualizuj licznik dostępów self.redis.hincrby(f"{key}:meta", "hits", 1) return json.loads(cached)
return None
def set(self, prompt, model, params, response, tokens_used): """Cachuj odpowiedź z metadanymi""" key = self.get_cache_key(prompt, model, params)
# Przechowaj odpowiedź self.redis.setex( key, self.ttl, json.dumps(response) )
# Przechowuj metadane meta_key = f"{key}:meta" self.redis.hset(meta_key, mapping={ 'prompt_tokens': len(prompt.split()), 'response_tokens': tokens_used, 'model': model, 'created': datetime.now().isoformat(), 'hits': 0 }) self.redis.expire(meta_key, self.ttl)
def get_cache_stats(self): """Pobierz statystyki cachowania""" stats = { 'total_cached': 0, 'total_hits': 0, 'tokens_saved': 0, 'cost_saved': 0 }
# Przeskanuj wszystkie klucze cache for key in self.redis.scan_iter("claude:cache:*:meta"): meta = self.redis.hgetall(key) stats['total_cached'] += 1 hits = int(meta.get('hits', 0)) stats['total_hits'] += hits
if hits > 0: tokens = int(meta.get('response_tokens', 0)) stats['tokens_saved'] += tokens * hits
# Oblicz zaoszczędzone koszty (przykładowe stawki) model = meta.get('model', 'claude-4-sonnet') if model == 'claude-4.1-opus': cost_per_token = 0.000015 elif model == 'claude-4-sonnet': cost_per_token = 0.000003 else: cost_per_token = 0.0000008
stats['cost_saved'] += tokens * hits * cost_per_token
return stats
Utwórz kompleksowe raporty
class ExecutiveReport: def __init__(self, db_connection): self.db = db_connection
def generate_monthly_report(self): """Generuj miesięczny raport wykonawczy"""
# Pobierz dane costs = self.get_monthly_costs() usage = self.get_usage_patterns() savings = self.calculate_savings()
report = f""" # Raport wykonawczy Claude Code - {datetime.now().strftime('%B %Y')}
## Przegląd finansowy - Całkowity wydatek: ${costs['total']:,.2f} - Wykorzystanie budżetu: {costs['utilization']}% - Koszt na programistę: ${costs['per_developer']:,.2f} - Zmiana m/m: {costs['mom_change']:+.1f}%
## Wgląd w użycie - Aktywni użytkownicy: {usage['active_users']} - Łączne żądania: {usage['total_requests']:,} - Średnio tokenów/żądanie: {usage['avg_tokens']:,} - Szczyt użycia: {usage['peak_time']}
## Optymalizacja kosztów - Oszczędności z cachowania: ${savings['caching']:,.2f} - Oszczędności z routowania modeli: ${savings['routing']:,.2f} - Oszczędności z optymalizacji promptów: ${savings['prompts']:,.2f} - Łączne oszczędności: ${savings['total']:,.2f}
## Podział według działów {self.format_department_table(costs['by_department'])}
## Rekomendacje {self.generate_recommendations(costs, usage, savings)} """
return report
def generate_recommendations(self, costs, usage, savings): """Generuj rekomendacje optymalizacji kosztów""" recs = []
# Sprawdź użytkowników o wysokich kosztach if costs['top_user_percentage'] > 20: recs.append("- Rozważ dodatkowe szkolenie dla użytkowników o wysokim użyciu")
# Sprawdź wskaźnik trafień cache if savings['cache_hit_rate'] < 30: recs.append("- Ulepsz strategię cachowania aby zwiększyć wskaźnik trafień")
# Sprawdź dystrybucję modeli if costs['opus_percentage'] > 50: recs.append("- Przejrzyj wybór modeli aby używać Sonnet gdzie to możliwe")
return "\n".join(recs) if recs else "- Brak natychmiastowych działań wymaganych"
Automatyzuj dystrybucję raportów
import smtplibfrom email.mime.multipart import MIMEMultipartfrom email.mime.text import MIMETextfrom email.mime.image import MIMEImage
def send_monthly_report(recipients, report_html, charts): """Wyślij miesięczny raport do kadry zarządzającej"""
msg = MIMEMultipart('related') msg['Subject'] = f"Raport użycia Claude Code - {datetime.now().strftime('%B %Y')}" msg['From'] = 'ai-ops@company.com' msg['To'] = ', '.join(recipients)
# Dołącz raport HTML msg.attach(MIMEText(report_html, 'html'))
# Dołącz wykresy for i, chart in enumerate(charts): img = MIMEImage(chart) img.add_header('Content-ID', f'<chart{i}>') msg.attach(img)
# Wyślij email with smtplib.SMTP('smtp.company.com', 587) as server: server.starttls() server.login('ai-ops@company.com', os.environ['EMAIL_PASSWORD']) server.send_message(msg)
Ograniczenia czasowe
# Pozwalaj na drogie modele tylko w godzinach pracydef check_time_restriction(model, user): current_hour = datetime.now().hour
if model == 'claude-4.1-opus': if current_hour < 8 or current_hour > 18: return False, "Model Opus ograniczony do godzin pracy"
return True, "Dozwolone"
Ograniczanie żądań
# Implementuj ograniczanie szybkości żądańfrom collections import defaultdictimport time
class RateLimiter: def __init__(self): self.requests = defaultdict(list)
def check_rate_limit(self, user, limit=10, window=60): now = time.time()
# Wyczyść stare żądania self.requests[user] = [ req for req in self.requests[user] if now - req < window ]
# Sprawdź limit if len(self.requests[user]) >= limit: return False, f"Przekroczono limit: {limit} żądań na {window}s"
# Zapisz żądanie self.requests[user].append(now) return True, "Dozwolone"
Implementacja wyłącznika bezpieczeństwa
class CostCircuitBreaker: def __init__(self, daily_limit=1000, emergency_limit=1500): self.daily_limit = daily_limit self.emergency_limit = emergency_limit self.current_usage = 0 self.emergency_mode = False
def check_request(self, estimated_cost, user, priority='normal'): """Sprawdź czy żądanie powinno być dozwolone"""
if self.current_usage >= self.emergency_limit: # Całkowite wyłączenie return False, "Osiągnięto limit awaryjny - wszystkie żądania zablokowane"
if self.current_usage >= self.daily_limit: # Tryb awaryjny - tylko krytyczne żądania if priority != 'critical': return False, "Osiągnięto dzienny limit - dozwolone tylko krytyczne żądania" self.emergency_mode = True
if self.emergency_mode: # Powiadom administratorów self.send_emergency_alert(user, estimated_cost)
return True, "Żądanie zatwierdzone"
def send_emergency_alert(self, user, cost): """Powiadom administratorów o awaryjnym użyciu""" alert = f""" PILNE: Aktywny tryb awaryjny Claude Code
Aktualne użycie: ${self.current_usage:.2f} Dzienny limit: ${self.daily_limit:.2f} Limit awaryjny: ${self.emergency_limit:.2f}
Żądanie od: {user} Szacowany koszt: ${cost:.2f}
Wymagane działanie: Przejrzyj i zatwierdź tylko krytyczne żądania. """
# Wyślij do Slack, email, itp. send_slack_alert('#ai-ops-emergency', alert)
Kontynuuj optymalizację kosztów z:
Pamiętaj: Efektywna kontrola kosztów to kwestia równowagi. Skup się na maksymalizacji wartości przy minimalizacji marnotrawstwa, nie tylko na redukcji kosztów. Regularne monitorowanie i optymalizacja mogą często osiągnąć 30-50% redukcji kosztów bez wpływu na produktywność.