Analizator dokumentów z OpenAI: streszczenie i raport zużycia tokenów

To trzecia część cyklu i projekt, który łączy wcześniejsze elementy. Zbudujemy program, który przyjmuje plik tekstowy, zwraca jego streszczenie i pokazuje, ile to kosztowało. Jeśli nie znasz jeszcze podstaw, zacznij od części o tokenach, kontekście i kosztach oraz części o testowaniu z atrapami. Jeśli dopiero uczysz się programować z pomocą AI, przeczytaj też o ChatGPT w nauce programowania.

Co zbudujemy i jak będzie wyglądał projekt?

Program uruchamiasz tak:

python analyzer.py raport.txt

Na wyjściu dostajesz streszczenie i raport:

Wywołania API: 4
Tokeny wejściowe: 9120
Tokeny wyjściowe: 310
Razem: 9430
Szacowany koszt: $0.0016

Wartości w raporcie są przykładowe. Struktura projektu jest prosta:

Zależności instalujesz poleceniem pip install openai tiktoken pytest, a klucz ustawiasz w zmiennej OPENAI_API_KEY.

Jak podzielić długi dokument na fragmenty?

Dokument dłuższy niż okno kontekstu spowodowałby błąd przekroczenia długości. Dlatego dzielimy go na kawałki o maksymalnej liczbie tokenów:

import tiktoken

MODEL = "gpt-4o-mini"
MAX_CHUNK_TOKENS = 3000


def count_tokens(text: str, model: str = MODEL) -> int:
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))


def split_into_chunks(text: str, max_tokens: int = MAX_CHUNK_TOKENS, model: str = MODEL) -> list[str]:
    encoding = tiktoken.encoding_for_model(model)
    tokens = encoding.encode(text)
    return [
        encoding.decode(tokens[start:start + max_tokens])
        for start in range(0, len(tokens), max_tokens)
    ]

To rozwiązanie jest najprostsze: tnie tekst co max_tokens tokenów, więc granica może wypaść w środku zdania. W prawdziwym projekcie warto dzielić po akapitach i dopiero wtedy sprawdzać limit, ale do nauki mechanizmu wystarczy wersja powyżej. Margines (3000 zamiast pełnego okna) zostawia miejsce na prompt i odpowiedź.

Jak zliczać zużycie tokenów i kosztu?

Zamiast rozrzucać liczniki po kodzie, zbieramy je w jednej klasie. Każda odpowiedź API zawiera usage, który dodajemy do raportu:

from dataclasses import dataclass

PRICE_INPUT_PER_1M = 0.15   # USD za milion tokenów wejściowych (przykład, sprawdź cennik)
PRICE_OUTPUT_PER_1M = 0.60  # USD za milion tokenów wyjściowych (przykład, sprawdź cennik)


@dataclass
class UsageReport:
    calls: int = 0
    prompt_tokens: int = 0
    completion_tokens: int = 0

    def add(self, usage) -> None:
        self.calls += 1
        self.prompt_tokens += usage.prompt_tokens
        self.completion_tokens += usage.completion_tokens

    @property
    def total_tokens(self) -> int:
        return self.prompt_tokens + self.completion_tokens

    @property
    def cost_usd(self) -> float:
        return (
            self.prompt_tokens * PRICE_INPUT_PER_1M
            + self.completion_tokens * PRICE_OUTPUT_PER_1M
        ) / 1_000_000

    def format(self) -> str:
        return (
            f"Wywołania API: {self.calls}\n"
            f"Tokeny wejściowe: {self.prompt_tokens}\n"
            f"Tokeny wyjściowe: {self.completion_tokens}\n"
            f"Razem: {self.total_tokens}\n"
            f"Szacowany koszt: ${self.cost_usd:.4f}"
        )

Dzięki temu cała wiedza o kosztach jest w jednym miejscu, a stawki zmieniasz w dwóch stałych.

Jak wywołać model i obsłużyć błędy API?

Funkcja ask_model wysyła zapytanie, dopisuje zużycie do raportu i ponawia próbę po błędzie limitu lub przekroczeniu czasu. Odstęp rośnie (1, 2, 4 sekundy), żeby nie zasypywać API kolejnymi zapytaniami:

import time

from openai import APITimeoutError, RateLimitError


def ask_model(client, prompt: str, report: UsageReport, retries: int = 3, sleep=time.sleep) -> str:
    for attempt in range(retries + 1):
        try:
            response = client.chat.completions.create(
                model=MODEL,
                messages=[
                    {"role": "system", "content": "Streszczasz dokumenty po polsku, zwięźle i rzeczowo."},
                    {"role": "user", "content": prompt},
                ],
                timeout=30,
            )
        except (RateLimitError, APITimeoutError):
            if attempt == retries:
                raise
            sleep(2 ** attempt)
            continue
        report.add(response.usage)
        return response.choices[0].message.content

Prompt systemowy jest stały, a zmienia się tylko treść dla użytkownika. Dzięki temu łatwo go zmienić w jednym miejscu, np. na streszczanie po angielsku.

Jak streścić dokument dłuższy niż okno kontekstu?

Stosujemy podejście dwuetapowe: streszczamy każdy fragment osobno, a potem prosimy model o połączenie tych streszczeń w jedno. Krótki dokument, który mieści się w jednym fragmencie, streszczamy od razu:

def summarize_document(client, text: str, max_chunk_tokens: int = MAX_CHUNK_TOKENS) -> tuple[str, UsageReport]:
    report = UsageReport()
    chunks = split_into_chunks(text, max_chunk_tokens)
    if len(chunks) == 1:
        summary = ask_model(client, f"Streść poniższy tekst w 5 zdaniach:\n\n{chunks[0]}", report)
        return summary, report

    partial = [
        ask_model(client, f"Streść ten fragment dokumentu w 3 zdaniach:\n\n{chunk}", report)
        for chunk in chunks
    ]
    joined = "\n\n".join(partial)
    summary = ask_model(client, f"Połącz streszczenia fragmentów w jedno, 5 zdań:\n\n{joined}", report)
    return summary, report

Funkcja zwraca dwie rzeczy naraz: streszczenie i raport. Przy dokumencie z czterech fragmentów raport pokaże pięć wywołań: cztery na fragmenty i jedno na połączenie.

Jak uruchomić program z wiersza poleceń?

Klienta tworzymy w osobnej funkcji, dzięki czemu w testach łatwo go podmienić:

from openai import OpenAI


def make_client() -> OpenAI:
    return OpenAI()  # klucz z zmiennej środowiskowej OPENAI_API_KEY


def main(path: str) -> None:
    with open(path, encoding="utf-8") as file:
        text = file.read()
    summary, report = summarize_document(make_client(), text)
    print(summary)
    print()
    print(report.format())


if __name__ == "__main__":
    import sys
    main(sys.argv[1])

Jak przetestować analizator w pytest?

Testy nie wysyłają żadnych zapytań. Klienta zastępujemy atrapą, a odpowiedź budujemy z prostego obiektu (szczegóły techniki opisaliśmy w artykule o atrapach):

from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import pytest

import analyzer
from analyzer import UsageReport, split_into_chunks, summarize_document


def fake_response(text="Streszczenie.", prompt_tokens=100, completion_tokens=20):
    return SimpleNamespace(
        choices=[SimpleNamespace(message=SimpleNamespace(content=text))],
        usage=SimpleNamespace(prompt_tokens=prompt_tokens, completion_tokens=completion_tokens),
    )


def test_long_document_is_summarized_in_parts():
    client = MagicMock()
    client.chat.completions.create.return_value = fake_response()

    summary, report = summarize_document(client, "słowo " * 2000, max_chunk_tokens=500)

    assert report.calls > 2  # kilka fragmentów + podsumowanie końcowe
    assert summary == "Streszczenie."


def test_chunks_respect_token_limit():
    chunks = split_into_chunks("słowo " * 2000, max_tokens=500)
    assert len(chunks) > 1
    assert all(analyzer.count_tokens(chunk) <= 500 for chunk in chunks)


def test_cost_is_calculated_from_tokens():
    report = UsageReport(calls=1, prompt_tokens=1_000_000, completion_tokens=1_000_000)
    assert report.cost_usd == pytest.approx(analyzer.PRICE_INPUT_PER_1M + analyzer.PRICE_OUTPUT_PER_1M)


def test_main_uses_patched_client(tmp_path, capsys):
    path = tmp_path / "doc.txt"
    path.write_text("Krótki tekst.", encoding="utf-8")
    fake_client = MagicMock()
    fake_client.chat.completions.create.return_value = fake_response("Wynik")

    with patch("analyzer.make_client", return_value=fake_client):
        analyzer.main(str(path))

    output = capsys.readouterr().out
    assert "Wynik" in output
    assert "Wywołania API: 1" in output

Test z podziałem dokumentu sprawdza, że przy małym limicie fragmentów powstaje więcej wywołań niż dwa, a test z patch uruchamia cały program na atrapie klienta. Testy ponawiania po błędzie 429 (z side_effect) znajdziesz w poprzednim artykule.

Uruchom je poleceniem pytest. Test liczący tokeny używa prawdziwego tiktoken, który przy pierwszym uruchomieniu pobiera słownik kodowania z internetu, więc pierwszy raz uruchom go z dostępem do sieci.

Jak rozbudować projekt dalej?

• Dziel tekst po akapitach zamiast co N tokenów, żeby fragmenty były spójne.
• Dodaj parametr języka i długości streszczenia w wierszu poleceń (argparse).
• Zapisuj raporty z wielu uruchomień do pliku CSV i porównuj koszty dokumentów.
• Dla bardzo dużych zbiorów dokumentów wyszukuj tylko trafne fragmenty zamiast streszczać wszystko. Opisujemy to w artykule o bazach wektorowych i RAG w Pythonie.
• Poproś narzędzie AI o napisanie kolejnej funkcji, ale zweryfikuj ją własnymi testami. Więcej o takim sposobie pracy w artykule o vibe codingu.
• Dodaj walidację odpowiedzi modelu w Pydantic, jeśli chcesz dostawać wynik w ustalonym formacie.

Projekt pasuje do portfolio, bo pokazuje pracę z zewnętrznym API, obsługę błędów i testy. Jeśli chcesz, żeby ktoś doświadczony obejrzał Twój kod, przeczytaj, czym jest code review, albo zobacz mentoring w ścieżce Python.