랭그래프(LangGraph) Agent에 대화 기억(Memory) 저장 및 관리 구현(Feat. PostgreSQL)

포스팅 개요

AI Agent를 구현하기 위해서는 기억(Memory) 기능이 필요합니다. 주로 랭그래프(LangGraph) 예제를 보면 InMemorySaver를 사용하기도 하는데요.

기존의 InMemorySaver는 프로그램이 종료되면 대화 기록이 모두 사라지는 한계가 있었습니다. 하지만 실제 서비스에서는 사용자와의 대화 내역을 영구적으로 보존하고, 언제든지 이전 대화를 이어갈 수 있어야 합니다.

이번 포스팅에서는 관계형 데이터베이스(RDB)인 PostgreSQL를 활용해서 영구적인 메모리 관리를 구현하는 방법을 알아봅니다. 

 

본 포스팅을 작성하면서 참고한 자료는 다음과 같습니다.

 

• https://langchain-ai.github.io/langgraph/concepts/memory/

Overview

---

포스팅 본문

지난 포스팅(https://lsjsj92.tistory.com/697)에서는 LangGraph의 도구 사용, 조건부 엣지, Human-in-the-Loop 기능을 통해 간단한 에이전트를 만드는 방법을 알아보았습니다. 하지만 실제 프로덕션 환경에서는 메모리 영속성이 필요하죠. 

보통 LangGraph 예제를 보면 주로 InMemorySaver를 사용하기 때문에 프로그램이 종료되면 모든 대화 내용이 사라지게 됩니다.

그렇기에 이번 포스팅에서는 이러한 한계를 극복하기 위해서 PostgreSQL를 활용해 실제 데이터베이스에 대화 상태를 저장하고 필요에 따라 불러올 수 있도록 합니다.

---

LangGraph의 기억(Memory) 구현의 핵심 기능 3가지

오늘 구현할 에이전트는 다음 세 가지 핵심 기능을 갖추고 있습니다.

 

영구적인 메모리 관리: PostgreSQL을 활용하여 대화 기록을 데이터베이스에 영구 저장합니다. 프로그램을 다시 실행해도 이전 대화를 정확히 불러올 수 있습니다.

비동기 처리: asyncio를 활용하여 동시에 여러 사용자의 요청을 처리할 수 있으며, I/O 작업 중에도 다른 작업을 병렬로 수행할 수 있습니다.

스트리밍 응답: 사용자는 LLM의 완전한 응답을 기다리지 않고도 실시간으로 응답을 확인할 수 있어 더 자연스러운 대화 경험을 제공합니다.

---

비동기 LangGraph 에이전트 구현하기

1. 데이터베이스 설정과 연결 구성

먼저 PostgreSQL 데이터베이스와의 연결을 설정합니다. 실제 환경에서는 환경변수나 설정 파일을 사용하는 것이 보안상 좋습니다.

만약, 여러 분들이 PostgreSQL를 사용하지 않으신다면, 다른 RDBMS를 사용하시면 됩니다.

import psycopg_pool
import psycopg

DB_CONFIG = {
    "user": "leesoojin",
    "password": "",
    "host": "localhost",
    "port": "5432",
    "dbname": "langgraph_study_tistory",
}

DATABASE_URL = (
    f"postgresql://{DB_CONFIG['user']}:{DB_CONFIG['password']}"
    f"@{DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['dbname']}"
)

2. vLLM 서버와의 연결 설정

외부 LLM 서버(vLLM 등)와 연결하여 더 강력한 모델을 활용합니다. 저는 vLLM을 활용하였습니다.

여러 분들이 사용하는 LLM 서버(Ollama, OpenAI 등)를 사용하시면 됩니다.

from langchain_openai import ChatOpenAI

NGROK_URL = "https://c1cbcd1b9597.ngrok-free.app/v1" 
MODEL_NAME = "Qwen/Qwen3-14B"

llm = ChatOpenAI(
    model=MODEL_NAME,
    openai_api_key="EMPTY",
    openai_api_base=NGROK_URL,
    temperature=0.2,
    max_tokens=512,
    top_p=0.5,
    model_kwargs={
        "presence_penalty": 0.5,
        "extra_body": {
            "top_k": 10,
            "chat_template_kwargs": {"enable_thinking": False},
        },
    }
)

3. 도구(Tool) 정의와 에이전트 노드 구성

이전 포스팅과 유사하게 도구를 정의합니다. 도구는 fake_web_search로서 실제 검색을 수행하기보다, 검색을 수행한 것처럼 수행하는 함수입니다. 예제와 빠른 이해를 위해서 이렇게 사용헀지만, 실제로 검색과 관련된 로직을 해당 함수에 넣으시면 됩니다.

여러 분들이 원하시는 도구(Tools)을 추가하셔도 됩니다.

from langchain_core.tools import tool

@tool
def fake_web_search(query: str) -> str:
    """가상의 웹 검색"""
    print(f"\n[FakeSearch] '{query}' 검색 수행")
    return f"'{query}'에 대한 최신 검색 결과(가상)"

tools = [fake_web_search]
llm_with_tools = llm.bind_tools(tools)

def chatbot_node(state: AgentState):
    print("\n--- 챗봇 노드 진입 ---")
    try:
        msgs = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
        print("--- LLM 호출 시작... ---")
        response = llm_with_tools.invoke(msgs)
        print(f"--- LLM 호출 성공 ---")
        return {"messages": [response]}
    except Exception as e:
        print(f"\n--- 챗봇 노드에서 오류 발생: {e} ---")
        error_message = AIMessage(content=f"죄송합니다, 모델 응답을 가져오는 중 오류가 발생했습니다: {e}")
        return {"messages": [error_message]}

 

그리고 chatbot_node 함수를 두어서 agent가 llm을 호출하고 LLM이 응답한 메세지를 return할 수 있도록 합니다.

llm.bind_tools를 활용해서 fake_web_search를 연결하여 사용할 수 있도록 합니다.

4. PostgreSQL 기반 메모리 관리

이번 포스팅에서의 핵심인 부분입니다. 바로 PostgreSQL을 활용한 영구 메모리 저장인데요.

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
import uuid

async def main() -> None:
	# AsyncPostgresSaver를 사용하여 DB에 비동기적으로 연결합니다.
    async with AsyncPostgresSaver.from_conn_string(DATABASE_URL) as memory:
        # 테이블 생성/확인
        await memory.setup()
        # 그래프 컴파일 시 checkpointer로 DB 연결 객체를 지정합니다.
        graph = builder.compile(checkpointer=memory)
        
        print("챗봇 시작 (exit/quit 입력 시 종료)")
        
        # 사용자에게 Thread ID를 입력받습니다.
        #    - 기존 ID 입력 시 -> 대화 복원
        #    - 그냥 Enter 입력 시 -> 새로운 ID 생성 및 새 대화 시작
        tid = input("Thread ID(새 대화는 Enter): ").strip() or str(uuid.uuid4())
        cfg = {"configurable": {"thread_id": tid}}
        
        # 이전 대화 기록 불러오기
        try:
        	# DB에서 해당 Thread ID의 과거 대화 기록을 가져옵니다.
            past = await graph.aget_state(cfg)
            if past and past.values.get("messages"):
                print("\n── 이전 대화 ──")
                for m in past.values["messages"]:
                    m.pretty_print()
                print("──────────────\n")
            else:
                print("\n새로운 대화를 시작합니다.\n")
        except Exception:
            print("\n저장된 대화가 없습니다. 새로운 대화를 시작합니다.\n")

 

AsyncPostgresSaver: 이전에 사용했던 InMemorySaver 대신 PostgreSQL 기반의 비동기 체크포인터를 사용합니다. 이를 통해 대화 기록이 데이터베이스에 영구적으로 저장됩니다.

await memory.setup(): 데이터베이스에 필요한 테이블들을 자동으로 생성하거나 확인합니다.

Thread ID 관리: 각 대화 세션을 고유하게 식별하기 위한 Thread ID를 관리합니다. 사용자가 이전 대화를 이어가고 싶다면 같은 Thread ID를 입력하면 됩니다.

5. 사용자 대화 저장과 실시간 스트리밍

사용자 경험에서 중요한 부분은 실시간 스트리밍입니다. 한 번에 답을 제공하는 것도 좋지만, 타이핑하듯 답변을 제공하는 것이 더 자연스러우니까요. 

while True:
    user_input = await asyncio.to_thread(input, "나: ")
    if user_input.lower() in {"exit", "quit"}:
        break
    if not user_input.strip():
        continue

    print("AI: ", end="", flush=True)
    # 사용자와의 모든 상호작용(입력, AI 응답, 도구 사용 등)은
    # checkpointer를 통해 DB에 자동으로 저장됩니다.
    events = graph.astream_events(
        {"messages": [HumanMessage(content=user_input)]}, 
        config=cfg, 
        version="v1"
    )
    
    final_content_printed = False
    async for e in events:
        kind = e["event"]
        if kind == "on_chat_model_stream":
            chunk = e["data"]["chunk"].content
            if chunk:
                print(chunk, end="", flush=True)
                final_content_printed = True
        elif kind == "on_tool_end":
            print(f"\n… '{e['name']}' 도구 사용 완료", flush=True)
        elif kind == "on_chain_end":
            if e["name"] == "chatbot" and not final_content_printed:
                output = e.get("data", {}).get("output", {})
                if messages := output.get("messages"):
                    print(messages[-1].content, end="", flush=True)
    print()  # 줄바꿈

 

astream_events(): 이 함수는 langgraph 워크플로우를 실행하고, 내부에서 발생하는 이벤트를 수행해줍니다. 즉, 비동기 이벤트 스트림을 사용하여 LLM의 응답을 실시간으로 받아올 수 있죠. 

 

이벤트 기반 처리:

• on_chat_model_stream: LLM이 토큰을 하나씩 생성할 때마다 발생하는 이벤트입니다. 실시간 타이핑 효과를 나타낼 수 있죠.
• on_tool_end: 도구 실행이 완료되었을 때 발생하는 이벤트이며, fake_web_search와 같은 도구를 호출하고 완료되었을 때 발생하는 것입니다.
• on_chain_end: 전체 체인이 완료되었을 때 발생하는 이벤트이며, 노드의 실행이 완전히 끝났을 때 발생합니다. 

---

실행 결과

 

위 사진은 처음 대화를 시작할 때입니다. 

저는 Thread ID를 입력하지 않고, 단순히 엔터를 눌러 시작했습니다. 그러면, 새로운 대화를 시작한다는 메세지와 함께 대화를 시작하게 되는데요. 저는 일상적인 제 소개와("제 이름은 이수진입니다.")와 langgraph를 공부하고 있다고 메세지를 날린 후 종료했습니다.

 

InMemorySaver를 사용했다면, 이렇게 프로그램을 종료하면 대화가 기억(memory)되지 않고(저장되지 않고) 휘발되는데요.

아래와 같이 실행하면 대화가 기억되면서 후속 대화가 가능하게 됩니다.

 

실행할 때 Thread ID를 입력하면 기존 대화를 가져오게 되는 것이죠.

이걸 웹 UI로 표현하면, 각 채팅창에 저런 Thread ID를 저장하도록 하고 가지고 오면 마치 ChatGPT, Gemini, Claude처럼 기존 대화를 가져올 수 있게 되는 것입니다.

---

마무리

이번 포스팅에서는 LangGraph를 활용하여 PostgreSQL 기반의 기억 장치(메모리 관리)를 구현하는 방법을 알아보았습니다.

InMemorySaver의 한계를 극복하고, 실제 서비스에 적용할 수 있는 수준의 안정적이고 확장 가능한 에이전트를 구축할 수 있게 되었습니다. 특히 Thread ID를 통한 대화 세션 관리와 실시간 스트리밍 응답은 사용자 경험을 크게 향상시키는 요소입니다.

도움이 되시길 바랍니다.