PythonでAIエージェントのトークンコストを追跡する: usage、予算、停止

エージェントが10分間動いただけで、タスクそのものより請求額が大きくなった。原因を突き止めるには、すべてのAPIレスポンスに含まれるusageオブジェクトを読み、タスクごとに合計し、モデルの100万トークンあたりの価格を掛ければよい。上限を設けるには、各ステップの後にその累計を予算と比較し、超えたらループを止める。以下に示す小さなPythonクラスは、まさにそれを行う。

この例ではOpenAI互換エンドポイントhttps://api.cheaptokens.dev/v1とモデルIDdeepseek-ai/DeepSeek-V4-Flash-0731（コンテキスト380K、1回答あたり最大32,768出力トークン）を使用する。実行を始める前に見積もりたい場合は別の話であり、AIエージェントのコストを見積もる方法で扱っている。本記事は実行中に測定し、上限を設けることについてだ。

数字はどこにあるか: usageオブジェクト#

非ストリーミングのチャット補完は、必ずusageオブジェクトを返す。DeepSeekのAPIリファレンス（2026年10月26日確認）には、以下のフィールドが記載されている。

• prompt_tokens — 今回送信したすべて: システムプロンプト、履歴、ツールの結果。
• completion_tokens — モデルが返したもの。
• total_tokens — 上記2つを足したもの。

DeepSeekはオプションの追加項目も文書化している: prompt_cache_hit_tokensとprompt_cache_miss_tokens（合計するとprompt_tokensになる）、および思考モデル用のcompletion_tokens_details.reasoning_tokens。このページには公開日がなく、これらの追加項目はOpenAI形式の拡張であるため、以下のコードではオプションとして扱い、最初の2つのフィールドのみに依存する。

なぜ最後のステップだけでなく、すべてのステップをログに記録するのか？ エージェントは毎回、成長するコンテキストを再送信するからだ。ステップ1では2,000入力トークンを送信し、ステップ30では9,000を送信するかもしれず、請求額はそれらすべての合計である。理由は長いチャットが高くつく理由で説明されており、入力/出力の内訳は入力トークンと出力トークンにある。

エージェントを止めるメーター#

全体はこうだ: トークンを合計し、ステップごとに1行を出力し、予算を超えたらエラーを発生させるメーター、およびその周りの素のエージェントループ。SDKをpip install openaiでインストールし、キーをCHEAPTOKENS_API_KEYに入れる。

import os
from openai import OpenAI

client = OpenAI(
    base_url='https://api.cheaptokens.dev/v1',
    api_key=os.environ['CHEAPTOKENS_API_KEY'],
)
MODEL = 'deepseek-ai/DeepSeek-V4-Flash-0731'

# example prices, USD per 1M tokens: use the current ones from the pricing table
PRICE_IN = 0.10
PRICE_OUT = 0.40


class BudgetExceeded(Exception):
    pass


class TaskMeter:
    def __init__(self, budget_usd):
        self.budget = budget_usd
        self.steps = 0
        self.tokens_in = 0
        self.tokens_out = 0
        self.reasoning = 0

    @property
    def cost(self):
        return (self.tokens_in * PRICE_IN + self.tokens_out * PRICE_OUT) / 1_000_000

    def add(self, usage):
        self.steps += 1
        self.tokens_in += usage.prompt_tokens
        self.tokens_out += usage.completion_tokens
        details = getattr(usage, 'completion_tokens_details', None)
        self.reasoning += getattr(details, 'reasoning_tokens', 0) or 0
        print(f'step {self.steps}: in={usage.prompt_tokens} '
              f'out={usage.completion_tokens} task=${self.cost:.4f}')
        if self.cost > self.budget:
            raise BudgetExceeded(
                f'budget ${self.budget:.2f} crossed at step {self.steps}: '
                f'${self.cost:.4f}')


def run_task(goal, budget_usd=0.50, max_steps=40):
    meter = TaskMeter(budget_usd)
    messages = [
        {'role': 'system', 'content': 'Work step by step. Reply DONE when finished.'},
        {'role': 'user', 'content': goal},
    ]
    for _ in range(max_steps):
        resp = client.chat.completions.create(
            model=MODEL, messages=messages, max_tokens=1024)
        meter.add(resp.usage)  # log first, then decide
        reply = resp.choices[0].message.content or ''
        messages.append({'role': 'assistant', 'content': reply})
        if 'DONE' in reply:
            break
        messages.append({'role': 'user', 'content': 'Continue.'})
    return meter


try:
    meter = run_task('Summarise the main risks of a database migration.')
    print(f'finished: {meter.steps} steps, ${meter.cost:.4f}')
except BudgetExceeded as err:
    print('stopped:', err)

そこにある3つの選択は、真似する価値がある。

決定する前にログを取る。 meter.addはまずステップを記録し、それから予算をチェックするので、停止した実行でも完全なトレースが残る。チェックはラインを超えた呼び出しの後に行われるため、最大で1ステップ分オーバーシュートする可能性がある。max_tokens=1024はその最後のステップの出力を制限する。

ハードストップは1つではなく2つ。 max_stepsはあまり使わずにループするエージェントを捕まえ、ドル予算は1ステップあたり多く使うエージェントを捕まえる。両方欲しい。

価格は1か所に置く。 PRICE_INとPRICE_OUTはプレースホルダーの数値である。あなたのモデルの現在の値を価格ページからコピーすること。現時点でDeepSeek-V4-Flashは100万入力トークンあたり$0.015、100万出力トークンあたり$0.10である。

40ステップの実行が合計いくらになるか#

エージェントが40ステップかかり、コンテキストが成長するため、平均的なステップが6,000入力トークンを送信し、400トークンを返すとしよう。それはタスクあたり40 × 6,000 = 240,000入力トークンと40 × 400 = 16,000出力トークンである。

DeepSeek-V4-Flashでは、これは$0.0052かかる。表は同じワークロードを他のモデルで実行し、OpenRouterの参考価格を当社のものと並べている。

合計で、入力と出力を合わせて数えると、このタスクではDeepSeek-V4-FlashがOpenRouterより安くなる。トークンがどこにあるかにも注目してほしい: 256,000のうち240,000が入力である。エージェントが引きずるコンテキストが請求額であり、だからこそログのステップごとのin=の数値が注目すべきものだ。上記の例のタスクあたり$0.50の予算は、暴走に対する上限であり、目標ではない。

モデルの切り替えはMODELを変更するだけだ。IDはCheapTokensドキュメントから選ぶこと: 名前はプロバイダー間で異なり、価格ラベルの短いIDはAPIでは機能しない。長いエージェントループでは、DeepSeek-V4-Flashは一日中いつでも同じ価格なので、タスクのコストは正午でも午前3時でも同じである。その制限についてはDeepSeekページを参照。

モデルが思考しているとき#

回答のトークン数が、目に見えるテキストよりずっと長い場合は、reasoning_tokensを見てほしい。思考トークンは出力としてカウントされるため、推論の多いステップは、目に見えるテキストが示唆するよりもはるかに多くの出力トークンを生成し、それらは出力として課金される。上記のメーターは、APIがフィールドを返すときはmeter.reasoningにそれらを収集し、返さないときは0のままにするので、キーが欠けてもクラッシュしない。出力のうちどれだけが思考だったかを見たい場合は、out=の隣にそれを出力すること。推論努力と思考トークンが理由を説明している。

ストリーミング: usageは最後に来る#

エージェントがストリーミングする場合、通常のチャンクにはusageがない。OpenAI形式ではstream_options={'include_usage': True}でそれを要求する。DeepSeekのリファレンス（2026年10月26日確認）によると、このオプションはstream=Trueと一緒にのみ機能し（そうでなければAPIは400を返す）、usageは1つを除くすべてのチャンクでnullであり、data: [DONE]の前の最後のチャンクがリクエスト全体の合計を運ぶ。したがってストリーミングループでは、usageが空でないチャンクで、meter.addを一度呼び出す。上記の非ストリーミングループが始めるのに最も簡単な場所であり、Pythonでのストリーミングにループ自体がある。

数字がおかしい場合#

• usageがNone。 include_usageなしでストリーミングしている。それを追加するか、非ストリーミング呼び出しに切り替える。
• 合計が残高の減少より少ない。 すべての呼び出しがメーターを通っているか確認する: リトライ、要約呼び出し、run_taskの外で行われるツール選択呼び出しは見落としやすい。client.chat.completions.createを1つの関数でラップし、すべてをそれ経由でルーティングする。
• ステップあたりのコストが上がり続ける。 これはmessagesにすべてを追加するループでは正常である。古いツール出力を削るか、履歴を要約する。コンテキスト外のツール出力がそのアイデアを示している。
• BudgetExceededがステップ2で発火する。 予算が1ステップより小さい。PRICE_INが100万トークンあたりであり、1Kあたりではないことを確認する。

数値を保存するには、meter.steps、meter.tokens_in、meter.tokens_out、meter.costをタスクごとに1つのJSON行として書き出す。1週間分のそれらの行は、あなたのエージェントの実際のステップあたりトークン数を教えてくれ、それがその後のすべての見積もりに入れるべき数値である。モデル、価格、最初のリクエストはドキュメントにあり、ツール呼び出しはPythonでの関数呼び出しで扱っている。