0Pricing
AI Agents · 课时

逐步分析令牌消耗与成本

衡量每次工具调用和每个推理步骤的令牌消耗

逐步分析令牌消耗与成本 是 CoddyKit 上的免费 AI Agents 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

为什么要分析令牌使用情况

LLM API 成本会随着令牌使用量直接增长。一次代理运行可能进行数十次 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 调用的美元成本。成本通常按每百万个令牌计算，因此：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
    }

调用前估算令牌数

使用 tiktoken 在进行 API 调用前估算令牌数。这样您就可以强制执行预算限制，并及早检测异常大的提示词。

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"]}')

成本预算与中止

通过设置每次运行和每日预算上限，防止代理成本失控。如果运行超出预算，请以优雅方式中止，并返回部分结果，而不是继续产生开支。

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')

令牌使用量警报

当单次代理运行超过预期令牌使用量时发出警报。意外峰值通常表示存在错误：上下文无限增长、重复调用工具，或缺少截断逻辑。

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 响应中捕获使用量；使用模型定价表计算每个步骤的成本；跟踪每次代理运行的累计成本；按工具调用类型细分成本；通过防护检查强制执行预算限制；以及持久化成本数据以进行趋势分析和优化。

常见问题解答

「逐步分析令牌消耗与成本」课时是免费的吗？

是的 — 「逐步分析令牌消耗与成本」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「逐步分析令牌消耗与成本」这节课中我会学到什么？

衡量每次工具调用和每个推理步骤的令牌消耗 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。

「逐步分析令牌消耗与成本」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 使用 LangSmith 和 Langfuse 分析追踪记录
  2. 逐步分析令牌消耗与成本
  3. 识别缓慢且昂贵的步骤
  4. 分析智能体故障的根本原因
← 返回 AI Agents