토큰 추적과 메시지 변환

모든 호출을 추적해

호출마다 토큰 수를 기록하지 않으면 "오늘 이 사용자가 비용을 얼마나 썼지?"나 "이 기능이 청구서를 폭발시키나?" 같은 단순한 질문에도 답할 수 없어. 첫날부터 추적기를 만들어. 나중에 모든 경로에 덧대려면 고통스러워.

기록할 값

모든 호출에서 모델, 프롬프트 토큰, 완성 토큰, 캐시된 토큰, 종료 이유, 시각, 어느 기능이 호출했는지를 나타내는 경로 키를 기록해. 작은 DB나 JSONL에 덧붙이면 대부분의 앱 규모에는 충분해.

OpenAI와 Gemini 사이의 메시지 변환

예전 OpenAI 구조의 코드나 OpenAI 형식 메시지를 받는 어댑터가 있다면 OpenAI → Gemini 변환을 자주 쓰게 돼. 세 가지가 중요해:

role: "system"은 메시지가 아니라 최상위 system_instruction 필드로 옮겨.
role: "assistant"는 contents 안의 role: "model"로 바꿔.
tool_call_id가 있는 role: "tool"은 id를 가진 사용자 차례의 functionResponse part로 바꿔.

Code

토큰 추적기 — 작은 버전·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 메시지 변환·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

어댑터에서 변환 사용하기·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!'},
])

Exercise

TokenTracker를 GeminiAdapter에 연결해 모든 generate_stream 호출이 사용량을 기록하게 해. 서로 다른 프롬프트 50개를 실행하고 모델별 내역과 총비용을 출력해. 그다음 어느 모델이 비용을 가장 많이 차지했는지, 어느 모델이 토큰 양을 가장 많이 차지했는지, 처음 시도할 최적화 한 가지가 무엇인지 짧게 보고해.