8. LangGraph 기초3 메모리

Short-term memory : MemorySaver

Memory Saver

• LangGraph에서 제공하는 체크포인트(Checkpoint) 저장소
• 그래프 실행 중 생성되는 상태(state)를 자동으로 저장하고 필요 시 복원할 수 있게 해주는 메모리 기반 저장소
• 주요 특징
  • 용도 : 그래프의 실행 상태(state)를 저장하고 복원
  • 저장 위치 : Python의 메모리(RAM)안에 저장(디스크 X)
  • 사용 방식 : LangGraph compile() 시 checkpointer로 넘김
  • 이어서 실행 : 이전 실행의 마지막 상태부터 재실행 가능
  • 한계 : 세션이 종료되면 저장된 상태는 사라짐(휘발성)

메모리 세팅 절차

• 메모리 준비

from langgraph.checkpoint.memory importMemorySaver # MemorySaver모듈 가져오기
memory = MemorySaver() # 메모리 기반 체크포인터 초기화

• 컴파일

# 체크포인터와 함께 그래프 컴파일
agent_with_memory = builder.compile(checkpointer = memory)

• 스레드 지정

# 대화 흐름을 구분하기 위한 thread_id설정
thread_1 = {"configurable": {"thread_id": "1"}}

• 실행

messages = [HumanMessage(content="3과 4를 더하라")]
messages = agent_with_memory.invoke({"messages": messages}, thread_1)
messages = [HumanMessage(content="거기에 2를 곱하라.")]
messages = agent_with_memory.invoke({"messages": messages}, thread_1)

Memory 활용

메시지를 계속 쌓을 때 생기는 문제 3가지

• 컨텍스트 초과 오류(Context Overflow)
  • GPT는 입력 길이에 한계(token limit)가 있음
  • 너무 많은 메시지가 쌓이면 오류 발생
  • 이전 대화가 잘리는 문제 생김
• 속도 저하 & 비용 증가
  • 메시지가 많아질수록 매번 LLM 호출 시 처리 시간이 느려짐
  • API 호출 비용도 계속 늘어남
• 기억이 모호해짐
  • GPT가 최근보다 과거 메시지에 주목할 수도 있음
  • 대화 흐름이 헷갈리거나 엉뚱한 답변이 나올 수 있음

이를 해결하기 위한 두 가지 방안

• 최근 메시지만 기억
  • 최근 n개만 기억하고 나머진 삭제
• 오래된 메시지 요약
  • 최근 n개만 남기고 나머진 요약해서 저장

최근 메시지만 기억

• 노드를 정의할 때 다음과 같은 간단한 함수로 구현 가능
  • GPT에게 최근 2개 메시지만 넘겨서 대화 문맥을 간결하게 유지하고, 불필요한 과거 메시지 누적을 피하는 목적의 구조
• node 정의 부분
  • filter_messages : GPT에게 최근 대화만 전달해서 중요 문맥만 반영되도록 함
  • chatbot : 최근 2개만 추출해서 GPT에게 넘김

  def filter_messages(messages: list):
  		# 최근 2개 메시지만 리턴하는 필터 함수 생성
  		return messages[-2:]
  def chatbot(state: State):
  		messages = filter_messages(state["messages"]) 
  		result = llm.invoke(messages)
  		return {"messages": [result]}
  

오래된 메시지 요약

• 메시지 6개(대화 3번)가 차면, 요약해서 기록하고 대화내용은 제거
  • 처음 6개 대화 → END
  • 7번째 입력 → summarize_conversation 실행
    • 요약 1 생성
    • 오래된 메시지 삭제
  • 또 6개 쌓이면 → 요약2 생성
    • 기존 요약(요약1) + 새 메시지로 “요약 이어 쓰기”
    • 다시 삭제
• 구현 절차 : State
  • MessagesState는 LangGraph가 기본 제공하는 메시지 기반 상태
  • 여기에 summary라는 새로운 필드를 추가

  class State(MessagesState):
  						summary: str
  

  • state 구조

  {
  	"messages": [...],  # 전체 대화 기록
  	"summary": "지금까지 요약된 내용"
  }
  

• 구현 절차 : 노드와 분기함수
  • 대화 노드
    • 기존 메시지 리스트로 GPT를 호출해서 응답을 받고, 그 응답을 메시지 리스트에 누적 시킬 준비를 해 줌

    def call_model(state: State):
    	summary = state.get("summary", "")
    	if summary:
    	# 요약을 system prompt로 넣어 대화 맥락 유지
    	    system_message= f"이전 대화 요약: {summary}"
    			messages = [SystemMessage(content=system_message)] + state["messages"]
    	else:
    			messages = state["messages"]
    	response = llm.invoke(messages)
    	return{"messages": [response]}
    

  • 분기 조건
    • 메시지가 6개 초과하면 summarize_conversation 노드로 분기
    • 그렇지 않으면 바로 END

    def should_continue(state: State) -> Literal["summarize_conversation", END]:
    		if len(state["messages"]) > 6:
    				return "summarize_conversation"
    		return END
    

  • 요약 노드
    • 지금까지 대화 내용으로 요약 생성
    • 메시지가 너무 많아지지 않도록 앞쪽 메시지 삭제
    • 요약은 state[’summary’]에 저장됨(누적 요약)

    def summarize_conversation(state: State):
    summary = state.get("summary", "") # 이전에 요약된 게 있다면, 불러오고, 없으면 ""
      if summary:                        # 이전 요약이 있다면
        summary_message = (
    f“지금까지 요약된 내용 : {summary}\\n\\n"
    "여기에 새 메시지를 반영해서 요약을 이어서 해주세요. :“ )
    else:
        summary_message = "대화 내용을 요약해주세요. :"
      # 기존 메시지 목록 끝에 요약 요청 프롬프트를 HumanMessage로 추가
      messages = state["messages"] + [HumanMessage(content=summary_message)]
      response = llm.invoke(messages)
      
      # 오래된 메시지를 제거 → 최신 2개만 유지
      delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
    return {"summary": response.content, "messages": delete_messages}
    

• 구현 절차 : 그래프 정의

# 그래프 초기화
workflow = StateGraph(State)
# 노드 추가
workflow.add_node(＂conversation＂, call_model)
workflow.add_node(＂summarize_conversation＂, summarize_conversation)
# 엣지 연결
workflow.add_edge(START, "conversation")
workflow.add_conditional_edges("conversation", should_continue)
workflow.add_edge("summarize_conversation", END)
# 메모리 포함하고, 컴파일
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

[참조] Long-term Memory : SqliteSaver

LangGraph가 지원하는 영구 저장소들

• FileSaver : 상태를 로컬 파일(JSON)로 저장
• SQLiteSaver : SQLite DB에 저장
• PostgresSaver : PostgreSQL DB에 저장
• WeaviateSaver : 벡터 DB 기반 저장
• RedisSaver : 빠른 캐시성 저장소
• CustomSaver : 직접 구현한 저장소

메모리 세팅 절차

• 다음의 절차로 SqliteSaver를 통해 DB에 저장하고 사용
  • 설치 및 라이브러리 로딩

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

  • DB 생성 및 연결

  # check_same_thread=False -> 멀티 스레드(여러 개 동시 연결) 허용
  sqlite3_conn = sqlite3.connect('checkpoints.sqlite’, check_same_thread=False)
  sqlite3_memory_checkpoint = SqliteSaver(sqlite3_conn)
  

  • 메모리 지정 컴파일

  graph = builder.compile(checkpointer=sqlite3_memory_checkpoint)
  

  • 실행

  conf_1 = {“configurable”: {“thread_id”: 1}}
  graph.invoke({“messages”: [HumanMessage(“안녕?”)]}, config=conf_1)
  

Sqlite DB는 3개의 파일로 구성됩니다

• checkpoints.sqlite
  • 메인 데이터베이스 파일
  • 모든 데이터의 기본 저장소
• checkpoints.sqlite-wal
  • 쓰기 로그(Write-Ahead Log)
  • 최신 변경사항이 일시적으로 저장됨
• checkpoints.sqlite-shm
  • 공유 메모리 파일 (Shared Memory)
• Memory에 대한 활용은 MemorySaver와 동일

https://github.com/427paul/AI_Agent/blob/main/AI_Agent_Basic/09.LangGraph%EA%B8%B0%EB%B0%98%20Agent%EA%B8%B0%EC%B4%883%20-%20%EB%A9%94%EB%AA%A8%EB%A6%AC.ipynb

AI_Agent/AI_Agent_Basic/09.LangGraph기반 Agent기초3 - 메모리.ipynb at main · 427paul/AI_Agent