Compaction : compresser les conversations longues

Le problème des conversations longues

Chaque message ajouté à une conversation augmente le nombre de tokens envoyés à l’API. Une conversation de 50 échanges peut facilement atteindre 30 000 tokens en entrée, ce qui augmente la latence et le coût. La compaction résout ce problème en compressant l’historique sans perdre les informations essentielles.

Qu’est-ce que la compaction ?

La compaction consiste à résumer les anciens échanges pour réduire le nombre de tokens tout en préservant le contexte nécessaire. Au lieu d’envoyer l’intégralité de l’historique, vous envoyez un résumé des échanges passés plus les messages récents.

Le mécanisme

import openai

client = openai.OpenAI()

def compacter_historique(
    messages: list[dict],
    seuil_tokens: int = 10_000,
    garder_recents: int = 6,
) -> list[dict]:
    """Compacte l'historique quand il dépasse le seuil."""

    # Estimer le nombre de tokens (approximation)
    tokens_estimes = sum(len(m["content"]) // 4 for m in messages)

    if tokens_estimes < seuil_tokens:
        return messages  # Pas besoin de compacter

    # Séparer : messages anciens vs récents
    anciens = messages[:-garder_recents]
    recents = messages[-garder_recents:]

    # Résumer les messages anciens
    texte_anciens = "\n".join(
        f"{m[role]}: {m[content]}" for m in anciens
    )

    response = client.responses.create(
        model="gpt-5.3",
        input=(
            "Résumez cette conversation de manière concise. "
            "Gardez les informations factuelles clés, les décisions prises, "
            "et le contexte nécessaire pour comprendre la suite.\n\n"
            f"{texte_anciens}"
        ),
        max_output_tokens=500,
    )

    # Reconstruire l'historique compacté
    message_resume = {
        "role": "system",
        "content": f"Résumé de la conversation précédente :\n{response.output_text}",
    }

    return [message_resume] + recents

Stratégies de compaction

Stratégie 1 : fenêtre glissante avec résumé

class ConversationCompactee:
    """Gère une conversation avec compaction automatique."""

    def __init__(
        self,
        prompt_systeme: str,
        seuil_compaction: int = 8_000,
        taille_fenetre: int = 8,
    ):
        self.prompt_systeme = prompt_systeme
        self.seuil = seuil_compaction
        self.taille_fenetre = taille_fenetre
        self.resume: str | None = None
        self.messages: list[dict] = []

    def construire_contexte(self) -> list[dict]:
        """Construit le contexte à envoyer à l'API."""
        contexte = [{"role": "system", "content": self.prompt_systeme}]

        if self.resume:
            contexte.append({
                "role": "system",
                "content": f"Contexte précédent : {self.resume}",
            })

        contexte.extend(self.messages[-self.taille_fenetre:])
        return contexte

    def ajouter_message(self, role: str, contenu: str):
        self.messages.append({"role": role, "content": contenu})
        self._verifier_compaction()

    def _verifier_compaction(self):
        tokens = sum(len(m["content"]) // 4 for m in self.messages)
        if tokens > self.seuil:
            self._compacter()

    def _compacter(self):
        anciens = self.messages[:-self.taille_fenetre]
        if not anciens:
            return

        texte = "\n".join(
            f"{m[role]}: {m[content]}" for m in anciens
        )

        contexte_existant = f"Résumé existant : {self.resume}\n\n" if self.resume else ""

        response = client.responses.create(
            model="gpt-5.3",
            input=(
                f"{contexte_existant}"
                f"Nouveaux échanges à intégrer au résumé :\n{texte}\n\n"
                "Produisez un résumé mis à jour, concis et factuel."
            ),
            max_output_tokens=400,
        )

        self.resume = response.output_text
        self.messages = self.messages[-self.taille_fenetre:]

Stratégie 2 : compaction sélective

Gardez certains messages intacts (ceux qui contiennent des données structurées) :

def compaction_selective(
    messages: list[dict],
    garder_recents: int = 6,
) -> list[dict]:
    """Compacte en gardant les messages structurés intacts."""
    anciens = messages[:-garder_recents]
    recents = messages[-garder_recents:]

    a_resumer = []
    a_garder = []

    for msg in anciens:
        contenu = msg["content"]
        # Garder les messages contenant du code ou des données
        if "```" in contenu or "{" in contenu:
            a_garder.append(msg)
        else:
            a_resumer.append(msg)

    # Résumer uniquement le texte conversationnel
    resume = ""
    if a_resumer:
        texte = "\n".join(f"{m[role]}: {m[content]}" for m in a_resumer)
        response = client.responses.create(
            model="gpt-5.3",
            input=f"Résumez ces échanges en 3 phrases :\n{texte}",
            max_output_tokens=200,
        )
        resume = response.output_text

    resultat = []
    if resume:
        resultat.append({
            "role": "system",
            "content": f"Résumé des échanges : {resume}",
        })
    resultat.extend(a_garder)
    resultat.extend(recents)
    return resultat

Mesurer l’efficacité de la compaction

def rapport_compaction(
    avant: list[dict],
    apres: list[dict],
) -> dict:
    tokens_avant = sum(len(m["content"]) // 4 for m in avant)
    tokens_apres = sum(len(m["content"]) // 4 for m in apres)
    reduction = 1 - (tokens_apres / tokens_avant) if tokens_avant else 0

    return {
        "tokens_avant": tokens_avant,
        "tokens_apres": tokens_apres,
        "reduction": f"{reduction:.0%}",
        "messages_avant": len(avant),
        "messages_apres": len(apres),
    }

Points clés à retenir

• La compaction réduit les tokens en résumant les anciens messages
• Gardez toujours les derniers messages intacts (fenêtre glissante)
• Préservez les messages contenant du code ou des données structurées
• Mesurez le ratio de compression pour calibrer vos seuils