C.W.K.
Stream
Lesson 04 of 04 · published

Token Tracking & Translation

~12 min · billing, tracking, openai-translation

Level 0Spark
0 XP0/35 lessons0/10 achievements
0/140 XP to next level140 XP to go0% complete

Track every call

If you don't log token counts per call, you can't answer simple questions like "how much did this user cost us today?" or "is this feature blowing up our bill?" Build the tracker on day one — retrofitting it after the fact is painful.

What to record

For every call: model, prompt tokens, completion tokens, cached tokens, finish reason, timestamp, route key (which feature called this). Store in a small DB or append to JSONL — both work for the volumes most apps see.

OpenAI ↔ Gemini message translation

If you have legacy OpenAI-shaped code or an adapter that takes OpenAI-style messages, you'll write the OpenAI → Gemini conversion many times. Three transformations matter:

  1. role: "system" → top-level system_instruction field, not a message.
  2. role: "assistant"role: "model" in contents.
  3. role: "tool" with tool_call_id → user-turn functionResponse part with id.

Code

Token tracker — the small version·python
from dataclasses import dataclass, field
from collections import defaultdict

@dataclass
class TokenTracker:
    by_model: dict = field(default_factory=lambda: defaultdict(
        lambda: {'prompt': 0, 'completion': 0, 'cached': 0, 'calls': 0}))

    def record(self, model: str, usage):
        bucket = self.by_model[model]
        bucket['prompt']     += getattr(usage, 'prompt_token_count', 0) or 0
        bucket['completion'] += getattr(usage, 'candidates_token_count', 0) or 0
        bucket['cached']     += getattr(usage, 'cached_content_token_count', 0) or 0
        bucket['calls']      += 1

    def estimate_cost_usd(self) -> float:
        rates = {  # USD per 1M tokens, simplified
            'gemini-2.5-pro':        {'in': 1.25, 'out': 10.00, 'cached': 0.125},
            'gemini-2.5-flash':      {'in': 0.30, 'out':  2.50, 'cached': 0.03},
            'gemini-2.5-flash-lite': {'in': 0.10, 'out':  0.40, 'cached': 0.0},
        }
        total = 0.0
        for model, b in self.by_model.items():
            r = rates.get(model, rates['gemini-2.5-flash'])
            uncached = b['prompt'] - b['cached']
            total += (uncached / 1e6) * r['in']
            total += (b['cached']  / 1e6) * r['cached']
            total += (b['completion'] / 1e6) * r['out']
        return total

tracker = TokenTracker()
# After each Gemini call:
tracker.record('gemini-2.5-flash', response.usage_metadata)
print(f'Spent so far: ${tracker.estimate_cost_usd():.4f}')
OpenAI → Gemini message translation·python
def openai_to_gemini(messages):
    """Convert OpenAI-style messages list -> (Gemini contents, system_instruction)."""
    system = None
    contents = []
    for msg in messages:
        role = msg['role']
        if role == 'system':
            system = msg['content']
        elif role == 'user':
            contents.append({
                'role':  'user',
                'parts': [{'text': msg['content']}],
            })
        elif role == 'assistant':
            # OpenAI's 'assistant' becomes Gemini's 'model'
            contents.append({
                'role':  'model',
                'parts': [{'text': msg['content']}],
            })
        elif role == 'tool':
            # OpenAI's 'tool' becomes Gemini's user-turn functionResponse
            contents.append({
                'role':  'user',
                'parts': [{
                    'functionResponse': {
                        'name':     msg.get('name', ''),
                        'id':       msg.get('tool_call_id', ''),
                        'response': {'result': msg['content']},
                    }
                }],
            })
    return contents, system
Use translation in adapter·python
from google import genai
from google.genai import types

async def call_via_openai_shape(messages):
    contents, system = openai_to_gemini(messages)
    config = types.GenerateContentConfig(
        system_instruction=system,
    ) if system else None

    response = await client.aio.models.generate_content(
        model='gemini-2.5-flash',
        contents=contents,
        config=config,
    )
    return response.text

# Same caller code as OpenAI:
text = await call_via_openai_shape([
    {'role': 'system', 'content': 'You are helpful.'},
    {'role': 'user',   'content': 'Hello!'},
])

External links

Exercise

Wire the TokenTracker into your GeminiAdapter so every generate_stream call records usage. Run 50 mixed prompts. Print the per-model breakdown and total cost. Then write a small report: which model dominated cost, which dominated token volume, and one optimization you'd try first.

Progress

Progress is local-only — sign in to sync across devices.
Spotted a bug or have feedback on this page?Report an Issue

Comments 0

🔔 Reply notifications (sign in)
Sign inPlease sign in to comment.

No comments yet — be the first.