에이전트 메모리 (Agent Memory)
에이전트 메모리는 대화 이력과 학습된 정보를 관리하는 시스템입니다. 단기 메모리(현재 대화)와 장기 메모리(대화 간 공유)로 구분됩니다.
메모리 유형
| 유형 | 범위 | 저장 방식 | 예시 |
|---|
| 단기 메모리 | 단일 Thread | 체크포인터 (messages) | 현재 대화 이력 |
| 장기 메모리 | Cross-thread | Store API | 사용자 선호도, 학습 정보 |
단기 메모리 (Thread 내)
MessagesState
대화 이력은 MessagesState를 통해 자동으로 관리됩니다.
from langgraph.graph import MessagesState
# MessagesState는 다음과 동일:
# class MessagesState(TypedDict):
# messages: Annotated[list[BaseMessage], add_messages]
graph = StateGraph(MessagesState)
메시지 관리 최적화
대화가 길어지면 메시지 수가 증가하여 토큰 비용이 높아집니다.
메시지 요약
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage, RemoveMessage
llm = init_chat_model("gpt-4o-mini", temperature=0)
def summarize_messages(state: MessagesState) -> MessagesState:
"""메시지가 많아지면 요약으로 대체"""
messages = state["messages"]
if len(messages) <= 10:
return state # 메시지 적으면 그대로
# 오래된 메시지 요약
old_messages = messages[:-4] # 최근 4개 제외
summary = llm.invoke([
SystemMessage(content="다음 대화 내용을 핵심만 2-3문장으로 요약하세요."),
*old_messages,
])
# 오래된 메시지 삭제 + 요약으로 대체
delete_messages = [RemoveMessage(id=m.id) for m in old_messages]
summary_message = SystemMessage(content=f"이전 대화 요약: {summary.content}")
return {"messages": delete_messages + [summary_message]}
메시지 필터링
def filter_messages(state: MessagesState) -> MessagesState:
"""최근 N개의 메시지만 유지"""
messages = state["messages"]
if len(messages) > 20:
# 시스템 메시지 + 최근 10개만 유지
system_msgs = [m for m in messages if isinstance(m, SystemMessage)]
recent_msgs = messages[-10:]
return {"messages": system_msgs + recent_msgs}
return state
장기 메모리 (Cross-thread)
Store API
Store API는 thread와 독립적으로 정보를 저장하고 조회합니다.
from langgraph.store.memory import InMemoryStore
# Store 생성
store = InMemoryStore()
# 그래프에 Store 연결
app = graph.compile(
checkpointer=checkpointer,
store=store,
)
사용자 선호도 저장
from langgraph.store.base import BaseStore
def agent_with_memory(state: MessagesState, *, store: BaseStore) -> MessagesState:
"""Store에서 사용자 정보를 읽고 저장하는 에이전트"""
user_id = "user-123"
namespace = ("preferences", user_id)
# 저장된 선호도 읽기
preferences = store.search(namespace)
pref_text = "\n".join(
f"- {item.value['key']}: {item.value['content']}"
for item in preferences
)
# 선호도를 포함하여 응답 생성
messages = state["messages"]
if pref_text:
messages = [
SystemMessage(content=f"사용자 선호도:\n{pref_text}"),
] + messages
response = llm.invoke(messages)
# 새로운 선호도가 감지되면 저장
if "좋아" in messages[-1].content or "선호" in messages[-1].content:
store.put(
namespace,
key="latest_preference",
value={"key": "선호도", "content": messages[-1].content},
)
return {"messages": [response]}
메모리 패턴 비교
| 패턴 | 범위 | 비용 | 적합한 경우 |
|---|
| 전체 이력 유지 | Thread | 높음 | 짧은 대화 |
| 메시지 요약 | Thread | 중간 | 긴 대화 |
| 최근 N개만 유지 | Thread | 낮음 | 독립적 질문 |
| Store (장기 메모리) | Cross-thread | 낮음 | 사용자 프로필, 선호도 |
시작 추천: 체크포인터로 단기 메모리를 설정하고, 대화가 길어지면 메시지 요약을 추가하세요. 사용자별 맞춤화가 필요하면 Store API로 장기 메모리를 구현합니다.