Краткосрочная память: in-context, история сообщений, суммаризация

import anthropic

client = anthropic.Anthropic()

# Вызов 1
r1 = client.messages.create(
    model="claude-opus-4-6", max_tokens=64,
    messages=[{"role": "user", "content": "Меня зовут Иван."}]
)
print(r1.content[0].text)  # «Привет, Иван! Рад познакомиться.»

# Вызов 2 — модель ничего не помнит
r2 = client.messages.create(
    model="claude-opus-4-6", max_tokens=64,
    messages=[{"role": "user", "content": "Как меня зовут?"}]
)
print(r2.content[0].text)  # «Не знаю, вы не представились.»

from anthropic import Anthropic

client = Anthropic()

SYSTEM = "Ты — ассистент для написания кода на Python."

def run_chat():
    """Простой диалоговый цикл с накоплением истории."""
    messages: list[dict] = []

    while True:
        user_input = input("Вы: ").strip()
        if not user_input or user_input.lower() in ("выход", "exit", "quit"):
            break

        # Добавляем сообщение пользователя в историю
        messages.append({"role": "user", "content": user_input})

        # Отправляем ВСЕЮ историю в API
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=1024,
            system=SYSTEM,
            messages=messages          # ← вся история
        )

        assistant_text = response.content[0].text
        print(f"Агент: {assistant_text}\n")

        # Сохраняем ответ ассистента — он войдёт в следующий запрос
        messages.append({"role": "assistant", "content": assistant_text})


if __name__ == "__main__":
    run_chat()

def count_tokens(messages: list[dict], system: str = "") -> int:
    """Считает input-токены для списка сообщений."""
    response = client.messages.count_tokens(
        model="claude-opus-4-6",
        system=system,
        messages=messages
    )
    return response.input_tokens


# Пример использования
messages = [
    {"role": "user", "content": "Привет!"},
    {"role": "assistant", "content": "Привет! Как могу помочь?"},
    {"role": "user", "content": "Расскажи про Python."},
]

tokens = count_tokens(messages, system="Ты — ассистент.")
print(f"История занимает {tokens} токенов")
# → История занимает 47 токенов

MAX_CONTEXT = 8_000   # ограничение для примера (модели поддерживают до 200 000)
SYSTEM = "Ты — ассистент для написания кода на Python."

def chat_with_metrics():
    """Диалог с отображением роста истории."""
    messages: list[dict] = []

    while True:
        user_input = input("Вы: ").strip()
        if not user_input:
            break

        messages.append({"role": "user", "content": user_input})

        # Считаем токены ПЕРЕД отправкой
        current_tokens = count_tokens(messages, system=SYSTEM)
        fill_pct = current_tokens / MAX_CONTEXT * 100
        print(f"  [tokens: {current_tokens}/{MAX_CONTEXT} = {fill_pct:.0f}%]")

        if current_tokens > MAX_CONTEXT * 0.9:
            print("  [⚠ контекст заполнен на 90%+ — нужно управление памятью]")

        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=1024,
            system=SYSTEM,
            messages=messages
        )
        text = response.content[0].text
        print(f"Агент: {text}\n")
        messages.append({"role": "assistant", "content": text})

def sliding_window(
    messages: list[dict],
    max_messages: int = 20
) -> list[dict]:
    """
    Оставляет последние max_messages сообщений.
    Гарантирует, что первое сообщение — user (инвариант API).
    """
    if len(messages) <= max_messages:
        return messages

    trimmed = messages[-max_messages:]

    # Если после обрезки первым оказался assistant — убираем его
    if trimmed and trimmed[0]["role"] == "assistant":
        trimmed = trimmed[1:]

    return trimmed


# Применение в цикле
messages = sliding_window(messages, max_messages=20)
response = client.messages.create(
    model="claude-opus-4-6",
    max_tokens=1024,
    system=SYSTEM,
    messages=messages
)

def trim_to_tokens(
    messages: list[dict],
    max_tokens: int = 50_000,
    system: str = ""
) -> list[dict]:
    """
    Удаляет старые сообщения, пока объём не войдёт в max_tokens.
    Удаляет парами (user + assistant), чтобы не нарушить инвариант чередования.
    Останавливается при <= 2 сообщениях (минимально нужное).
    """
    messages = list(messages)  # не мутируем оригинал

    while len(messages) > 2:
        current = count_tokens(messages, system=system)
        if current <= max_tokens:
            break

        # Удаляем первую пару user + assistant
        if len(messages) >= 2 and messages[0]["role"] == "user":
            messages = messages[2:]
        else:
            messages = messages[1:]

        # Убеждаемся, что после удаления первым идёт user
        if messages and messages[0]["role"] == "assistant":
            messages = messages[1:]

    return messages

SUMMARIZE_PROMPT = """Сожми переписку в краткое резюме (3-5 предложений).

Обязательно сохрани:
- Ключевые факты и цифры
- Принятые решения
- Имена, названия, технологии
- Незакрытые вопросы и задачи

Пиши в прошедшем времени, сжато. Не добавляй мнений."""


def summarize(messages_to_compress: list[dict]) -> str:
    """
    Суммаризирует список сообщений в один текстовый блок.
    Использует быструю и дешёвую модель.
    """
    # Форматируем переписку как текст
    conversation = "\n".join(
        f"{m['role'].upper()}: {m['content']}"
        for m in messages_to_compress
    )

    response = client.messages.create(
        model="claude-haiku-4-5-20251001",   # дешевле для вспомогательных задач
        max_tokens=400,
        messages=[{
            "role": "user",
            "content": f"{SUMMARIZE_PROMPT}\n\n---\n{conversation}"
        }]
    )
    return response.content[0].text.strip()


def maybe_summarize(
    messages: list[dict],
    system: str,
    token_threshold: int = 6_000,
    keep_recent: int = 4
) -> tuple[list[dict], str]:
    """
    Проверяет объём истории и при превышении порога суммаризирует старые сообщения.

    Возвращает:
        messages — обновлённая история (только «живые» сообщения)
        summary  — текущее резюме (пустая строка если суммаризации не было)
    """
    if count_tokens(messages, system=system) <= token_threshold:
        return messages, ""

    # Делим историю: что сжимаем и что оставляем нетронутым
    to_compress = messages[:-keep_recent]
    recent = messages[-keep_recent:]

    if not to_compress:
        return messages, ""

    summary_text = summarize(to_compress)
    return recent, summary_text

from dataclasses import dataclass, field
import anthropic

client = anthropic.Anthropic()
SYSTEM = "Ты — полезный ассистент."


@dataclass
class ConversationMemory:
    """
    Краткосрочная память агента с автоматической суммаризацией.

    messages      — «живые» сообщения (недавние, полный текст)
    summary       — сжатый контекст всего предыдущего диалога
    token_threshold — при превышении этого порога запускается суммаризация
    keep_recent   — сколько последних сообщений оставлять нетронутыми
    """
    messages: list[dict] = field(default_factory=list)
    summary: str = ""
    token_threshold: int = 6_000
    keep_recent: int = 6

    def add(self, role: str, content: str) -> None:
        """Добавляет сообщение и при необходимости суммаризирует историю."""
        self.messages.append({"role": role, "content": content})
        self._maybe_compress()

    def get_messages(self) -> list[dict]:
        """
        Возвращает список сообщений для отправки в API.
        Если есть резюме — вставляет его как первый блок (user + ack).
        """
        if not self.summary:
            return list(self.messages)

        # Резюме оборачиваем в user/assistant пару,
        # чтобы не нарушать инвариант чередования
        return [
            {
                "role": "user",
                "content": f"[Резюме предыдущего диалога]\n{self.summary}"
            },
            {
                "role": "assistant",
                "content": "Понял, учту контекст."
            },
            *self.messages
        ]

    def _maybe_compress(self) -> None:
        """Суммаризирует старые сообщения при превышении порога токенов."""
        context = self.get_messages()
        if count_tokens(context, system=SYSTEM) <= self.token_threshold:
            return

        # Делим историю: что сжимаем и что оставляем
        to_compress = self.messages[:-self.keep_recent]
        recent = self.messages[-self.keep_recent:]

        if not to_compress:
            return  # нечего сжимать — слишком мало сообщений

        new_summary = summarize(to_compress)

        # Мёрджим с предыдущим резюме, если оно уже есть
        if self.summary:
            merged_input = (
                f"Предыдущее резюме:\n{self.summary}\n\n"
                f"Новые события для добавления:\n{new_summary}"
            )
            self.summary = summarize([
                {"role": "user", "content": merged_input}
            ])
        else:
            self.summary = new_summary

        self.messages = list(recent)
        print(f"  [memory] суммаризация: {len(to_compress)} сообщ. → резюме")

def run_agent_with_memory():
    """Агент с управляемой краткосрочной памятью."""
    memory = ConversationMemory(
        token_threshold=6_000,  # суммаризировать при ~6к токенов
        keep_recent=6           # последние 3 хода всегда «живые»
    )

    while True:
        user_input = input("Вы: ").strip()
        if not user_input:
            break

        # Добавляем сообщение пользователя
        memory.add("user", user_input)

        # Получаем контекст для API (с резюме если есть)
        api_messages = memory.get_messages()

        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=1024,
            system=SYSTEM,
            messages=api_messages
        )

        assistant_text = response.content[0].text
        print(f"Агент: {assistant_text}\n")

        # Сохраняем ответ ассистента
        memory.add("assistant", assistant_text)


if __name__ == "__main__":
    run_agent_with_memory()