حافظه‌ی گفتگو: checkpointer، thread و مدیریت تاریخچه

در دوره‌ی اول دیدیم مدل حافظه ندارد و برای اینکه اسم کاربر را بداند، باید کل تاریخچه را در هر درخواست بفرستیم. این کار را دستی با یک لیست انجام دادیم. در برنامه‌ی واقعی چند مشکل دیگر هم هست: کاربرهای مختلف هم‌زمان گفتگو می‌کنند و تاریخچه‌شان نباید قاطی شود، گفتگوها طولانی و گران می‌شوند، و با ری‌استارت شدن سرور همه‌چیز از دست می‌رود.

pip install -U langchain langchain-openai langgraph

create_agent و checkpointer#

در LangChain نسخه‌ی ۱، حافظه‌ی گفتگو از طریق create_agent ساخته می‌شود. ایجنت‌ها و ابزارها را در دوره‌ی بعد کامل می‌بینیم. فعلاً ایجنتی می‌سازیم که هیچ ابزاری ندارد و فقط گفتگو می‌کند. این روش جای کلاس‌های قدیمی ConversationChain و ConversationBufferMemory را گرفته است.

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

llm = init_chat_model("gpt-5.2", model_provider="openai", temperature=0)

agent = create_agent(
    model=llm,
    tools=[],                        # بدون ابزار؛ فقط گفتگو
    checkpointer=InMemorySaver(),    # حافظه را در RAM نگه می‌دارد
    system_prompt="You are a helpful assistant.",
)

config = {"configurable": {"thread_id": "session-1"}}

• checkpointer وضعیت گفتگو را بعد از هر نوبت ذخیره می‌کند. InMemorySaver آن را در حافظه‌ی RAM نگه می‌دارد.
• thread_id شناسه‌ی هر گفتگوست. هر رشته‌ای می‌تواند باشد، مثلاً شناسه‌ی کاربر یا شناسه‌ی جلسه‌ی چت.

def ask(text, config):
    response = agent.invoke({"messages": [{"role": "user", "content": text}]}, config=config)
    return response["messages"][-1].content

ask("Hi, my name is Alireza", config)   # Hi Alireza — nice to meet you. …
ask("What is 1+1?", config)             # 1 + 1 = 2.
ask("What is my name?", config)         # Your name is **Alireza**.

فقط پیام جدید را فرستادیم، اما ایجنت اسم را به یاد داشت.

هر thread یک گفتگوی جدا#

config2 = {"configurable": {"thread_id": "session-2"}}
ask("my name is Gholam", config2)    # Hi Gholam — nice to meet you. …

ask("what is my name", config)       # Your name is **Alireza**.
ask("what is my name", config2)      # Your name is **Gholam**.

ask("what is my name", {"configurable": {"thread_id": "gtkkgthgthgth"}})
# I don't know your name from our conversation so far. …

پشت صحنه: state#

ایجنت یک state (وضعیت) دارد که در طول گفتگو تغییر می‌کند. مهم‌ترین بخشش لیست پیام‌هاست. در هر نوبت این اتفاق‌ها می‌افتد:

1. پیام‌های قبلی این thread از checkpointer خوانده می‌شوند.
2. پیام جدید کاربر به آن‌ها اضافه می‌شود.
3. کل لیست به مدل فرستاده می‌شود.
4. پاسخ مدل هم به لیست اضافه و دوباره ذخیره می‌شود.

یعنی همان کاری که در دوره‌ی اول دستی انجام دادیم. می‌شود state را دید و حتی تغییرش داد:

state = agent.get_state(config)
for msg in state.values["messages"]:
    print(f"{type(msg).__name__}: {msg.content[:200]}")

HumanMessage: Hi, my name is Alireza
AIMessage: Hi Alireza — nice to meet you. What can I help you with today?
HumanMessage: What is 1+1?
AIMessage: 1 + 1 = 2.
HumanMessage: What is my name?
AIMessage: Your name is **Alireza**.
…

وقتی تاریخچه طولانی می‌شود#

ارائه‌دهنده‌ها به ازای همه‌ی توکن‌های ورودی پول می‌گیرند، یعنی کل تاریخچه در هر نوبت. پنجره‌ی کانتکست هم سقف دارد. پس تاریخچه را باید مدیریت کرد. LangChain برای این کار middleware دارد: تابعی که قبل یا بعد از فراخوانی مدل اجرا می‌شود و می‌تواند state را تغییر دهد.

۱. فقط چند پیام آخر (Trim)#

این روش جای ConversationBufferWindowMemory قدیمی را می‌گیرد:

from typing import Any
from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langchain.agents import AgentState
from langchain.agents.middleware import before_model
from langgraph.runtime import Runtime

@before_model
def trim_old_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """فقط ۴ پیام آخر را نگه می‌دارد"""
    messages = state["messages"]
    if len(messages) <= 4:
        return None                       # تغییری لازم نیست
    return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *messages[-4:]]}

agent_window = create_agent(model=llm, tools=[], middleware=[trim_old_messages],
                            checkpointer=InMemorySaver(), system_prompt="You are a helpful assistant.")

تابع دو چیز دریافت می‌کند: state (پیام‌ها) و runtime (اطلاعات اجرا). خروجی‌اش تغییری است که روی state اعمال می‌شود: «همه را پاک کن، بعد این چهار تا را بگذار». None یعنی دست نزن.

# اسم ← ۱+۱ ← ۲+۲ ← «اسم من چیه؟»
# I don't know your name from this conversation. …

پیام اسم از پنجره بیرون رفته بود. یک حالت جالب هم پیش آمد. وقتی فقط یک سؤال وسط بود (اسم ← ۲+۲ ← «اسم من چیه؟»)، پیام اول کاربر حذف شده بود، اما پاسخ مدل «Hi Alireza» هنوز در پنجره بود. مدل جواب داد:

I don't actually know your name unless you tell me. In this chat, I addressed you as "Alireza," but that was just an assumption—not something I can verify.

یعنی بریدن مکانیکی تاریخچه ممکن است تکه‌های نیمه‌کاره‌ای باقی بگذارد که مدل را گیج می‌کنند.

۲. حذف بعد از پاسخ (Delete)#

from langchain.agents.middleware import after_model

@after_model
def delete_old_messages(state: AgentState, runtime: Runtime) -> dict | None:
    """بعد از هر پاسخ، اگر ۴ پیام یا بیشتر داشتیم، ۲ پیام اول را پاک کن"""
    messages = state["messages"]
    if len(messages) >= 4:
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}
    return None

این‌بار به‌جای «همه را پاک کن» پیام‌ها با شناسه‌شان (m.id) حذف می‌شوند، و به‌جای قبل از مدل، بعد از آن اجرا می‌شود. نتیجه شبیه قبل است: بعد از «اسمم علی است»، «در تهران زندگی می‌کنم» و «۳۵ سالمه»، مدل دیگر اسم را نمی‌دانست.

۳. خلاصه‌سازی خودکار (Summarization)#

دو روش قبل اطلاعات را دور می‌ریزند. روش بهتر این است که پیام‌های قدیمی خلاصه شوند. این روش جای ConversationSummaryBufferMemory قدیمی را می‌گیرد:

from langchain.agents.middleware import SummarizationMiddleware

agent_summary = create_agent(
    model=llm,
    tools=[],
    middleware=[SummarizationMiddleware(
        model=llm,                      # می‌تواند مدل ارزان‌تری باشد
        trigger=("tokens", 200),        # وقتی تاریخچه از ۲۰۰ توکن گذشت
        keep=("messages", 4),           # ۴ پیام آخر دست‌نخورده بماند
    )],
    checkpointer=InMemorySaver(),
    system_prompt="You are a helpful assistant.",
)

در کلاس چند پیام بی‌اهمیت فرستادیم («سلام»، «فقط دارم وقت می‌گذرونم») و بعد برنامه‌ی روز را دادیم: «ساعت ۸ جلسه با تیم محصول … ساعت ۱۲ ناهار با یک مشتری … حتماً لپ‌تاپت رو بیار تا آخرین دموی LLM رو نشون بدی». بعد پرسیدیم «چه دموی خوبی می‌شه نشون داد؟». مدل با در نظر گرفتن همان برنامه، چند پیشنهاد مرتبط داد. state بعد از خلاصه‌سازی این‌طور شروع می‌شد:

HumanMessage: Here is a summary of the conversation to date:

## SESSION INTENT
None (user has not requested a specific task; they are just passing time).

## SUMMARY
- User greeted in Persian ("سلام").
- Assistant responded in Persian asking how to help.
…

پیام‌های قدیمی جای خودشان را به یک پیام خلاصه دادند و پیام‌های اخیر دست‌نخورده ماندند. پرامپتی که خلاصه را می‌سازد بخش‌هایی مثل هدف کاربر و خلاصه دارد. اگر لازم باشد، می‌شود با پارامتر summary_prompt پرامپت خودتان را جایگزینش کنید، مثلاً «اسم‌ها، تاریخ‌ها و تصمیم‌ها را حتماً نگه دار».

حافظه‌ی ماندگار: SQLite و Postgres#

InMemorySaver با بسته شدن برنامه پاک می‌شود. برای نگه داشتن گفتگو بین اجراها، checkpointer را عوض کنید. بقیه‌ی کد تغییری نمی‌کند:

# pip install langgraph-checkpoint-sqlite
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

conn = sqlite3.connect("chat_memory.db", check_same_thread=False)
agent = create_agent(model=llm, tools=[], checkpointer=SqliteSaver(conn),
                     system_prompt="You are a helpful assistant.")

config = {"configurable": {"thread_id": "session-1"}}
ask("اسم من چیه؟", config)
# اسم شما «سارا» است.

در این اجرا هیچ اسمی نگفته بودیم، اما مدل «سارا» را می‌دانست. در اجرای قبلی، که روی همان فایل و همان thread بود، کاربر خودش را سارا معرفی کرده بود و گفتگو روی دیسک مانده بود.

برای استفاده در یک بلوک with هم شکل کوتاه‌تری هست:

with SqliteSaver.from_conn_string("chat_memory.db") as checkpointer:
    agent = create_agent(model=llm, tools=[], checkpointer=checkpointer)
    ...

در محیط عملیاتی معمولاً از PostgresSaver استفاده می‌شود (pip install langgraph-checkpoint-postgres):

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:password@localhost:5432/mydb"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()        # بار اول جدول‌ها را می‌سازد
    agent = create_agent(model=llm, tools=[], checkpointer=checkpointer)

خلاصه#

• create_agent با checkpointer حافظه‌ی گفتگو می‌سازد و thread_id گفتگوها را از هم جدا می‌کند.
• state ایجنت لیست پیام‌هاست و با get_state قابل مشاهده است.
• middlewareها قبل یا بعد از مدل state را تغییر می‌دهند: Trim با before_model، حذف با after_model، و SummarizationMiddleware برای خلاصه کردن پیام‌های قدیمی.
• برای حافظه‌ی ماندگار از SqliteSaver یا PostgresSaver استفاده کنید.