Токены и контекстное окно:
как LLM видит текст

pip install tiktoken

import tiktoken

# Получаем энкодер для конкретной модели
enc = tiktoken.encoding_for_model("gpt-4o")
# Или по имени кодировки:
# enc = tiktoken.get_encoding("cl100k_base")  # GPT-4, GPT-3.5
# enc = tiktoken.get_encoding("o200k_base")   # GPT-4o, GPT-4o-mini

text = "AI-агент получает запрос и выполняет задачу"

# Кодируем текст → список токенов (чисел)
tokens = enc.encode(text)
print(f"Токены: {tokens}")
# [15836, 44, 18822, 61066, 25, 2099, 23485, 11, 7, 19897, 9847, 12, 2234, 44]

print(f"Количество токенов: {len(tokens)}")
# Количество токенов: 14

# Декодируем обратно (для отладки)
for token_id in tokens[:5]:
    fragment = enc.decode([token_id])
    print(f"  {token_id:6d} → {repr(fragment)}")

import tiktoken

def count_message_tokens(
    messages: list[dict],
    model: str = "gpt-4o",
) -> int:
    """
    Точный подсчёт токенов в списке сообщений чата.
    Учитывает overhead на разметку ролей (~4 токена на сообщение).
    Формула из официальной документации OpenAI.
    """
    enc = tiktoken.encoding_for_model(model)

    # Каждое сообщение: 3 токена overhead (role + content separators)
    tokens_per_message = 3
    # Каждый ответ начинается с: <|start|>assistant<|message|>
    tokens_per_reply_prefix = 3

    total = tokens_per_reply_prefix
    for message in messages:
        total += tokens_per_message
        for key, value in message.items():
            total += len(enc.encode(str(value)))

    return total

# Пример: история диалога агента
messages = [
    {"role": "system", "content": "Ты — полезный AI-агент. Отвечай кратко и по делу."},
    {"role": "user", "content": "Найди информацию о библиотеке LangGraph"},
    {"role": "assistant", "content": "LangGraph — библиотека для создания агентов на основе графов..."},
    {"role": "user", "content": "Как установить LangGraph?"},
]

total = count_message_tokens(messages)
print(f"Токенов в истории: {total}")
# Токенов в истории: ~85

# Проверка перед отправкой запроса
MAX_CONTEXT = 128_000  # gpt-4o
MAX_RESPONSE = 4_096

if total + MAX_RESPONSE > MAX_CONTEXT:
    print("⚠️ Контекст переполнен! Нужно укоротить историю")
else:
    available = MAX_CONTEXT - total - MAX_RESPONSE
    print(f"✅ Свободно для документов/инструментов: {available:,} токенов")

import tiktoken

def show_tokens(text: str, model: str = "gpt-4o") -> None:
    """Распечатывает текст с разбивкой на токены."""
    enc = tiktoken.encoding_for_model(model)
    token_ids = enc.encode(text)

    print(f"Текст ({len(text)} символов) → {len(token_ids)} токенов\n")
    for i, tid in enumerate(token_ids):
        fragment = enc.decode([tid])
        print(f"  [{i:3d}] id={tid:6d}  {repr(fragment)}")

show_tokens("Hello, world!")
# Текст (13 символов) → 4 токена
#   [  0] id= 9906  'Hello'
#   [  1] id=   11  ','
#   [  2] id=  995  ' world'
#   [  3] id=   0  '!'

show_tokens("Привет, мир!")
# Текст (12 символов) → 6 токенов
# (русский кодируется менее эффективно)

import tiktoken

def estimate_agent_context(
    system_prompt: str,
    tools: list[dict],
    history: list[dict],
    rag_docs: list[str],
    model: str = "gpt-4o",
    model_max_tokens: int = 128_000,
) -> dict:
    """Показывает, сколько токенов занимает каждая часть контекста агента."""
    import json
    enc = tiktoken.encoding_for_model(model)

    def count(text: str) -> int:
        return len(enc.encode(text))

    system_tokens  = count(system_prompt)
    tools_tokens   = count(json.dumps(tools))
    history_tokens = sum(count(m["content"]) + 4 for m in history)
    docs_tokens    = sum(count(doc) for doc in rag_docs)
    used           = system_tokens + tools_tokens + history_tokens + docs_tokens

    print(f"{'Компонент':<25} {'Токены':>8}  {'%':>5}")
    print("─" * 42)
    print(f"{'System prompt':<25} {system_tokens:>8,}  {system_tokens/model_max_tokens*100:>4.1f}%")
    print(f"{'Описания инструментов':<25} {tools_tokens:>8,}  {tools_tokens/model_max_tokens*100:>4.1f}%")
    print(f"{'История диалога':<25} {history_tokens:>8,}  {history_tokens/model_max_tokens*100:>4.1f}%")
    print(f"{'RAG-документы':<25} {docs_tokens:>8,}  {docs_tokens/model_max_tokens*100:>4.1f}%")
    print("─" * 42)
    print(f"{'ИТОГО занято':<25} {used:>8,}  {used/model_max_tokens*100:>4.1f}%")
    print(f"{'Свободно для ответа':<25} {model_max_tokens - used:>8,}  {(model_max_tokens-used)/model_max_tokens*100:>4.1f}%")

    return {
        "used": used, "free": model_max_tokens - used,
        "overflow": used > model_max_tokens,
    }

# Пример
result = estimate_agent_context(
    system_prompt="Ты — AI-агент для анализа документов...",
    tools=[{"name": "search", "description": "..."}],
    history=[
        {"role": "user", "content": "Проанализируй этот отчёт"},
        {"role": "assistant", "content": "Хорошо, начинаю анализ..."},
    ],
    rag_docs=["Текст документа 1...", "Текст документа 2..."],
)

import tiktoken

def trim_history(
    messages: list[dict],
    max_tokens: int = 8_000,
    model: str = "gpt-4o",
    keep_system: bool = True,
) -> list[dict]:
    """
    Обрезает историю сообщений до max_tokens.
    Всегда сохраняет системный промпт и последнее сообщение пользователя.
    """
    enc = tiktoken.encoding_for_model(model)

    def count(msgs: list[dict]) -> int:
        return sum(len(enc.encode(m["content"])) + 4 for m in msgs)

    # Отделяем system prompt
    system = [m for m in messages if m["role"] == "system"]
    dialog = [m for m in messages if m["role"] != "system"]

    # Всегда сохраняем последнее сообщение пользователя
    last = dialog[-1:] if dialog else []
    middle = dialog[:-1]

    # Убираем старые сообщения с начала истории пока не влезем
    while middle and count(system + middle + last) > max_tokens:
        middle.pop(0)  # Удаляем самое старое

    trimmed = system + middle + last
    original_count = count(messages)
    new_count = count(trimmed)

    if new_count < original_count:
        print(f"История обрезана: {original_count} → {new_count} токенов")

    return trimmed

from openai import AsyncOpenAI
import tiktoken

client = AsyncOpenAI()

async def compress_history(
    messages: list[dict],
    token_threshold: int = 6_000,
    model: str = "gpt-4o-mini",
) -> list[dict]:
    """
    Если история превышает порог — суммаризируем старые сообщения
    и заменяем их одним компактным сообщением-резюме.
    """
    enc = tiktoken.encoding_for_model(model)
    total = sum(len(enc.encode(m["content"])) + 4 for m in messages)

    if total <= token_threshold:
        return messages  # Ещё влезает — ничего не делаем

    # Берём первые 70% сообщений для суммаризации
    system_msgs = [m for m in messages if m["role"] == "system"]
    dialog      = [m for m in messages if m["role"] != "system"]
    split       = max(2, int(len(dialog) * 0.7))
    to_compress = dialog[:split]
    to_keep     = dialog[split:]

    # Суммаризируем через LLM
    history_text = "\n".join(
        f"{m['role'].upper()}: {m['content']}" for m in to_compress
    )
    summary_resp = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": (
                "Сделай краткое резюме следующего диалога (2-4 предложения). "
                "Сохрани ключевые факты, решения и контекст:\n\n" + history_text
            ),
        }],
        max_tokens=300,
    )
    summary = summary_resp.choices[0].message.content

    # Заменяем старую историю резюме
    summary_msg = {
        "role": "assistant",
        "content": f"[Резюме предыдущего разговора]: {summary}",
    }

    compressed = system_msgs + [summary_msg] + to_keep
    new_total = sum(len(enc.encode(m["content"])) + 4 for m in compressed)
    print(f"История сжата: {total} → {new_total} токенов")
    return compressed

from dataclasses import dataclass

@dataclass
class ModelPricing:
    input_per_1m: float   # $ за 1M input токенов
    output_per_1m: float  # $ за 1M output токенов
    cached_per_1m: float = 0.0  # $ за 1M кэшированных input

PRICING = {
    "gpt-4o":           ModelPricing(2.50,  10.00, 1.25),
    "gpt-4o-mini":      ModelPricing(0.15,   0.60, 0.075),
    "claude-sonnet-4":  ModelPricing(3.00,  15.00, 0.30),
    "claude-haiku-4":   ModelPricing(0.80,   4.00, 0.08),
    "gemini-2.0-flash": ModelPricing(0.10,   0.40),
}

def calc_cost(
    model: str,
    input_tokens: int,
    output_tokens: int,
    cached_tokens: int = 0,
) -> float:
    p = PRICING[model]
    cost = (
        (input_tokens - cached_tokens) / 1_000_000 * p.input_per_1m
        + cached_tokens               / 1_000_000 * p.cached_per_1m
        + output_tokens               / 1_000_000 * p.output_per_1m
    )
    return cost

# Пример: 1000 запросов агента в день
input_tok  = 2_000   # токенов на запрос (история + RAG + system)
output_tok = 500     # токенов на ответ

for model in PRICING:
    daily = calc_cost(model, input_tok, output_tok) * 1000
    print(f"{model:<22} {daily:>8.3f}$/день  {daily*30:>8.1f}$/мес")

import anthropic

client = anthropic.Anthropic()

# Большой статичный системный промпт — идеальный кандидат для кэширования
SYSTEM_PROMPT = """Ты — опытный AI-агент для анализа финансовых документов.
... (длинный промпт на несколько тысяч токенов) ...
"""

# Добавляем cache_control: {"type": "ephemeral"} к статичным блокам
message = client.messages.create(
    model="claude-haiku-4-5-20251001",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": SYSTEM_PROMPT,
            "cache_control": {"type": "ephemeral"},  # ← кэшируем!
        }
    ],
    messages=[{"role": "user", "content": "Проанализируй этот отчёт..."}],
)

# Смотрим использование кэша в ответе
usage = message.usage
print(f"Cache hit:   {usage.cache_read_input_tokens:,} токенов  (дёшево!)")
print(f"Cache write: {usage.cache_creation_input_tokens:,} токенов  (первый раз дороже)")
print(f"Обычные:     {usage.input_tokens:,} токенов")

import tiktoken

# ── Подсчёт токенов ───────────────────────────────────
enc = tiktoken.encoding_for_model("gpt-4o")   # o200k_base
# enc = tiktoken.encoding_for_model("gpt-3.5-turbo")  # cl100k_base

tokens = enc.encode("Текст для подсчёта")
count  = len(tokens)                  # Количество токенов
text   = enc.decode(tokens)           # Обратно в текст

# ── Оценка без библиотеки (грубо) ─────────────────────
# ~4 символа = 1 токен (английский)
# ~2 символа = 1 токен (русский)
# ~3/4 слова = 1 токен (английский)
# ~1/2 слова = 1 токен (русский)
# 1 страница A4 (~500 слов) ≈ 600–800 токенов

# ── Контекстные окна (2025) ───────────────────────────
# gpt-4o:                 128K input / 16K max output
# gpt-4o-mini:            128K input / 16K max output
# claude-sonnet-4:        200K input / 64K max output
# claude-haiku-4:         200K input / 16K max output
# gemini-2.0-flash:         1M input / 8K max output

# ── Правило «потеря в середине» ───────────────────────
# Важное → в начало (system prompt) или конец (последнее сообщение)
# Середина длинного контекста → модель хуже «помнит»

# ── Стоимость (примерно, $/1M токенов, 2025) ─────────
# gpt-4o:        $2.50 in / $10 out
# gpt-4o-mini:   $0.15 in / $0.60 out
# claude-sonnet: $3.00 in / $15 out / $0.30 cached
# claude-haiku:  $0.80 in / $4.00 out / $0.08 cached

# ── Overhead сообщений чата ───────────────────────────
# ~4 токена на каждое сообщение (разметка роли)
# ~3 токена на начало ответа ассистента