0Pricing
AI Agents · レッスン

ステップごとのトークン数とコストのプロファイリング

ツール呼び出しごと、推論ステップごとのトークン消費量を測定します。

「ステップごとのトークン数とコストのプロファイリング」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン2/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。

トークン使用量をプロファイリングする理由

LLM APIのコストは、トークン使用量に応じて直接増加します。エージェントの1回の実行で、数十回のLLM呼び出しが行われることもあります。ステップごとのプロファイリングを行わなければ、どのステップにコストがかかっているのか、どこをキャッシュすべきか、どうコストを削減すべきかを把握できません。

OpenAIからトークン使用量を読み取る

OpenAIのすべての補完レスポンスには、usageオブジェクトが含まれており、prompt_tokens、completion_tokens、total_tokensを確認できます。必ず取得してください。

import openai

client = openai.OpenAI(api_key='sk-...')

def call_llm_with_tracking(prompt: str, model: str = 'gpt-4o-mini') -> dict:
    response = client.chat.completions.create(
        model=model,
        messages=[{'role': 'user', 'content': prompt}]
    )
    
    usage = response.usage
    return {
        'content': response.choices[0].message.content,
        'prompt_tokens': usage.prompt_tokens,
        'completion_tokens': usage.completion_tokens,
        'total_tokens': usage.total_tokens,
        'model': model
    }

result = call_llm_with_tracking('What is the capital of France?')
print(f'Response: {result["content"]}')
print(f'Tokens - Prompt: {result["prompt_tokens"]}, Completion: {result["completion_tokens"]}, Total: {result["total_tokens"]}')

呼び出しごとのコスト計算

料金表を使って、各LLM呼び出しのドル換算コストを計算します。通常、コストは100万トークン単位で設定されるため、計算式は次のようになります。cost = (prompt_tokens / 1_000_000) * input_price + (completion_tokens / 1_000_000) * output_price。

# Pricing per million tokens (as of early 2025 - verify current prices)
MODEL_PRICING = {
    'gpt-4o': {'input': 2.50, 'output': 10.00},
    'gpt-4o-mini': {'input': 0.15, 'output': 0.60},
    'gpt-4-turbo': {'input': 10.00, 'output': 30.00},
    'claude-3-5-sonnet-20241022': {'input': 3.00, 'output': 15.00},
    'claude-3-haiku-20240307': {'input': 0.25, 'output': 1.25}
}

def calculate_cost(prompt_tokens: int, completion_tokens: int, model: str) -> float:
    pricing = MODEL_PRICING.get(model)
    if not pricing:
        return 0.0
    
    input_cost = (prompt_tokens / 1_000_000) * pricing['input']
    output_cost = (completion_tokens / 1_000_000) * pricing['output']
    return input_cost + output_cost

# Example
prompt_tokens = 500
completion_tokens = 200
model = 'gpt-4o-mini'
cost = calculate_cost(prompt_tokens, completion_tokens, model)
print(f'Cost for {prompt_tokens}+{completion_tokens} tokens on {model}: ${cost:.6f}')

累積コストトラッカー

エージェントの実行全体にわたる累積コストを追跡します。コストトラッカーは、ステップごとのトークン使用量とコストを累積するため、どのステップが最も多くの予算を消費したかを簡単に確認できます。

from dataclasses import dataclass, field
from typing import List

@dataclass
class StepCost:
    step_name: str
    model: str
    prompt_tokens: int
    completion_tokens: int
    cost_usd: float

@dataclass
class CostTracker:
    steps: List[StepCost] = field(default_factory=list)
    
    def record(self, step_name: str, model: str, prompt_tokens: int, completion_tokens: int):
        cost = calculate_cost(prompt_tokens, completion_tokens, model)
        self.steps.append(StepCost(
            step_name=step_name,
            model=model,
            prompt_tokens=prompt_tokens,
            completion_tokens=completion_tokens,
            cost_usd=cost
        ))
    
    @property
    def total_cost(self) -> float:
        return sum(s.cost_usd for s in self.steps)
    
    @property
    def total_tokens(self) -> int:
        return sum(s.prompt_tokens + s.completion_tokens for s in self.steps)
    
    def summary(self) -> str:
        lines = ['=== Cost Summary ===']
        for step in self.steps:
            lines.append(f'{step.step_name}: {step.prompt_tokens}+{step.completion_tokens} tokens = ${step.cost_usd:.6f}')
        lines.append(f'TOTAL: {self.total_tokens} tokens = ${self.total_cost:.6f}')
        return '\n'.join(lines)

tracker = CostTracker()
tracker.record('entity_extraction', 'gpt-4o-mini', 200, 50)
tracker.record('vector_search_query', 'gpt-4o-mini', 100, 30)
tracker.record('answer_generation', 'gpt-4o-mini', 1500, 300)
print(tracker.summary())

ツール呼び出しの種類ごとのコスト

複数のエージェント実行にわたって、ツール呼び出しの種類ごとにコストを分解します。特定のツールは呼び出し頻度が大幅に高く、主要なコスト要因になっていることがあります。

from collections import defaultdict

class ToolCostAnalyzer:
    def __init__(self):
        self.tool_stats = defaultdict(lambda: {
            'call_count': 0,
            'total_prompt_tokens': 0,
            'total_completion_tokens': 0,
            'total_cost_usd': 0.0
        })
    
    def record_tool_call(self, tool_name: str, prompt_tokens: int, completion_tokens: int, model: str):
        cost = calculate_cost(prompt_tokens, completion_tokens, model)
        stats = self.tool_stats[tool_name]
        stats['call_count'] += 1
        stats['total_prompt_tokens'] += prompt_tokens
        stats['total_completion_tokens'] += completion_tokens
        stats['total_cost_usd'] += cost
    
    def report(self):
        print('=== Tool Cost Breakdown ===')
        sorted_tools = sorted(
            self.tool_stats.items(),
            key=lambda x: x[1]['total_cost_usd'],
            reverse=True
        )
        for tool_name, stats in sorted_tools:
            avg_cost = stats['total_cost_usd'] / stats['call_count']
            print(f'{tool_name}: {stats["call_count"]} calls, total ${stats["total_cost_usd"]:.4f}, avg ${avg_cost:.6f}/call')

analyzer = ToolCostAnalyzer()
analyzer.record_tool_call('search_web', 800, 200, 'gpt-4o-mini')
analyzer.record_tool_call('search_web', 750, 180, 'gpt-4o-mini')
analyzer.record_tool_call('read_email', 300, 100, 'gpt-4o-mini')
analyzer.record_tool_call('generate_report', 2000, 500, 'gpt-4o')
analyzer.report()

エージェントループへのコスト追跡の統合

LLM呼び出し関数をラップして、エージェントループの一部としてコストを自動的に追跡します。トラッカーを各所に渡し、すべての呼び出しがセッションの合計に加算されるようにします。

import openai

client = openai.OpenAI(api_key='sk-...')

def tracked_completion(tracker: CostTracker, step_name: str, messages: list, model: str = 'gpt-4o-mini') -> str:
    response = client.chat.completions.create(
        model=model,
        messages=messages
    )
    usage = response.usage
    tracker.record(
        step_name=step_name,
        model=model,
        prompt_tokens=usage.prompt_tokens,
        completion_tokens=usage.completion_tokens
    )
    return response.choices[0].message.content

def run_agent_with_cost_tracking(question: str) -> dict:
    tracker = CostTracker()
    
    # Step 1: Entity extraction
    entities_str = tracked_completion(
        tracker, 'entity_extraction',
        [{'role': 'user', 'content': f'Extract entities from: {question}'}]
    )
    
    # Step 2: Answer generation
    answer = tracked_completion(
        tracker, 'answer_generation',
        [{'role': 'user', 'content': question}]
    )
    
    return {
        'answer': answer,
        'cost_summary': tracker.summary(),
        'total_cost_usd': tracker.total_cost
    }

呼び出し前にトークン数を見積もる

APIを呼び出す前に、tiktokenを使ってトークン数を見積もります。これにより、予算上限を適用し、想定外に大きなプロンプトを早期に検出できます。

import tiktoken

DEFAULT_ENCODER = tiktoken.encoding_for_model('gpt-4o-mini')

def estimate_tokens(text: str, model: str = 'gpt-4o-mini') -> int:
    try:
        encoding = tiktoken.encoding_for_model(model)
    except KeyError:
        encoding = DEFAULT_ENCODER
    return len(encoding.encode(text))

def check_prompt_budget(messages: list, max_tokens: int = 8000) -> dict:
    total = 0
    breakdown = []
    for msg in messages:
        count = estimate_tokens(msg.get('content', ''))
        total += count
        breakdown.append({'role': msg['role'], 'tokens': count})
    
    return {
        'total_tokens': total,
        'within_budget': total <= max_tokens,
        'budget': max_tokens,
        'breakdown': breakdown
    }

messages = [
    {'role': 'system', 'content': 'You are a helpful assistant that...'},
    {'role': 'user', 'content': 'Explain the concept of quantum entanglement in simple terms.'}
]
result = check_prompt_budget(messages)
print(f'Total tokens: {result["total_tokens"]}, Within budget: {result["within_budget"]}')

コスト予算と打ち切り

実行ごとおよび1日ごとの予算上限を設定し、エージェントのコストが際限なく増加することを防ぎます。実行が予算を超えた場合は、支出を続けるのではなく、部分的な結果を返して適切に中止します。

class BudgetGuard:
    def __init__(self, max_cost_per_run: float = 0.10, max_cost_per_day: float = 5.00):
        self.max_run = max_cost_per_run
        self.max_day = max_cost_per_day
        self.day_spend = 0.0
    
    def check_and_spend(self, tracker: 'CostTracker', about_to_spend_estimate: float = 0.001):
        if tracker.total_cost >= self.max_run:
            raise RuntimeError(
                f'Run budget exceeded: ${tracker.total_cost:.4f} >= ${self.max_run}'
            )
        if self.day_spend + tracker.total_cost >= self.max_day:
            raise RuntimeError(
                f'Daily budget exceeded: ${self.day_spend:.4f} daily spend'
            )
    
    def finalize_run(self, tracker: 'CostTracker'):
        self.day_spend += tracker.total_cost
        print(f'Run cost: ${tracker.total_cost:.6f}, Day total: ${self.day_spend:.4f}')

guard = BudgetGuard(max_cost_per_run=0.05, max_cost_per_day=2.00)
tracker = CostTracker()
tracker.record('test_step', 'gpt-4o-mini', 100, 50)
guard.check_and_spend(tracker)
guard.finalize_run(tracker)

分析のためのコストデータ保存

傾向分析、請求先への帰属、最適化の判断に利用できるよう、実行コストのデータをデータベースに保存します。ほとんどのエージェントでは、シンプルなSQLiteテーブルで十分に対応できます。

import sqlite3
from datetime import datetime

def init_cost_db(db_path: str = 'agent_costs.db'):
    conn = sqlite3.connect(db_path)
    conn.execute('''
        CREATE TABLE IF NOT EXISTS run_costs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            run_id TEXT NOT NULL,
            step_name TEXT NOT NULL,
            model TEXT NOT NULL,
            prompt_tokens INTEGER,
            completion_tokens INTEGER,
            cost_usd REAL,
            timestamp TEXT
        )
    ''')
    conn.commit()
    return conn

def save_run_costs(conn, run_id: str, tracker: 'CostTracker'):
    for step in tracker.steps:
        conn.execute(
            'INSERT INTO run_costs (run_id, step_name, model, prompt_tokens, completion_tokens, cost_usd, timestamp) VALUES (?, ?, ?, ?, ?, ?, ?)',
            (run_id, step.step_name, step.model, step.prompt_tokens, step.completion_tokens, step.cost_usd, datetime.utcnow().isoformat())
        )
    conn.commit()
    print(f'Saved {len(tracker.steps)} cost records for run {run_id}')

conn = init_cost_db()
print('Cost database initialized')

トークン使用量のアラート

エージェントの1回の実行で、想定されるトークン使用量を超えたときにアラートを発生させます。予期しない急増は、コンテキストが無限に増加している、ツールを繰り返し呼び出している、切り詰め処理がないといったバグを示していることがよくあります。

def check_token_spike(tracker: 'CostTracker', expected_max_tokens: int = 10000) -> dict:
    total = tracker.total_tokens
    if total > expected_max_tokens:
        # Find the biggest steps
        sorted_steps = sorted(tracker.steps, key=lambda s: s.prompt_tokens + s.completion_tokens, reverse=True)
        top_steps = [
            {'step': s.step_name, 'tokens': s.prompt_tokens + s.completion_tokens}
            for s in sorted_steps[:3]
        ]
        message = (
            f'Token spike: {total} tokens (expected <= {expected_max_tokens}). '
            f'Top consumers: {top_steps}'
        )
        print(f'ALERT: {message}')
        return {'alert': True, 'total_tokens': total, 'message': message, 'top_steps': top_steps}
    return {'alert': False, 'total_tokens': total}

tracker = CostTracker()
tracker.record('context_builder', 'gpt-4o-mini', 8000, 200)  # Unusually large prompt
result = check_token_spike(tracker, expected_max_tokens=5000)
print('Spike check:', result['alert'], '-', result.get('message', 'OK'))

コストレポート用クエリ

コストデータベースをクエリして、モデルごとの日次支出、最もコストの高いステップ、時間の経過に伴うコストの傾向などのレポートを生成します。これらの情報を最適化の判断に役立てます。

import sqlite3
from datetime import datetime, timedelta

def cost_report(db_path: str = 'agent_costs.db', days: int = 7) -> dict:
    conn = sqlite3.connect(db_path)
    since = (datetime.utcnow() - timedelta(days=days)).isoformat()
    
    # Total cost by model
    model_costs = conn.execute('''
        SELECT model, SUM(cost_usd) as total_cost, COUNT(*) as call_count
        FROM run_costs WHERE timestamp >= ?
        GROUP BY model ORDER BY total_cost DESC
    ''', (since,)).fetchall()
    
    # Top expensive steps
    step_costs = conn.execute('''
        SELECT step_name, SUM(cost_usd) as total_cost, AVG(cost_usd) as avg_cost
        FROM run_costs WHERE timestamp >= ?
        GROUP BY step_name ORDER BY total_cost DESC LIMIT 10
    ''', (since,)).fetchall()
    
    conn.close()
    return {
        'period_days': days,
        'by_model': [{'model': r[0], 'total_usd': r[1], 'calls': r[2]} for r in model_costs],
        'by_step': [{'step': r[0], 'total_usd': r[1], 'avg_usd': r[2]} for r in step_costs]
    }

print('Cost report function defined')

理解度チェック：トークンとコストのプロファイリング

ステップごとのトークンとコストのプロファイリングについて理解度を確認します。

コストプロファイリングのまとめ

効果的なコストプロファイリングには、すべてのAPIレスポンスから使用量を取得すること、モデルの料金表を使ってステップごとのコストを計算すること、エージェントの実行ごとに累積コストを追跡すること、ツール呼び出しの種類ごとにコストを分解すること、ガードチェックで予算上限を適用すること、傾向分析と最適化のためにコストデータを保存することが必要です。

よくある質問

「ステップごとのトークン数とコストのプロファイリング」レッスンは無料ですか？

はい。「ステップごとのトークン数とコストのプロファイリング」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。

「ステップごとのトークン数とコストのプロファイリング」で何を学びますか？

ツール呼び出しごと、推論ステップごとのトークン消費量を測定します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

AI Agentsを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。

「ステップごとのトークン数とコストのプロファイリング」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このAI Agentsレッスンでコードを書いて実行できますか？

はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. LangSmith と Langfuse によるトレース分析
  2. ステップごとのトークン数とコストのプロファイリング
  3. 遅くてコストの高いステップを特定する
  4. エージェントの失敗に対する根本原因分析
← AI Agentsに戻る