Практичні поради: LangGraph Memory: Повний практичний посібник з керування

Наведені нижче примітки описують практичний підхід до роботи з книгою «LangGraph Memory: The Complete Practical Guide to Managing What Your Agent Remembers». Основна увага приділяється контрактам, перевіркам та шаблонам коду, а не мотиваційним аспектам. Під час вивчення огляду спочатку запишіть контракт: необхідні вхідні дані, сигнал про успіх та наслідки часткової невдачі. Такий перелік допоможе зберегти чесність пізніших змін у коді. Віддавайте перевагу невеликим, тестованим одиницям коду замість об’ємних скриптів. Якщо якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на складну ієрархію операцій.

Проблема пам’яті, про яку ніхто не говорить у навчальних матеріалах

Проблема пам’яті, про яку ніхто не говорить у посібниках, найкраще розв’язується, якщо її розглядати як вимірювану поверхню. Запишіть один ідеальний приклад роботи, один випадок збою та примітки щодо скасування змін перед розширенням обсягу роботи. Розглядайте цей етап як угоду між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте мовчазного часткового виконання завдань. Зберігайте стан графа у простому та типованому вигляді. Вкладені структури приховують інформацію про те, який вузол заповнив яке поле, що ускладнює продовження роботи після перерв.

Ландшафт пам’яті: чотири речі, які потрібно зрозуміти спочатку

«Ландшафт пам’яті: чотири речі, які потрібно зрозуміти спочатку» найкраще функціонує, якщо розглядати його як вимірювану поверхню. Запишіть один ідеальний приклад роботи, один випадок збою та примітки щодо скасування змін перед розширенням обсягу роботи. Записуйте час виконання та витрати на токени чи запити поруч із функціональними результатами. Візуалізація витрат заздалегідь запобігає несподіваним рахункам під час переходу від демо-середовища до спільних. Тримайте стан графа простим та типованим. Вкладені структури приховують інформацію про те, який вузол заповнив яке поле, що призводить до порушення послідовності виконання після перерв.

User sends message
        ↓
Graph loads State from Checkpointer (using thread_id)
        ↓
Memory Management runs (trim / filter / summarize)
        ↓
LLM receives cleaned messages
        ↓
LLM response added to State
        ↓
Checkpointer saves updated State (to memory / SQLite / Postgres)

Розширена версія модуля 2: Стан, обізнаний про пам’ять

Модуль 2 Розширений: Стан, що враховує пам’ять, працює найкраще, коли його розглядають як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу. Тримайте конфігурацію поза кодом додатку. Файли середовища, сховища секретів та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь граф. Тримайте стан графа простим та типованим. Вкладені блоки приховують інформацію про те, який вузол записав яке поле, і ускладнюють продовження роботи після перерв. Модуль 2 Розширений: Стан, що враховує пам’ять, працює найкраще, коли його розглядають як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу. Віддавайте перевагу малим, тестованим одиницям перед величезними скриптами. Коли якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану послідовність дій.

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

Стандартний стан, що враховує пам’ять

Для стандартного режиму, що враховує обсяг пам’яті, необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Розглядайте цю стадію як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте мовчазного часткового завершення. Запровадьте людське схвалення для тих кроків, які призводять до витрат грошей чи змінюють дані у продакшені. Підключення елементів під час компіляції не є гарантією повності виконання завдань у бізнес-сенсі.

# ── MODULE 2: STATE (Memory-Aware Version) ──────────────────
from langgraph.graph import MessagesState
# Option A: The shorthand (recommended for most agents)
# MessagesState is a pre-built TypedDict with just the messages field.
# Use it when you don't need custom fields.
class State(MessagesState):
    summary: str   # The only addition needed for memory management

Три стратегії керування довжиною розмови

У розділі «Три стратегії керування тривалістю розмови» необхідно визначити вхідні дані, відповідальну особу за кожен крок та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись визначити прихований стан. Записуйте час виконання та витрати на токени чи запити поруч із функціональними результатами. Відображення витрат заздалегідь запобігає несподіваним рахункам під час переходу з демо-середовища у спільні середовища. Встановлюйте людське схвалення для тих кроків, які спричиняють витрати чи змінюють дані у продакшені. Підключення під час компіляції не є гарантією повності бізнес-функціоналу.

Стратегія 1: Фільтрація повідомлень

Для стратегії 1: фільтрація повідомлень, перед зміною коду необхідно визначити вхідні дані, власника кроку та критерії завершення. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Зберігайте конфігурацію поза кодом додатку. Файли середовища, сховища конфіденційних даних та флаги функцій мають знаходитися в одному місці, яке оператори можуть перевірити, не читаючи весь алгоритм. Встановіть людське схвалення для кроків, які спричиняють витрати грошей або змінюють дані в продакшені. Підключення на етапі компіляції не є гарантією повноти бізнес-функцій. Для стратегії 1: фільтрація повідомлень, перед зміною коду необхідно визначити вхідні дані, власника кроку та критерії завершення. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Віддавайте перевагу невеликим, тестованим одиницям перед об’ємними скриптами. Коли крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану систему обробки даних.

Шаблон А: Фільтрувати лише те, що бачить LLM (стан незмінний)

При роботі за шаблоном А: Фільтрувати лише те, що бачить LLM (стан незмінний), спочатку запишіть угоду: необхідні вхідні дані, сигнал про успіх та те, що відбувається при частковій невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Розглядайте цей етап як угоду між вхідними даними та перевіреними результатами. Позначте елементи коду, визначте критерії успіху та не допускайте беззвучного часткового виконання. Зберігайте у кеші стабільні інструкції системи та схеми інструментів. Повторна передача ідентичного вступу є поширеною причиною зайвих витрат.

# ── MODULE 4: NODES ──────────────────────────────────────────
def chat_model_node(state: State) -> dict:
    """Calls the LLM with only the last 2 messages, but state keeps the full history."""

    # Filter: only send last 2 messages to the LLM
    # State is NOT modified - full history stays in the checkpoint
    recent_messages = state["messages"][-2:]

    response = llm.invoke(recent_messages)
    return {"messages": [response]}

Шаблон Б: RemoveMessage — постійне видалення зі стану

Під час роботи за схемою B: RemoveMessage — Permanently Delete from State, спочатку запишіть контракт: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Запишіть час виконання та витрати на токени або запити поруч із функціональними результатами. Відображення витрат заздалегідь запобігає несподіваним рахункам, коли процес переходить від демо-середовища до спільних. Робіть контрольні пункти після дорогих кроків. Функція відновлення не повинна знову стягувати плату за один і той самий виклик LLM, коли оператор перезапускає пізнішу ланку.

# ── IMPORTS needed ──────────────────────────────────────────
from langchain_core.messages import RemoveMessage

# ── MODULE 4: NODES ──────────────────────────────────────────
def filter_messages_node(state: State) -> dict:
    """A dedicated node that removes all but the last 2 messages from state.

    RemoveMessage works by ID: you pass it the ID of a message you want deleted,
    and the add_messages reducer handles the actual deletion. This is NOT the same
    as just slicing the list - it permanently removes from the checkpoint.
    """

    # Build a list of RemoveMessage objects for every message we want to delete
    # state["messages"][:-2] = all messages EXCEPT the last 2
    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]

    return {"messages": delete_messages}  # Reducer sees these and deletes them

def chat_model_node(state: State) -> dict:
    """After filtering, the state already has a short message list."""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# ── MODULE 6: GRAPH ASSEMBLY ─────────────────────────────────
graph_builder = StateGraph(State)
graph_builder.add_node("filter", filter_messages_node)
graph_builder.add_node("chat_model", chat_model_node)
# Filter runs FIRST, then chat_model sees the cleaned state
graph_builder.add_edge(START, "filter")
graph_builder.add_edge("filter", "chat_model")
graph_builder.add_edge("chat_model", END)
graph = graph_builder.compile(checkpointer=MemorySaver())

Стратегія 2: Обробка даних на основі токенів за допомогою trim_messages

Під час роботи над Стратегією 2: Видалення на основі токенів за допомогою trim_messages спочатку запишіть контракт: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність подальших змін у коді. Зберігайте конфігурацію окремо від коду додатку. Файли середовища, сховища секретів та флаги функцій мають знаходитися в одному місці, яке оператори можуть перевірити, не читаючи весь код. Зберігайте у кеші стабільні інструкції системи та схеми інструментів. Повторна передача ідентичних даних є поширеною причиною витрат ресурсів. Під час роботи над Стратегією 2: Видалення на основі токенів за допомогою trim_messages спочатку запишіть контракт: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність подальших змін у коді. Віддавайте перевагу малим, тестованим одиницям коду перед величезними скриптами. Коли якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану структуру.

пайплайн.

# ── IMPORTS needed ──────────────────────────────────────────
from langchain_core.messages import trim_messages

# ── MODULE 4: NODES ──────────────────────────────────────────
def chat_model_node(state: State) -> dict:
    """Uses token-based trimming to control exactly how much history the LLM sees."""

    trimmed = trim_messages(
        state["messages"],

        # Maximum number of tokens to allow (tune this to your model and budget)
        max_tokens=1000,

        # "last" = keep the MOST RECENT messages (the ones that fit within max_tokens)
        # This is almost always what you want
        strategy="last",

        # The LLM used to count tokens. Use the same model you're chatting with
        # for accurate counting. This does NOT make an API call - just counts.
        token_counter=ChatOpenAI(model="gpt-4o"),

        # False = never cut a message in half. Either include it fully or drop it.
        # Always set this to False in production to avoid feeding partial tool results.
        allow_partial=False,

        # True = always include the most recent HumanMessage even if it alone
        # exceeds max_tokens. Prevents the agent from receiving an empty context.
        include_system=True,
    )

    response = llm.invoke(trimmed)
    return {"messages": [response]}

Швидкий посібник з ключовими параметрами

Швидкий посібник з ключовими параметрами найкраще функціонує, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу роботи. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте мовчазного часткового виконання завдань. Зберігайте стан графа у простій та типованій формі. Вкладені блоки приховують інформацію про те, який вузол заповнив певне поле, що ускладнює продовження роботи після перерв.

Стратегія 3: Узагальнення на основі LLM (стандарт у продакшені)

Стратегія 3: Сумаризація на основі LLM (стандарт виробництва) працює найкраще, коли її розглядають як вимірювану характеристику. Збережіть один ідеальний запис, один випадок збою та примітку про скасування перед розширенням обсягу роботи. Записуйте час виконання та витрати на токени або запити поруч із функціональними результатами. Відомість витрат заздалегідь запобігає несподіваним рахункам під час переходу від демо-версії до спільних середовищ. Встановіть ліміти на кількість токенів за хід та за сеанс. Інструменти з агентним підходом активно розширюють контекст; жорсткі обмеження не дозволяють демо-версіям перетворюватися на несподівані рахунки.

Крок 1: Визначення завдання (Модуль 2)

Крок 1: Стан (Модуль 2) найкраще функціонує, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу. Тримайте конфігурацію окремо від коду додатку. Файли середовища, сховища конфіденційних даних та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь граф. Зберігайте стан графа у простій та типованій формі. Вкладені блоки приховують інформацію про те, який вузол записав яке поле, і ускладнюють продовження роботи після перерв.

class State(MessagesState):
    summary: str   # Stores the running summary. Empty string by default.

Крок 2: Два вузли (Модуль 4)

Крок 2: Модуль «Два вузли» (Модуль 4) найкраще функціонує, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування дій перед розширенням обсягу. Документуйте одночасно шлях успішного виконання та шлях відновлення. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації. Зберігайте стан графу у простій та типованій формі. Вкладені структури приховують інформацію про те, який вузол заповнив яке поле, та ускладнюють продовження роботи після перерв.

def call_model(state: State) -> dict:
    """The main agent node. Uses summary as context if it exists."""

    summary = state.get("summary", "")

    if summary:
        # If we have a summary, prepend it as a system message
        # This gives the LLM the compressed history without the raw messages
        system_message = SystemMessage(
            content=f"Summary of the conversation so far:\n\n{summary}"
        )
        messages = [system_message] + state["messages"]
    else:
        # No summary yet — just use the raw messages
        messages = state["messages"]

    response = llm.invoke(messages)
    return {"messages": [response]}
def summarize_conversation(state: State) -> dict:
    """Triggered when conversation gets long. Compresses history into a summary,
    then deletes the old raw messages to keep state lean."""

    existing_summary = state.get("summary", "")

    # Build the summarization prompt.
    # If a summary already exists, extend it with new messages.
    # This is the "rolling summary" pattern - it never reads all messages from scratch.
    if existing_summary:
        summary_message = (
            f"This is the summary of our conversation so far:\n{existing_summary}\n\n"
            f"Extend this summary with the new messages above. "
            f"Keep it concise - focus on facts and decisions that matter for future turns."
        )
    else:
        summary_message = (
            "Create a concise summary of the conversation above. "
            "Focus on key facts, user intent, and any decisions made. "
            "This summary will replace the full message history."
        )

    # Call the LLM with all current messages + the summarization instruction
    messages = state["messages"] + [HumanMessage(content=summary_message)]
    response = llm.invoke(messages)

    # After summarizing, delete all messages except the last 2.
    # Why keep 2? So the LLM has immediate conversational context
    # (it can see what was just said) even after summarization runs.
    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]

    return {
        "summary": response.content,    # Updated summary stored in state
        "messages": delete_messages      # Old messages removed from state
    }

Крок 3: Логіка маршрутизації (Модуль 5)

Крок 3: Логіка маршрутизації (Модуль 5) працює найкраще, якщо її розглядати як вимірювану структуру. Збережіть один ідеальний зразок роботи, один випадок збою та примітки щодо скасування змін перед розширенням обсягу роботи. Віддавайте перевагу невеликим, тестованим одиницям перед складними скриптами. Коли якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану послідовність дій. Зберігайте стан графа у простій та типованій формі. Вкладені структури приховують інформацію про те, який вузол заповнив певне поле, що ускладнює продовження роботи після перерв.

from typing import Literal
def should_continue(state: State) -> Literal["summarize_conversation", "__end__"]:
    """After the main agent responds, decide: keep going or trigger summarization?

    The threshold (6 messages here) is your tuning knob.
    Too low = you summarize frequently, losing some nuance.
    Too high = you let the context grow before compressing.
    For most production agents, 6–10 messages is a good starting range.
    """

    if len(state["messages"]) > 6:
        return "summarize_conversation"

    return "__end__"

Крок 4: Складання графа (Модуль 6)

Крок 4: Складання графа (Модуль 6) найкраще функціонує, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний зразок роботи, один випадок збою та примітку про скасування змін перед розширенням обсягу роботи. Розглядайте цей етап як угоду між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте мовчазного часткового виконання завдань. Зберігайте стан графа у простій та типованій формі. Вкладені структури приховують інформацію про те, який вузол заповнив певне поле, що ускладнює продовження роботи після перерв.

graph_builder = StateGraph(State)

graph_builder.add_node("call_model", call_model)
graph_builder.add_node("summarize_conversation", summarize_conversation)
graph_builder.add_edge(START, "call_model")
graph_builder.add_conditional_edges(
    "call_model",
    should_continue,
    {
        "summarize_conversation": "summarize_conversation",
        "__end__": END
    }
)
# After summarization, the conversation ends for this turn
graph_builder.add_edge("summarize_conversation", END)
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

Як це працює на практиці

Підхід «Як це працює на практиці» найкраще функціонує, якщо його розглядати як вимірювану поверхню. Запишіть один ідеальний приклад роботи, один випадок збою та примітку про скасування змін перед розширенням обсягу. Записуйте час виконання та витрати на токени чи запити поруч із функціональними результатами. Візуалізація витрат заздалегідь запобігає несподіваним рахункам, коли процес переходить від демо-середовища до спільних. Зберігайте стан графа у випрямленому та типованому форматі. Вкладені блоки приховують інформацію про те, який вузол заповнив яке поле, та ускладнюють продовження роботи після перерв.

Turn 1–3:   Messages accumulate. should_continue → "__end__" each time.
Turn 4:     call_model runs. len(messages) > 6 → should_continue → "summarize_conversation"
            summarize_conversation:
              - LLM reads all 7 messages, generates a 2-sentence summary
              - state["summary"] = "User is troubleshooting a Python import error..."
              - Deletes messages[:-2] (keeps only last 2 raw messages)
Turn 5:     call_model prepends summary as SystemMessage + 2 recent messages
            LLM sees: [SystemMessage(summary), HumanMessage, AIMessage, HumanMessage(latest)]
            Token cost: 4 messages instead of potentially 20+

Створення справді постійної пам’яті: зовнішні контрольні точки

Створення справді постійної пам’яті: зовнішні чекпоїнтери найкраще функціонують, коли їх розглядають як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу. Тримайте конфігурацію поза кодом додатку. Файли середовища, сховища секретів та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь граф. Тримайте стан графа простим та типованим. Вкладені блоки приховують інформацію про те, який вузол записав яке поле, і ускладнюють продовження роботи після перерв. Створення справді постійної пам’яті: зовнішні чекпоїнтери найкраще функціонують, коли їх розглядають як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу. Віддавайте перевагу малим, тестованим одиницям перед величезними скриптами. Коли якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану послідовність дій.

Інтерфейс чекпоїнтера

Для інтерфейсу Checkpointer необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість знову виконати крок, починаючи з відомої точки контролю, без необхідності здогадуватися про прихований стан. Розглядайте цю стадію як контракт між вхідними даними та перевіреними результатами. Позначте елементи проекту, визначте критерії успіху та не допускайте мовчазного часткового завершення. Запровадьте людське схвалення для операцій, які призводять до витрат грошей чи змінюють дані у продакшені. Підключення під час компіляції не є гарантією повності виконання завдань у бізнес-сенсі.

# Development: everything in RAM (lost on restart)
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()

# Local / Single-machine production: persists to a .db file on disk
from langgraph_checkpoint_sqlite import SqliteSaver
import sqlite3
conn = sqlite3.connect("chat_memory.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
# Cloud / Distributed production: persists to a Postgres database
from langgraph_checkpoint_postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:password@host:5432/dbname"
)
# This one line is the ONLY thing that changes between environments:
graph = graph_builder.compile(checkpointer=checkpointer)

SqliteSaver: детальний огляд

Щоб детальніше ознайомитися з SqliteSaver, необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Записуйте час виконання та витрати на токени чи запити поруч із функціональними результатами. Візуалізація витрат заздалегідь запобігає несподіваним рахункам під час переходу з демо-середовища у спільні. Встановлюйте людське схвалення для кроків, які витрачають гроші чи змінюють дані в продакшені. Підключення під час компіляції не є гарантією повноти бізнес-функціоналу.

# ── Full SqliteSaver Setup ───────────────────────────────────
import sqlite3
from langgraph_checkpoint_sqlite import SqliteSaver

# File-based: state persists across restarts
conn = sqlite3.connect(
    "agent_memory.db",
    check_same_thread=False  # Required for LangGraph's multi-threaded execution
)
checkpointer = SqliteSaver(conn)
graph = graph_builder.compile(checkpointer=checkpointer)
# Usage is IDENTICAL to MemorySaver:
config = {"configurable": {"thread_id": "user-123"}}
response = graph.invoke({"messages": [HumanMessage(content="Hello")]}, config)
# Stop and restart your app. Run this again with the SAME thread_id:
# The agent picks up exactly where it left off.
response = graph.invoke({"messages": [HumanMessage(content="What did I just say?")]}, config)
# Agent: "You said 'Hello'."

Коли використовувати SqliteSaver замість PostgresSaver

Щодо того, коли використовувати SqliteSaver чи PostgresSaver, необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість знову запустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Зберігайте конфігурацію поза кодом додатку. Файли середовища, сховища секретів та флаги функцій мають знаходитися в одному місці, яке оператори можуть перевірити, не читаючи весь граф. Встановлюйте людське схвалення для кроків, які спричиняють витрати чи змінюють дані у продакшені. Підключення на етапі компіляції не є гарантією повноти бізнес-функціоналу. Щодо того, коли використовувати SqliteSaver чи PostgresSaver, необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість знову запустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Віддавайте перевагу малим, тестованим одиницям перед величезними скриптами. Коли крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на щось більше.

заплутана система трубопроводів.

Перевірка пам’яті: get_state та get_state_history

Під час роботи з функціями Перевірка пам’яті: get_state та get_state_history спочатку складіть контракт: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допоможе зберегти чесність подальших змін у коді. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте беззвучного часткового виконання завдань. Створюйте контрольні точки після дорогих операцій. Функція відновлення не повинна знову оплачувати однаковий виклик LLM, коли оператор намагається виконати наступний етап.

# ── Inspecting Current State ─────────────────────────────────
config = {"configurable": {"thread_id": "user-123"}}
# Get the current state snapshot for this thread
state_snapshot = graph.get_state(config)
# What's in the snapshot?
print(state_snapshot.values["messages"])  # Current message list
print(state_snapshot.values.get("summary", ""))  # Summary, if any
print(state_snapshot.next)  # Which nodes would run next (empty list = graph is at END)
# ── Inspecting State History ─────────────────────────────────
# get_state_history gives you every checkpoint ever saved for this thread.
# Each checkpoint is a moment in time after a node ran.
for state in graph.get_state_history(config):
    print(f"Step: {state.metadata['step']}")
    print(f"Messages: {len(state.values['messages'])}")
    print(f"Next: {state.next}")
    print("---")

Об’єднання всього: повний агент з керуванням пам’яттю

Під час роботи над книгою «Putting It All Together: The Complete Memory-Managed Agent» спочатку запишіть умови використання: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Запишіть час виконання та витрати на токени або запити поруч із функціональними результатами. Чітке бачення витрат заздалегідь запобігає несподіваним рахункам, коли процес переходить від демо-версії до спільних середовищ. Робіть контрольні пункти після дорогих кроків. Функція відновлення не повинна знову стягувати плату за один і той самий виклик LLM, коли оператор перезапускає пізніший етап.

# ============================================================
# LANGGRAPH MEMORY-MANAGED AGENT — COMPLETE TEMPLATE
# Combines: Summarization + External Persistence (SqliteSaver)
# ============================================================

# ── MODULE 1: IMPORTS & CONFIGURATION ───────────────────────
import os
import sqlite3
from typing import Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import (
    HumanMessage, AIMessage, SystemMessage,
    BaseMessage, RemoveMessage
)
from langgraph.graph import StateGraph, START, END
from langgraph.graph import MessagesState
from langgraph.checkpoint.memory import MemorySaver
# For production, swap MemorySaver for one of these:
# from langgraph_checkpoint_sqlite import SqliteSaver
# from langgraph_checkpoint_postgres import PostgresSaver
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ── MODULE 2: STATE ─────────────────────────────────────────
class State(MessagesState):
    # summary stores the rolling compressed history.
    # It starts empty and gets written by summarize_conversation.
    summary: str

# ── MODULE 4: NODES ─────────────────────────────────────────
def call_model(state: State) -> dict:
    """Main reasoning node. Uses rolling summary for long-context efficiency."""

    summary = state.get("summary", "")

    if summary:
        system_message = SystemMessage(
            content=f"Summary of the conversation so far:\n\n{summary}"
        )
        messages = [system_message] + state["messages"]
    else:
        messages = state["messages"]

    response = llm.invoke(messages)
    return {"messages": [response]}

def summarize_conversation(state: State) -> dict:
    """Compression node. Summarizes history and prunes old messages from state."""

    existing_summary = state.get("summary", "")

    if existing_summary:
        summary_prompt = (
            f"Current conversation summary:\n{existing_summary}\n\n"
            f"Extend this summary with the new messages. "
            f"Be concise. Preserve facts, user intent, and key decisions."
        )
    else:
        summary_prompt = (
            "Summarize this conversation concisely. "
            "Focus on facts, user intent, and decisions made."
        )

    messages = state["messages"] + [HumanMessage(content=summary_prompt)]
    response = llm.invoke(messages)

    # Keep only the 2 most recent raw messages in state.
    # Everything older is now captured in the summary.
    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]

    return {
        "summary": response.content,
        "messages": delete_messages
    }

# ── MODULE 5: ROUTING ────────────────────────────────────────
def should_continue(state: State) -> Literal["summarize_conversation", "__end__"]:
    """Trigger summarization when conversation exceeds 6 messages.

    Tune this threshold based on your average message length and token budget.
    For GPT-4o at ~$0.005/1K tokens, 6 messages is a practical starting point.
    """
    if len(state["messages"]) > 6:
        return "summarize_conversation"
    return "__end__"

# ── MODULE 6: GRAPH ASSEMBLY ─────────────────────────────────
graph_builder = StateGraph(State)
graph_builder.add_node("call_model", call_model)
graph_builder.add_node("summarize_conversation", summarize_conversation)
graph_builder.add_edge(START, "call_model")
graph_builder.add_conditional_edges(
    "call_model",
    should_continue,
    {
        "summarize_conversation": "summarize_conversation",
        "__end__": END
    }
)
graph_builder.add_edge("summarize_conversation", END)
# ── CHECKPOINTER SELECTION ───────────────────────────────────
# Development (in-memory, lost on restart):
checkpointer = MemorySaver()
# Uncomment for local persistence (survives restarts):
# conn = sqlite3.connect("agent_memory.db", check_same_thread=False)
# checkpointer = SqliteSaver(conn)
# Uncomment for cloud/production (requires PostgreSQL):
# checkpointer = PostgresSaver.from_conn_string(os.environ["DATABASE_URL"])
graph = graph_builder.compile(checkpointer=checkpointer)

# ── MODULE 7: ENTRYPOINT ─────────────────────────────────────
if __name__ == "__main__":
    config = {"configurable": {"thread_id": "user-001"}}

    print("Memory-managed agent ready. Type 'exit' to quit, 'state' to inspect memory.\n")

    while True:
        user_text = input("You: ").strip()

        if not user_text or user_text.lower() == "exit":
            break

        # Debug command: inspect what the agent currently remembers
        if user_text.lower() == "state":
            snapshot = graph.get_state(config)
            print(f"\n[DEBUG] Messages in state: {len(snapshot.values['messages'])}")
            print(f"[DEBUG] Summary: {snapshot.values.get('summary', '(none)')}\n")
            continue

        response = graph.invoke(
            {"messages": [HumanMessage(content=user_text)]},
            config=config
        )

        print(f"Agent: {response['messages'][-1].content}\n")

Посібник з вибору стратегії керування пам’яттю

Під час роботи з «The Memory Strategy Decision Guide» спочатку запишіть умови використання: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність подальших змін у коді. Тримайте конфігурацію окремо від коду додатку. Файли середовища, сховища конфіденційних даних та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь код. Створюйте контрольні точки після дорогих кроків. Система відновлення не повинна знову стягувати плату за один і той самий виклик LLM, коли оператор перезапускає пізніший етап. Під час роботи з «The Memory Strategy Decision Guide» спочатку запишіть умови використання: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність подальших змін у коді. Віддавайте перевагу невеликим, тестованим одиницям коду перед величезними скриптами. Коли якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану послідовність операцій.

Оновлена карта довідки ключових слів

Оновлена карта довідки ключових слів найкраще функціонує, якщо її розглядати як вимірювану основу. Збережіть один ідеальний зразок результату, один випадок невдачі та примітку про скасування змін перед розширенням обсягу роботи. Розглядайте цей етап як угоду між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте мовчазного часткового виконання завдань. Зберігайте структуру графа простою та типованою. Вкладені елементи приховують інформацію про те, який вузол заповнив певне поле, що ускладнює продовження роботи після перерв.

Висновок: «Пам’ять — це архітектура, а не щось додане згодом» найкраще працює, якщо її розглядати як вимірювану характеристику. Запишіть один ідеальний приклад роботи, один випадок збою та примітки щодо скасування змін перед розширенням обсягу роботи. Фіксуйте час виконання та витрати на обробку даних поруч із функціональними результатами. Чітке бачення витрат заздалегідь запобігає несподіваним рахункам під час переходу від демо-версії до спільних середовищ. Тримайте стан графа простим та типованим; вкладені структури даних приховують інформацію про те, який вузол заповнив певне поле, що ускладнює продовження роботи після перерв.

Чек-лист операцій

Для чек-листу операцій необхідно визначити вхідні дані, відповідальну особу за кожен крок та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан системи.

Документуйте як успішний, так і відновлювальний сценарії роботи разом. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.

Необхідно отримувати схвалення людини для операцій, які спричиняють витрати грошей чи змінюють дані виробництва. Підключення під час компіляції не є гарантією повноти бізнес-функціоналу.

Напишіть короткий посібник: як змінювати ключі, як спорожнювати чергу, як скасовувати останню операцію вставки даних.

Віддавайте перевагу невеликим, тестованим одиницям коду перед складними скриптами. Якщо якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану послідовність операцій.

Необхідно отримувати схвалення людини для операцій, які спричиняють витрати грошей чи змінюють дані виробництва. Підключення під час компіляції не є гарантією повноти бізнес-функціоналу.

Перш ніж підвищувати рівень складності системи, заморозьте версії, зафіксуйте ідеальний варіант виконання операцій для критичних шляхів та підтвердьте кроки скасування змін. У спільних середовищах необхідні обмеження на частоту виконання операцій, перевірки прав доступу та чіткий власник для зміни секретних даних. Віддавайте перевагу надійності перед креативними, одноразовими демонстраціями.

Примітка до пакету d865d505a59e: не включайте ключі постачальників у репозиторій, встановіть ліміт токенів на сеанс та зберігайте транскрипції поруч із фіксами для оцінки, щоб подальша заміна моделей залишалася порівнянною.