Dein KI-Agent vergisst alles nach 50 Nachrichten. So behebst du das.

Fuehre einen Pydantic AI Agenten fuer 50 Nachrichten aus und frage ihn nach etwas aus Nachricht #3. Er wird sich nicht erinnern. Nicht weil das Modell schlecht ist - sondern weil die Konversation zu lang wurde, das Kontextfenster voll lief und die fruehen Nachrichten verloren gingen.

Das ist kein Modellproblem. Es ist ein Infrastrukturproblem. Und es wird schlimmer bei Agenten, die Tools verwenden, weil jeder Tool-Aufruf und jede Antwort eine Nachricht ist. Eine 20-Runden-Konversation mit 3 Tool-Aufrufen pro Runde sind tatsaechlich 80+ Nachrichten. Bei ~4 Zeichen pro Token sind das Zehntausende von Tokens, die verbraucht werden, bevor man es merkt.

TL;DR

• Kontextueberlauf ist ein Infrastrukturproblem, kein Modellproblem. Jeder langlaufende Agent braucht eine Strategie zur Kontextverwaltung.
• SummarizationProcessor nutzt ein LLM, um alte Nachrichten intelligent zu einer Zusammenfassung zu komprimieren - hoehere Qualitaet, kostet aber einen API-Aufruf pro Trigger.
• SlidingWindowProcessor schneidet alte Nachrichten einfach ab, ohne Kosten - schnell, aber der Kontext geht vollstaendig verloren.
• Die Erhaltung von Tool-Call-Paaren ist kritisch. Beide Prozessoren stellen sicher, dass Tool-Aufrufe und ihre Antworten nie getrennt werden.
• Fraktionsbasierte Trigger (("fraction", 0.8)) sind die portabelste Option - sie passen sich automatisch an das Kontextfenster jedes Modells an.

Wir sind bei jedem langlaufenden Agenten-Deployment auf diese Wand gestossen. Kundensupport-Bots, die den Kundennamen vergessen. Code-Assistenten, die den Ueberblick verlieren, welche Dateien sie bereits geaendert haben. Research-Agenten, die Themen erneut untersuchen, die sie vor 30 Nachrichten behandelt haben.

Deshalb haben wir summarization-pydantic-ai gebaut - zwei Prozessoren, die die Konversationshistorie deines Agenten verwalten: einer, der intelligent mit einem LLM zusammenfasst, und einer, der alte Nachrichten einfach ohne Kosten abschneidet.

Zwei Prozessoren, zwei Kompromisse

Beide funktionieren als Pydantic AI History Processors - Drop-in-Funktionen, die die Nachrichtenhistorie vor jedem Agentenlauf transformieren.

SummarizationProcessor: Intelligente Kompression

from pydantic_ai import Agent
from pydantic_ai_summarization import create_summarization_processor

processor = create_summarization_processor(
    model="openai:gpt-4.1",
    trigger=("tokens", 100000),  # Trigger at 100k tokens
    keep=("messages", 20),       # Keep last 20 messages
)

agent = Agent(
    "openai:gpt-4.1",
    history_processors=[processor],
)

Wenn die Tokenanzahl den Trigger ueberschreitet, macht der Prozessor:

1. Berechnet einen sicheren Abschneidepunkt (trennt nie Tool-Call-Paare)
2. Sendet die aelteren Nachrichten an ein LLM zur Zusammenfassung
3. Ersetzt die alten Nachrichten durch eine kompakte Zusammenfassung
4. Behaelt die letzten 20 Nachrichten intakt

Das Ergebnis: Dein Agent behaelt den Kontext vom Anfang der Konversation, ohne das gesamte Token-Budget zu verbrauchen.

Drei Trigger-Typen

Du kannst die Zusammenfassung basierend auf Nachrichten, Tokens oder Kontextanteil ausloesen:

Nachrichtenbasiert - einfach und vorhersagbar:

processor = SummarizationProcessor(
    model="openai:gpt-4.1",
    trigger=("messages", 50),  # After 50 messages
)

Tokenbasiert - beruecksichtigt Nachrichtenlaenge:

processor = SummarizationProcessor(
    model="openai:gpt-4.1",
    trigger=("tokens", 100000),  # After 100k tokens
)

Fraktionsbasiert - passt sich dem Kontext jedes Modells an:

processor = SummarizationProcessor(
    model="openai:gpt-4.1",
    trigger=("fraction", 0.8),    # At 80% of context
    max_input_tokens=128000,       # GPT-4o's context window
)

Mehrere Trigger - ODER-Logik, erster gewinnt:

processor = SummarizationProcessor(
    model="openai:gpt-4.1",
    trigger=[
        ("messages", 50),     # OR
        ("tokens", 100000),   # OR
        ("fraction", 0.8),
    ],
    max_input_tokens=128000,
)

Fraktionsbasierte Trigger sind die robustesten fuer die Produktion - sie passen sich automatisch an, wenn du zwischen Modellen mit unterschiedlichen Kontextfenstern wechselst.

Tool-Call-Paar-Erhaltung

Das ist das Detail, das die meisten Kontextverwaltungsloesungen falsch machen. Betrachte diese Nachrichtensequenz:

User: "Search for Python tutorials"
Assistant: [tool_call: search("Python tutorials"), id=call_1]
Tool: [tool_return: "Found 5 results...", id=call_1]
Assistant: "Here are the top results..."
User: "Tell me more about the first one"

Wenn du zwischen dem Tool-Aufruf und seiner Rueckgabe schneidest, sieht das Modell einen verwaisten Tool-Aufruf - und bricht. Unsere Prozessoren behandeln das:

def _is_safe_cutoff_point(self, messages, cutoff_index):
    """Check if cutting at index would separate tool call/response pairs."""
    search_start = max(0, cutoff_index - 5)
    search_end = min(len(messages), cutoff_index + 5)

    for i in range(search_start, search_end):
        msg = messages[i]
        if not isinstance(msg, ModelResponse):
            continue

        tool_call_ids = set()
        for part in msg.parts:
            if isinstance(part, ToolCallPart) and part.tool_call_id:
                tool_call_ids.add(part.tool_call_id)

        if not tool_call_ids:
            continue

        # Check if cutoff separates this tool call from its response
        for j in range(i + 1, len(messages)):
            check_msg = messages[j]
            if isinstance(check_msg, ModelRequest):
                for part in check_msg.parts:
                    if (isinstance(part, ToolReturnPart)
                        and part.tool_call_id in tool_call_ids):
                        tool_before = i < cutoff_index
                        response_before = j < cutoff_index
                        if tool_before != response_before:
                            return False  # Unsafe - would split pair
    return True

Der Prozessor durchsucht 5 Nachrichten in jede Richtung vom Abschneidepunkt, findet alle Tool-Call/Return-Paare und stellt sicher, dass sie auf derselben Seite des Schnitts bleiben.

SlidingWindowProcessor: Kostenloses Trimmen

Wenn du keine intelligente Zusammenfassung brauchst - wenn Geschwindigkeit und Kosten wichtiger sind als Kontextqualitaet:

from pydantic_ai_summarization import create_sliding_window_processor

processor = create_sliding_window_processor(
    trigger=("messages", 100),  # Trim at 100 messages
    keep=("messages", 50),      # Keep last 50
)

agent = Agent(
    "openai:gpt-4.1",
    history_processors=[processor],
)

Kein LLM-Aufruf. Keine Latenz. Alte Nachrichten werden einfach verworfen. Die gleiche Tool-Call-Paar-Erhaltung gilt - das Sliding Window schneidet nicht mitten in einer Tool-Call-Sequenz.

Token-Zaehlung: Standardmaessig approximiert

Der Standard-Token-Zaehler verwendet eine einfache Heuristik - ~4 Zeichen pro Token:

def count_tokens_approximately(messages):
    total_chars = 0
    for msg in messages:
        if isinstance(msg, ModelRequest):
            for part in msg.parts:
                if isinstance(part, UserPromptPart):
                    if isinstance(part.content, str):
                        total_chars += len(part.content)
                elif isinstance(part, SystemPromptPart):
                    total_chars += len(part.content)
                elif isinstance(part, ToolReturnPart):
                    total_chars += len(str(part.content))
        elif isinstance(msg, ModelResponse):
            for part in msg.parts:
                if isinstance(part, TextPart):
                    total_chars += len(part.content)
                elif isinstance(part, ToolCallPart):
                    total_chars += len(part.tool_name)
                    total_chars += len(str(part.args))
    return total_chars // 4

Das ist schnell und fuer die meisten Faelle gut genug. Fuer Praezision verwende tiktoken:

import tiktoken

def accurate_counter(messages):
    encoding = tiktoken.encoding_for_model("gpt-4")
    total = 0
    for msg in messages:
        total += len(encoding.encode(str(msg)))
    return total

processor = create_summarization_processor(
    token_counter=accurate_counter,
    trigger=("tokens", 100000),
)

Benutzerdefinierte Zusammenfassungs-Prompts

Der Standard-Summary-Prompt extrahiert wichtigen Kontext. Du kannst ihn anpassen:

processor = create_summarization_processor(
    summary_prompt="""
    You are summarizing an agent conversation. Extract:

    1. **Key Decisions**: What was decided?
    2. **Code Changes**: What code was written/modified?
    3. **Pending Tasks**: What still needs to be done?
    4. **Important Context**: What context is crucial to preserve?

    Conversation to summarize:
    {messages}

    Provide a concise summary that preserves essential information.
    """,
)

Der {messages}-Platzhalter wird durch die formatierte Nachrichtenhistorie ersetzt. Passe dies an deine Domaene an - ein Kundensupport-Agent koennte Kundennamen und Problem extrahieren, waehrend ein Code-Assistent sich auf Dateiaenderungen und Testergebnisse konzentrieren koennte.

Konversationsschleifen-Muster

Hier ist das typische Produktionsmuster:

from pydantic_ai import Agent
from pydantic_ai_summarization import create_summarization_processor

processor = create_summarization_processor(
    trigger=("messages", 20),
    keep=("messages", 5),
)

agent = Agent(
    "openai:gpt-4.1",
    history_processors=[processor],
)

async def chat():
    message_history = []

    while True:
        user_input = input("You: ")
        if user_input.lower() == "quit":
            break

        result = await agent.run(
            user_input,
            message_history=message_history,
        )

        print(f"Assistant: {result.output}")
        message_history = result.all_messages()

Der History Processor wird automatisch vor jedem agent.run()-Aufruf ausgefuehrt. Du uebergibst die vollstaendige Nachrichtenhistorie, der Prozessor prueft, ob eine Zusammenfassung noetig ist, und gibt eine (moeglicherweise komprimierte) Historie zurueck. Kein manuelles Token-Zaehlen erforderlich.

Wichtigste Erkenntnisse

• Kontextueberlauf ist ein Infrastrukturproblem, kein Modellproblem. Jeder langlaufende Agent braucht eine Strategie zur Kontextverwaltung. Waehle zwischen intelligenter Zusammenfassung (hoehere Qualitaet, LLM-Kosten) und Sliding Window (null Kosten, Kontextverlust).
• Tool-Call-Paar-Erhaltung ist kritisch. Schneide nie zwischen einem Tool-Aufruf und seiner Antwort. Beide Prozessoren behandeln das automatisch - durchsuchen 5 Nachrichten in jede Richtung nach verwaisten Paaren.
• Fraktionsbasierte Trigger sind am portabelsten. ("fraction", 0.8) funktioniert unabhaengig davon, welches Modell du verwendest oder wie lang deine Nachrichten sind. Token- und Nachrichtenzahlen sind modell- und inhaltsspezifisch.
• Die ~4-Zeichen/Token-Heuristik ist gut genug. Fuer Produktionspraezision verwende tiktoken. Fuer alles andere haelt der Standard-Zaehler die Dinge schnell ohne zusaetzliche Abhaengigkeit.
• Passe den Zusammenfassungs-Prompt an deine Domaene an. Der Standard extrahiert generischen Kontext. Ein domaenenspezifischer Prompt (“extrahiere Kundennamen, Ticket-ID und Loesungsstatus”) produziert deutlich bessere Zusammenfassungen.

Probiere es selbst aus

summarization-pydantic-ai - Automatische Konversationszusammenfassung und Kontextverwaltung fuer Pydantic AI Agenten.

pip install summarization-pydantic-ai