Agent 시리즈 (6): 메모리 관리 - Agent가 중요한 일을 기억하게 하기

기억, Agent을 "도구"에서 "도우미"로 바꿔줍니다.

기억이 없는 Agent은 각 대화가 처음부터 시작합니다. 그에게 '이름은 리레이이고 파이썬 엔지니어이며 손으로 직접 해보는 것을 좋아한다'고 말해도 다음 대화에서는 전혀 기억하지 못합니다.

이것은 버그가 아니라 아키텍처의 결함입니다.

LLM 자체에는 영속적인 기억이 없으며, 각 호출은 무상태입니다. Agent이 일을 기억하려면 아키텍처 레이어에서 명시적으로저장하고 관리하고 읽어들여야 합니다.기억

이것이 기억 관리 모듈이 해결해야 할 문제입니다.

이 글은 Agent 기억을 네 가지 차원으로 분석합니다: 기억 유형의 분류 모델, 맥락 관리의 세 가지 전략, LangGraph가 제공하는 두 가지 기억 메커니즘(checkpointer와 store), 그리고 초장 대화의 자동 압축 방안입니다.__JHSNS_SEG_a04323c2_8__네 가지 기억 유형: 인지 과학에서 공학 실현까지

인지 과학에서 인간의 기억의 분류를 참고하여, Agent의 기억도 네 개의 레벨로 나눌 수 있습니다. 각 레벨은 LangGraph에서 대응되는 구현 방식이 있습니다:

┌──────────────────────────────────────────────────────────┐
│                     记忆层级                              │
├──────────────────────┬───────────────────────────────────┤
│ 感觉记忆 (Sensory)   │ 当前 Turn 的 in-flight 消息       │
│                      │ 生命周期：单次 LLM 调用            │
├──────────────────────┼───────────────────────────────────┤
│ 工作记忆 (Working)   │ 对话历史 MessageHistory（有限 K 轮）│
│                      │ 实现：messages 列表注入 Prompt     │
├──────────────────────┼───────────────────────────────────┤
│ 情景记忆 (Episodic)  │ 向量化/摘要化的历史片段            │
│                      │ 实现：摘要压缩 + VectorStore 检索  │
├──────────────────────┼───────────────────────────────────┤
│ 语义记忆 (Semantic)  │ 长期存储的用户偏好、事实           │
│                      │ 实现：LangGraph store (KV Store)  │
└──────────────────────┴───────────────────────────────────┘

감각 기억: 현재 턴의 메시지

가장 짧은 기억. 한 번의 LLM 호출의 입력과 출력은 사용하고 버립니다:

q = "Python 中 len([1, 2, 3]) 等于多少"
answer = llm.invoke([HumanMessage(q)])
# answer.content → "len([1, 2, 3]) 的结果等于 3。"
# 这次 invoke 结束后，这个 answer 就消失了

감각 기억을 "관리"하기 위해 어떤 메커니즘도 필요하지 않습니다 - 그것은 LLM 호출 자체입니다.

작업 기억: 제한된 대화 이력

이전 몇 라운드의 대화 메시지를 프롬프트에 연결하는 것이 가장 직접적인 기억 구현입니다. 효과는 즉각적입니다:

history = [
    HumanMessage("我叫李雷，是一名 Python 工程师"),
    AIMessage("你好，李雷！很高兴认识你。"),
    HumanMessage("我最近在学习 LangGraph"),
    AIMessage("LangGraph 很强大，特别适合构建有状态的 Agent。"),
]
test_q = "我之前告诉你我叫什么名字？"

실제 비교:

有历史 → "是的，你之前告诉我你的名字是李雷，并且你是一名 Python 工程师..."
无历史 → "抱歉，我无法回忆起您之前告诉我的名字，因为作为一个 AI，我没有
          持久的记忆功能来存储个人数据..."

차이는 매우 직관적입니다. 작업 기억의 한계는 토큰 비용이 대화 길이에 비례하여 선형적으로 증가하는 것입니다. 잘라내거나 요약을 결합하여 관리해야 합니다.

상황 기억: 요약된 역사적 조각

대화 이력이 매우 길 때, 전체 이력을 직접 프롬프트에 넣는 비용이 너무 높습니다. 맥락 기억 방법은 먼저 압축한 다음, 저장 입니다:

long_history = history * 4  # 16 条消息
summary = llm.invoke([
    SystemMessage("将以下对话压缩为 60 字以内的摘要，保留关键信息"),
    HumanMessage(str([m.content for m in long_history])),
])
# → "李雷，Python工程师，积极学习LangGraph，赞其强大，适合构建有状态Agent。"

16개의 메시지를 28자로 압축하고, 다음 라운드에는 요약을 원본 이력 대신 사용하여 토큰 소모를 크게 줄입니다.

의미 기억: 회의 간의 사용자 사실

가장 오래 지속되는 기억 계층입니다. 대화가 끝나도 사라지지 않으며, 사용자에 대한 장기적인 사실(이름, 직업, 선호 등)을 전용으로 저장합니다:

# 把用户信息存入 KV Store，下一次会话直接读取
user_profile = {
    "name": "李雷",
    "role": "Python 工程师",
    "interests": ["LangGraph", "Agent 开发"],
    "level": "中级",
}
# 基于这些信息，Agent 能给出个性化的回答
# "下一步学习方向" → 推荐 LangGraph 进阶用法而不是 Python 入门

맥락 관리 세 가지 전략: 잘라내기 / 요약 / 검색

대화 이력이 점점 길어지면서, 맥락 창이 감당하지 못하면 어떡해요? 세 가지 전략은 각각의 선택 사항이 있습니다:

전략 1: 잘라내기 (Truncation)

가장 간단하고 무뚝뚝한 방법 - 최근 N개의 메시지를 유지하고 나머지는 모두 버립니다:

# 保留最近 4 条消息
truncated = history[-4:]
resp = llm.invoke(truncated + [HumanMessage(test_q)])

실제로 8개의 주제(16개의 메시지)의 역사를 사용하고, 최근 4개를 잘라내서 "Python 리스트는 무엇인가요"라고 묻는 것:

截断后最早可见：'解释一下 Python 装饰器'（第 5 个主题，"列表"在第 1 个）
回答：Python 中的列表是一种内置的数据结构... （靠 LLM 自身知识回答）
⚠ 丢失了我们"学过"列表这个事实，LLM 只是在用通用知识回答

적용 시나리오：역사의 연속성 요구가 높지 않은 시나리오 또는 역사 자체가 중요하지 않은 순수 질의응답 유형의 에이전트.

전략 2: 요약(Summarization)

LLM을 사용하여 긴 역사를 한 구절의 요약으로 압축하고, 이후 대화에서는 원본 역사 대신 요약을 사용:

summary_resp = llm.invoke([
    SystemMessage("将对话历史压缩为一段摘要（不超过 80 字），保留所有已介绍的主题名称"),
    HumanMessage("\n".join([f"{m.type}: {m.content}" for m in history])),
])
# → "Python列表可变有序，元组不可变省内存，字典键值对映射，集合唯一元素，
#    函数封装逻辑，类面向对象，装饰器函数包装，生成器惰性计算。"
# 16 条消息 → 66 字摘要

같은 질문 "Python 리스트는 무엇인가요"에 대해, 요약 방안의 답변은 "이것은 우리가 논의한 주제입니다"를 반영할 수 있으며, 단순한 일반 지식이 아닙니다.

세 가지 전략 비교：

전략	Token 소모	정보 보존	복잡도	적용 시나리오
잘라내기	최소	최근에만	낮음	순수 질의응답, 과거는 중요하지 않음
요약	낮음	전체 맥락	중간	교육, 상담, 장기 대화
검색	최소	정확한 관련성	높음（벡터 라이브러리 필요）	지식 베이스、다 분야 Agent

전략 3: 검색(Retrieval)

현재 질문의 의미와 관련된 과거 조각만 가져옴——가장 효과적이면서도 가장 복잡한 방법:

# 简化演示：按关键词过滤（生产中用向量相似度）
relevant = [m for m in history if "列表" in m.content or "list" in m.content.lower()]
# 16 条 → 3 条相关历史
resp = llm.invoke(relevant + [HumanMessage(test_q)])

적용 시나리오 ：지식 베이스형 Agent, 사용자가 많은 이력을 가진 개인 비서.

LangGraph checkpointer: 세션 내 상태 영속화

LangGraph의 MemorySaver (checkpointer)는 thread_id를 사용하여 다른 세션을 구분하고, 동일한 세션 내에서 자동으로 대화 이력을 축적:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent

checkpointer = MemorySaver()
agent = create_react_agent(model=llm, tools=[get_weather], checkpointer=checkpointer)

# 同一 thread_id = 同一个会话，状态持久化
config_a = {"configurable": {"thread_id": "weather_001"}}

세션 내 연속 세 번째 날씨 질문 테스트

[Turn 1] 用户: 北京今天天气怎么样？
         Agent: 北京今天的天气是晴，温度为 25°C，东北风 3 级，空气良好。
         (当前状态中消息数: 4)

[Turn 2] 用户: 那上海呢？          ← "那" 字没有明确指代，需要上文理解
         Agent: 上海今天的天气是多云，22°C，东南风 2 级，轻度雾霾。
         (当前状态中消息数: 8)

[Turn 3] 用户: 这两个城市哪个更适合今天出行？   ← 需要前两轮的查询结果
         Agent: 考虑到上海的雾霾情况，建议您在北京出行。
         (当前状态中消息数: 10)

두 번째와 세 번째 턴의 답변은 이전의 이력에 의존했으며, checkpointer는 자동으로 터널 간 맥락 관리를 완료했습니다.

세션 격리

다른 thread_id 사이는 완전히 독립적입니다:

[新会话 - thread_id: weather_002]
用户: 我刚才问的是哪个城市？
Agent: 您刚才问的是"哪个城市"，但是没有提供具体的城市名称。
      如果您需要查询某个城市的天气，请告诉我具体的城市名称。
→ 新 thread_id 没有任何历史，完全不知道刚才问过什么

세션은 계속

동일한 thread_id 사이에 며칠 간격을 두고 돌아오면, 이력은 여전히 존재합니다:

[会话 A 继续 - 同一 thread_id]
用户: 刚才查的两个城市，再查一下深圳对比一下
Agent: 深圳今天的天气是阵雨，27°C，西南风 2 级，同时还有雷暴预警。
       （进行了北京/上海/深圳三城比较）
→ 记住了前面查过北京和上海

MemorySaver는 메모리 구현이며, 프로세스 재시작 후 데이터가 손실됩니다. 생산 환경에서는 SqliteSaver (로컬 파일) 또는 PostgresSaver (데이터베이스)를 사용하여 대체합니다.

LangGraph InMemoryStore: 터널 간 장기 기억

checkpointer는 단일 세션 내의 기억을 해결합니다. 터널 간의 장기 기억은 필요합니다.store：

checkpointer  →  绑定 thread_id，会话生命周期内有效
store         →  绑定 user_id，跨会话永久存在

핵심 API

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# 写入：(namespace, key, value)
store.put(("user_facts", user_id), key, {"fact": "李雷，后端工程师"})

# 读取：搜索某个 namespace 下的所有条目
facts = store.search(("user_facts", user_id))
for item in facts:
    print(item.value["fact"])

# 精确读取
item = store.get(("user_facts", user_id), specific_key)

세션 간 기억 테스트

세션 A에서, 에이전트는 대화에서 사용자 정보를 자동으로 추출하여 저장소에 저장합니다:

[会话 A] 用户说了三句话 → 自动提取并存储：
  • 李雷，后端工程师
  • Python, Go, LangGraph, Agent 开发
  • 动手实践，不喜欢纯看文档

완전히 다른 세션 B(새로운 thread_id, 하지만 동일한 user_id)에서 "너 나를 알아?"라고 묻습니다:

[会话 B - 全新 thread_id]
用户: 你好，你认识我吗？
Agent: 你好！根据你提供的信息，我认识你。你是李雷，一位后端工程师，
       擅长使用 Python、Go、LangGraph 和 Agent 进行开发。
       你更喜欢动手实践，而不是仅仅阅读文档。有什么可以帮助你的吗？
→ 虽然是全新 thread_id，但 store 中的用户信息跨会话持久

다른 user_id의 데이터는 완전히 분리되어 서로 방해하지 않습니다.

checkpointer vs store 비교

# 短期记忆：checkpointer — 绑定 thread_id，会话内有效
app = graph.compile(checkpointer=MemorySaver())
result = app.invoke(input, config={"configurable": {"thread_id": "abc"}})

# 长期记忆：store — 绑定 user_id，跨会话有效
store = InMemoryStore()
app = graph.compile(store=store, checkpointer=MemorySaver())
# 在 node 中操作 store
store.put(("user_facts", user_id), key, {"fact": "..."})
stored = store.search(("user_facts", user_id))

생산 환경에서 InMemoryStore를 PostgresStore 또는 RedisStore로 대체하면 실제 영구 저장이 가능합니다.

자동 요약 압축: RemoveMessage + 요약 순환

메시지 수가 임계값을 초과하면 자동 압축이 트리거됩니다—이것은 무한한 대화에서 에이전트가 정신을 차리도록 하는 핵심 메커니즘입니다.

그래프 구조

[chat 节点]
    │
    ├─ 消息数 ≤ 阈值 → END
    └─ 消息数 > 阈值 → [compress 节点] → END

compress 노드: RemoveMessage는 오래된 메시지를 삭제합니다

def compress_node(state: SummaryState) -> dict:
    messages = state["messages"]
    to_compress = messages[:-2]   # 保留最新 2 条，其余全部压缩
    keep = messages[-2:]

    # 旧消息 → 新摘要
    new_summary = llm.invoke([
        SystemMessage("将以下内容压缩为 120 字以内的摘要"),
        HumanMessage(existing_summary + old_messages_text),
    ]).content

    # RemoveMessage：告诉 add_messages reducer 删除这些消息
    remove_ops = [RemoveMessage(id=m.id) for m in to_compress]
    return {"messages": remove_ops, "summary": new_summary}

RemoveMessage LangGraph 전용 메시지 삭제 연산자로, add_messages reducer는 이를 볼 때 상태에서 해당 ID의 메시지를 삭제합니다.

실제 효과

11 라운드 대화, 임계값을 8개 메시지로 설정:

[Turn 1-4]  消息数:  2/4/6/8  | 摘要: ○ 无

  [压缩触发] 10 条 → 压缩 8 条，保留 2 条
  [新摘要]   Python列表常用方法包括查找、排序、添加删除等。
             `dict.get()` 避免 `KeyError`，返回默认值。
             `*args` 接收任意位置参数，`**kwargs` 接收任意关键字参数...

[Turn 5]    消息数:  2  | 摘要: ✓ 已压缩  ← 压缩后从 2 条重新开始

  [压缩再次触发] 10 条 → 压缩 8 条，保留 2 条

[Turn 11]  最终汇总：
           "根据我们之前的讨论，以下是您掌握的 Python 知识点汇总：
            1. Python 列表推导式...
            2. 集合推导式...  ← 通过摘要链条传承，第 1 轮的内容还在
            3. Lambda 函数..."

주요 결과:11 라운드 대화에서 항상 2-8개의 active 메시지만 존재합니다는, 요약 체인을 통해 모든 역사적 지식이 계승되었습니다.

요약 상태 설계

class SummaryState(TypedDict):
    messages: Annotated[list, add_messages]  # add_messages 处理 RemoveMessage
    summary: Optional[str]                   # 累积的历史摘要，注入 system prompt

def chat_node(state: SummaryState) -> dict:
    summary = state.get("summary") or ""
    system_parts = ["你是助手。"]
    if summary:
        system_parts.append(f"\n\n【历史摘要】{summary}")  # 摘要注入 system prompt
    resp = llm.invoke([SystemMessage("".join(system_parts))] + state["messages"])
    return {"messages": [resp]}

기억 관리 설계 체크리스트

완전한 Agent 기억 시스템을 구현하기 위해 고려해야 할 점:

단기 기억 (checkpointer)

적절한 checkpointer 백엔드 선택 (개발용 MemorySaver, 생산용 SqliteSaver/PostgresSaver)
각 사용자/세션에 고유한 thread_id
할당 설정历史 자르기 임계값, 토큰 무한 증가 방지

장기 기억 (store)

네임스페이스별로 사용자 데이터를 구성:(类型, user_id) 와 ("user_facts", uid)
에서 기억을 추출할 때 신뢰도 필터링을 적용하여 의미 없는 노이즈를 저장하지 않도록 해야 합니다.
생산 환경을 PostgresStore / RedisStore 로 대체합니다.

컨텍스트 압축

압축 임계값을 결정합니다 (추천 8-20개의 메시지, 시나리오에 따라 다름).
요약 Prompt에서 보존해야 할 정보를 명확히 지정합니다 (주제 이름, 핵심 결정, 사용자 선호도).
요약 체인 테스트: N차례 압축된 요약이 1차례의 핵심 정보를 포함하고 있는지 확인합니다.
RemoveMessage 를 전체 메시지 목록을 대체하는 대신 사용합니다 (후자는 checkpointer가 있을 때 문제가 발생합니다).

기억의 읽기/쓰기 시점

기억 읽기: chat_node 시작 부분에 system prompt 주입
기억 쓰기: chat_node 끝 부분에 사용자 새 정보 추출
각 라운드마다 쓰지 않도록 (설정 추출 신뢰도 또는 내용 길이 필터링)

이 글의 요약

몇 가지 핵심 관점:

네 가지 기억 유형은 각자의 역할을 수행: 감각 기억은 LLM 호출 자체; 작업 기억은 대화 이력; 상황 기억은 압축 요약; 의미 기억은 세션 간 KV Store
checkpointer는 세션 내에서, store는 세션 간에서 관리: thread_id는 세션 차원이며, user_id는 사용자 차원이며, 두 차원은 분리하여 관리
요약 압축은 긴 대화의 핵심 솔루션입니다：RemoveMessage + 요약 주입으로 Agent가 무한한 대화에서 Token을 통제할 수 있게 합니다
세션 분리는 기초입니다：다른 thread_id의 이력은 서로 방해하지 않으며, 다른 user_id의 장기 기억은 서로 방해하지 않습니다
InMemoryStore에서 PostgresStore로 바꾸기만 하면 됩니다：아키텍처는 동일하며, 백엔드는 플러그인될 수 있습니다

다음 글：지식베이스 통합——RAG를 Agent 도구로 사용하는 것과 RAG 파이프라인의 본질적인 차이, 다수의 지식베이스를 라우팅하는 방법, Agent가 언제 검색해야 할지, 무엇을 검색해야 할지, 몇 번 검색해야 할지 결정하는 방법입니다

참고자료

제 개인 홈페이지로 방문해 주세요더 많은 유용한 지식과 재미있는 제품을 찾으실 수 있습니다

추천 피드

掘金