Kompaktierung: Komprimierung langer Gespräche

Das Problem mit langen Gesprächen

Jede hinzugefügte Nachricht in einem Gespräch erhöht die Anzahl der an die API gesendeten Tokens. Ein Gespräch mit 50 Runden kann leicht 30.000 Eingabetokens erreichen, was sowohl die Latenz als auch die Kosten in die Höhe treibt. Das Problem ist heimtückisch: Die erste Runde eines Support-Assistenten kostet nur wenige hundert Tokens, während die dreißigste Runde mehrere tausend kostet, obwohl der Nutzer die exakt gleiche Art von Frage stellt. Sie zahlen immer mehr für einen identischen Service, und der Nutzer wartet immer länger auf eine Antwort.

Die Kompaktierung löst dies, indem sie die Historie komprimiert, ohne wesentliche Informationen zu verlieren. Dies beinhaltet das Zusammenfassen älterer Austausche, um die Token-Anzahl zu reduzieren, während der notwendige Kontext erhalten bleibt. Anstatt die gesamte Historie zu senden, senden Sie eine Zusammenfassung vergangener Austausche sowie die jüngsten Nachrichten. Das Modell weiß dann immer noch, dass der Kunde Dupont heißt, dass eine umstrittene Bestellung vorliegt und dass Sie bereits eine Rückerstattung angeboten haben, ohne die vierzig Nachrichten neu lesen zu müssen, die diese Fakten etabliert haben.

Der Mechanismus

Die folgende Funktion wendet diese Logik in ihrer einfachsten Form an: Sie schätzt das Volumen, tut nichts, bis die Schwelle überschritten ist, und teilt dann die Historie in zwei Teile auf. Ältere Nachrichten gehen in die Zusammenfassung; die jüngsten bleiben unverändert.

import openai

client = openai.OpenAI()

def compact_history(
    messages: list[dict],
    token_threshold: int = 10_000,
    keep_recent: int = 6,
) -> list[dict]:
    """Kompaktiert die Historie, wenn sie die Schwelle überschreitet."""

    # Schätzung der Token-Anzahl (Approximation)
    estimated_tokens = sum(len(m["content"]) // 4 for m in messages)

    if estimated_tokens < token_threshold:
        return messages  # Keine Kompaktierung notwendig

    # Aufteilung: alte Nachrichten vs. jüngste
    old_messages = messages[:-keep_recent]
    recent_messages = messages[-keep_recent:]

    # Zusammenfassung der alten Nachrichten
    old_text = "\n".join(
        f"{m['role']}: {m['content']}" for m in old_messages
    )

    response = client.responses.create(
        model="gpt-5.6-terra",
        input=(
            "Fassen Sie dieses Gespräch kurz zusammen. "
            "Behalten Sie wichtige Fakten, getroffene Entscheidungen "
            "und den Kontext bei, der notwendig ist, um das Folgende zu verstehen.\n\n"
            f"{old_text}"
        ),
        max_output_tokens=500,
    )

    # Neuaufbau der kompaktierten Historie
    summary_message = {
        "role": "system",
        "content": f"Zusammenfassung des vorherigen Gesprächs:\n{response.output_text}",
    }

    return [summary_message] + recent_messages

Zwei Details verdienen Ihre Aufmerksamkeit. Der Zusammenfassungs-Prompt fordert explizit getroffene Entscheidungen an, da eine Zusammenfassung, die nur Themen behält, dazu führt, dass das Modell sich seiner eigenen Verpflichtungen nicht mehr erinnert. Und max_output_tokens=500 begrenzt die Zusammenfassung: ohne diese Begrenzung wächst die Zusammenfassung mit jeder Kompaktierung, und Sie landen wieder beim Problem, das Sie zu eliminieren versucht haben.

Zwei Strategien im Überblick

Gleitendes Fenster mit Zusammenfassung

Die folgende Klasse industrialisiert das Prinzip. Sie verwaltet eine einzige Zusammenfassung, die bei jeder Kompaktierung neu geschrieben wird, indem sie neue Austausche integriert, während sie ein Fenster mit rohen Nachrichten behält. Der an die API gesendete Kontext besteht daher immer aus drei Blöcken: dem System-Prompt, der kumulativen Zusammenfassung und dem jüngsten Fenster.

class CompactedConversation:
    """Verwaltet ein Gespräch mit automatischer Kompaktierung."""

    def __init__(
        self,
        system_prompt: str,
        compaction_threshold: int = 8_000,
        window_size: int = 8,
    ):
        self.system_prompt = system_prompt
        self.threshold = compaction_threshold
        self.window_size = window_size
        self.summary: str | None = None
        self.messages: list[dict] = []

    def build_context(self) -> list[dict]:
        """Erstellt den Kontext, der an die API gesendet wird."""
        context = [{"role": "system", "content": self.system_prompt}]

        if self.summary:
            context.append({
                "role": "system",
                "content": f"Vorheriger Kontext: {self.summary}",
            })

        context.extend(self.messages[-self.window_size:])
        return context

    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        self._check_compaction()

    def _check_compaction(self):
        tokens = sum(len(m["content"]) // 4 for m in self.messages)
        if tokens > self.threshold:
            self._compact()

    def _compact(self):
        old_messages = self.messages[:-self.window_size]
        if not old_messages:
            return

        text = "\n".join(
            f"{m['role']}: {m['content']}" for m in old_messages
        )

        existing_context = f"Bestehende Zusammenfassung: {self.summary}\n\n" if self.summary else ""

        response = client.responses.create(
            model="gpt-5.6-terra",
            input=(
                f"{existing_context}"
                f"Neue Austausche zur Integration in die Zusammenfassung:\n{text}\n\n"
                "Erstellen Sie eine aktualisierte, kurze und faktische Zusammenfassung."
            ),
            max_output_tokens=400,
        )

        self.summary = response.output_text
        self.messages = self.messages[-self.window_size:]

Beachten Sie, dass _compact die bestehende Zusammenfassung in den Prompt vor den neuen Austauschen wieder einfügt. Dies verhindert progressiven Gedächtnisverlust: ohne diese Wiedereinjection würde jede Kompaktierung alles löschen, was die vorherige behielt.

Selektive Kompaktierung

Zusammenfassung ist destruktiv, und manche Inhalte überleben die Neuformulierung nicht. Ein Code-Schnipsel, eine JSON-API-Antwort oder eine Tabelle mit Zahlen verliert seine Nützlichkeit, sobald sie paraphrasiert wird. Die selektive Kompaktierung sortiert daher alte Nachrichten vor der Zusammenfassung: Nachrichten mit Code oder strukturierten Daten gehen unverändert durch, während nur gesprochener Text komprimiert wird.

def selective_compaction(
    messages: list[dict],
    keep_recent: int = 6,
) -> list[dict]:
    """Kompaktiert, während strukturierte Nachrichten intakt bleiben."""
    old_messages = messages[:-keep_recent]
    recent_messages = messages[-keep_recent:]

    to_summarize = []
    to_keep = []

    for msg in old_messages:
        content = msg["content"]
        # Behalten Sie Nachrichten mit Code oder Daten
        if "```" in content or "{" in content:
            to_keep.append(msg)
        else:
            to_summarize.append(msg)

    # Zusammenfassung nur von gesprochenerm Text
    summary = ""
    if to_summarize:
        text = "\n".join(f"{m['role']}: {m['content']}" for m in to_summarize)
        response = client.responses.create(
            model="gpt-5.6-terra",
            input=f"Fassen Sie diese Austausche in 3 Sätzen zusammen:\n{text}",
            max_output_tokens=200,
        )
        summary = response.output_text

    result = []
    if summary:
        result.append({
            "role": "system",
            "content": f"Zusammenfassung der Austausche: {summary}",
        })
    result.extend(to_keep)
    result.extend(recent_messages)
    return result

Das hier verwendete Sortierungskriterium – das Vorhandensein von drei Backticks oder einer geschweiften Klammer – ist absichtlich grob. Passen Sie es an das an, was Ihre Tools tatsächlich produzieren: Für einen Assistenten, der sich mit Fallreferenzen befasst, möchten Sie vielleicht jede Nachricht erhalten, die eine Kunden-ID enthält.

Überprüfung, ob Kompaktierung sich lohnt

Schlecht kalibrierte Kompaktierung kostet einen zusätzlichen API-Aufruf ohne spürbaren Gewinn oder komprimiert so aggressiv, dass das Modell den Faden verliert und Fragen stellt, die bereits beantwortet wurden. Messen Sie daher das Verhältnis, bevor Sie Ihre endgültigen Schwellenwerte festlegen, unter Verwendung einer Stichprobe echter Gespräche.

def compaction_ratio(
    before: list[dict],
    after: list[dict],
) -> dict:
    tokens_before = sum(len(m["content"]) // 4 for m in before)
    tokens_after = sum(len(m["content"]) // 4 for m in after)
    reduction = 1 - (tokens_after / tokens_before) if tokens_before else 0

    return {
        "tokens_before": tokens_before,
        "tokens_after": tokens_after,
        "reduction": f"{reduction:.0%}",
        "messages_before": len(before),
        "messages_after": len(after),
    }

Nehmen Sie ein Dutzend Transkripte aus Ihrer Produktionsumgebung, führen Sie sie durch die Kompaktierung und lesen Sie die erzeugten Zusammenfassungen: Wenn Sie das Gespräch nicht allein auf Basis der Zusammenfassung fortsetzen können, ist Ihre Schwelle zu niedrig oder Ihr Zusammenfassungs-Prompt zu vage.

Kernaussagen

• Kompaktierung reduziert Tokens durch das Zusammenfassen älterer Nachrichten
• Behalten Sie immer die jüngsten Nachrichten intakt (gleitendes Fenster)
• Bewahren Sie Nachrichten mit Code oder strukturierten Daten auf
• Injectieren Sie die bestehende Zusammenfassung bei jeder neuen Kompaktierung, um progressiven Gedächtnisverlust zu vermeiden
• Messen Sie das Kompressionsverhältnis, um Ihre Schwellenwerte zu kalibrieren