7장 · 기억하는 에이전트 — 절 목차01. 2장에서 손으로 쌓던 대화 이제는 라이브러리에게 02. StateGraph 세 가지 상태 노드 엣지 03. 따라하기 상태를 정의하고 노드로 잇기 04. 체크포인터 한 줄이면 기억이 생긴다 05. 따라하기 CSV 분석 에이전트에 기억 붙이기 06. 따라하기 그 상품 지난달은 이어서 묻기 07. thread_id 사용자별로 대화를 갈라 두기 08. 따라하기 세션이 섞이지 않는지 검증하기 09. 정리 체크리스트

따라하기 — 상태를 정의하고 노드로 잇기

한 줄 요약

code/ch07_memory.py 를 만듭니다. 노드 하나짜리 그래프를 조립해 가장 단순한 형태의 StateGraph 를 손에 익힙니다.

1. 파일 만들기

code 폴더에 ch07_memory.py 를 만들고 아래를 전부 붙여 넣습니다.

새 파일을 만드세요 — agentic_ai/code/ch07_memory.py
VS Code 왼쪽 탐색기에서 code 폴더를 오른쪽 클릭 → New File → 파일명 ch07_memory.py 입력 → 아래를 전부 복사해 붙여 넣고 저장(Ctrl+S / Cmd+S).

# -*- coding: utf-8 -*-
"""7장 — 멀티턴 기억: LangGraph StateGraph

실행:  python code/ch07_memory.py

지금까지 만든 에이전트는 한 번 묻고 답하면 끝이었습니다.
  손님: "스마트워치 Fit 5 재고 얼마예요?"
  손님: "그거 리뷰는요?"            ← '그거'가 무엇인지 모른다

2장에서는 대화 기록을 우리가 손으로 쌓았습니다. 실제 서비스에서는 손님이 수백 명입니다.
손님마다 기록을 따로 보관하고, 다음 요청 때 꺼내 오는 일을 직접 짜면 금방 복잡해집니다.

핵심 개념 — StateGraph
  LangGraph 는 프로그램을 '상태(state)를 들고 노드 사이를 이동하는 그래프'로 봅니다.
    상태(state) : 지금까지의 대화 기록처럼, 계속 들고 다니는 값
    노드(node)  : 상태를 받아 일을 하고, 바뀐 부분을 돌려주는 함수
    엣지(edge)  : 다음에 어느 노드로 갈지
  여기에 체크포인터(checkpointer)를 붙이면 상태가 자동으로 저장됩니다.
  대화 상대는 thread_id 로 구분합니다. 손님마다 서랍이 하나씩 생기는 셈입니다.
"""
import sys
import pathlib

sys.path.append(str(pathlib.Path(__file__).resolve().parent))

from typing import Annotated, TypedDict

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

from common import get_chat
from ch05_csv_agent import run_pandas          # 5장에서 만든 pandas 도구를 그대로 재사용

llm = get_chat(temperature=0)

SYSTEM = (
    "너는 승승장구몰의 고객 상담원이다. "
    "한국어로 두 문장 안에 간결하게 답하고, 이모지는 쓰지 않는다. "
    "앞에서 손님이 말한 내용은 기억해서 이어 답한다."
)


# ── ① 상태를 정의한다 ──────────────────────────────────────────
class State(TypedDict):
    """그래프가 들고 다니는 값.

    add_messages 는 '덮어쓰지 말고 뒤에 이어 붙여라'라는 표시다.
    이것 하나로 대화 기록이 쌓인다.
    """
    messages: Annotated[list, add_messages]


# ── ② 노드를 만든다 ────────────────────────────────────────────
def counselor(state: State) -> dict:
    """상태에 든 대화 기록 전체를 모델에 넘기고, 답변을 돌려준다.

    돌려주는 것은 '바뀐 부분'뿐이다. 이어 붙이기는 add_messages 가 한다.
    """
    answer = llm.invoke([{"role": "system", "content": SYSTEM}] + state["messages"])
    return {"messages": [answer]}


# ── ③ 그래프를 조립한다 ────────────────────────────────────────
graph = StateGraph(State)
graph.add_node("counselor", counselor)
graph.add_edge(START, "counselor")     # 시작하면 상담원으로
graph.add_edge("counselor", END)       # 답하면 끝

# ── ④ 기억을 붙여 컴파일한다 ───────────────────────────────────
# InMemorySaver 는 메모리에 저장한다 → 프로그램을 끄면 사라진다.
# 실제 서비스에서는 SqliteSaver·PostgresSaver 로 바꾸기만 하면 된다.
chatbot = graph.compile(checkpointer=InMemorySaver())


def say(text: str, thread_id: str) -> str:
    """thread_id 로 대화 상대를 구분해 한 마디 주고받는다."""
    config = {"configurable": {"thread_id": thread_id}}
    result = chatbot.invoke({"messages": [{"role": "user", "content": text}]}, config)
    return result["messages"][-1].text.strip()


# ── ⑤ 5장 CSV 에이전트에 기억 붙이기 ─────────────────────────
# create_agent 도 결국 StateGraph 다. checkpointer 인자 하나로 멀티턴이 된다.
CSV_SYSTEM = (
    "너는 승승장구몰의 데이터 분석가다. "
    "숫자는 반드시 run_pandas 도구로 계산한다. "
    "손님이 '그거'·'그 상품'처럼 앞에서 말한 것을 가리키면 대화 기록에서 찾아 이어서 답한다. "
    "한국어로 간결하게 답한다."
)
analyst_with_memory = create_agent(
    llm, tools=[run_pandas], system_prompt=CSV_SYSTEM,
    checkpointer=InMemorySaver(),          # ← 이 한 줄이 전부다
)


def ask_analyst(text: str, thread_id: str) -> str:
    config = {"configurable": {"thread_id": thread_id}}
    result = analyst_with_memory.invoke(
        {"messages": [{"role": "user", "content": text}]}, config)
    return result["messages"][-1].text.strip()


if __name__ == "__main__":
    print("=" * 62)
    print("[1] 같은 thread_id — 앞말을 기억한다")
    for q in ["제 이름은 오지우입니다. 스마트워치를 샀어요.",
              "제가 뭘 샀다고 했죠?",
              "제 이름도 기억하세요?"]:
        print(f"\n  손님: {q}")
        print(f"  봇  : {say(q, thread_id='손님-오지우')}")

    print("\n" + "=" * 62)
    print("[2] thread_id 를 바꾸면 다른 손님 — 대화가 섞이지 않는다")
    q = "제가 뭘 샀다고 했죠?"
    print(f"\n  손님: {q}")
    print(f"  봇  : {say(q, thread_id='손님-박서준')}")

    print("\n" + "=" * 62)
    print("[3] 기억을 붙인 CSV 분석 에이전트 — '그거'가 통한다")
    for q in ["P0003 스마트워치 재고가 얼마나 남았나요?",
              "그거 재주문 기준은요?",
              "그럼 지금 발주해야 하나요?"]:
        print(f"\n  손님: {q}")
        print(f"  봇  : {ask_analyst(q, thread_id='재고담당-1')}")

    print("\n" + "=" * 62)
    print("관찰: 기록을 손으로 쌓지 않았습니다. thread_id 만 넘겼습니다.")
    print("      서랍은 LangGraph 가 관리합니다. 우리는 어느 서랍인지만 알려 주면 됩니다.")

2. 실행

터미널에서 실행 — 프로젝트 폴더 agentic_ai 에서, (agentic) 표시를 확인한 뒤.

python code/ch07_memory.py

약 15초 걸립니다. [1] 부터 [3] 까지 순서대로 나옵니다. 이 절에서는 [1] 만 봅니다.

==============================================================
[1] 같은 thread_id — 앞말을 기억한다

  손님: 제 이름은 오지우입니다. 스마트워치를 샀어요.
  봇  : 오지우 고객님, 스마트워치 구매 건으로 연락 주셨군요. 어떤 문의사항이 있으신지 말씀해주시면 확인해드리겠습니다.

  손님: 제가 뭘 샀다고 했죠?
  봇  : 고객님께서는 스마트워치를 구매하셨다고 말씀해주셨습니다. 해당 구매 건에 대해 궁금하신 점이 있으신가요?

  손님: 제 이름도 기억하세요?
  봇  : 네, 고객님의 성함은 오지우님이십니다. 무엇을 도와드릴까요?

2장 [2] 와 비교해 보세요.

[2장 — 기억 못 함]
  고객: 제 이름은 김민준입니다.
  상담원: 김민준 고객님, 안녕하세요.
  고객: 제 이름이 뭐라고 했죠?
  상담원: 고객님의 성함은 확인 후 안내드리겠습니다.     ← 잊었다

[7장 — 기억함]
  손님: 제가 뭘 샀다고 했죠?
  봇  : 스마트워치를 구매하셨다고 말씀해주셨습니다.       ← 기억한다

같은 모델입니다. 달라진 것은 대화 기록을 자동으로 쌓아 주는 구조뿐입니다.

3. 코드를 네 단계로 다시 보기

① 상태

아래는 code/ch07_memory.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.

class State(TypedDict):
    messages: Annotated[list, add_messages]

딕셔너리 키 하나짜리입니다. 2장의 history 가 여기 들어갑니다.

② 노드

아래는 code/ch07_memory.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.

def counselor(state: State) -> dict:
    answer = llm.invoke([{"role": "system", "content": SYSTEM}] + state["messages"])
    return {"messages": [answer]}

두 줄짜리 함수입니다.

state["messages"] — 지금까지의 대화 전체를 꺼냅니다
앞에 시스템 지시를 붙여 모델에게 보냅니다
새로 생긴 답변만 돌려줍니다

2장의 chat_multi() 와 하는 일이 같습니다.

아래는 개념 설명용 코드입니다 — 실습 파일에 넣지 않습니다.

# 2장
def chat_multi(system, history):
    resp = client.models.generate_content(
        model=..., contents=history,
        config=types.GenerateContentConfig(system_instruction=system, temperature=0))
    return resp.text.strip()

③ 조립

아래는 code/ch07_memory.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.

graph = StateGraph(State)
graph.add_node("counselor", counselor)
graph.add_edge(START, "counselor")
graph.add_edge("counselor", END)

  START ──▶ [상담원] ──▶ END

노드 이름은 아무거나 됩니다. "counselor" 대신 "chat" 이라고 해도 되고요. 나중에 그래프가 복잡해지면 이 이름이 로그와 그림에 나오므로 알아보기 쉽게 짓습니다.

④ 컴파일

아래는 code/ch07_memory.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.

chatbot = graph.compile(checkpointer=InMemorySaver())

checkpointer 가 다음 절의 주제입니다. 지금은 "이걸 붙이면 기억이 생긴다" 정도만 알아 두세요.

4. 불러 쓰기

아래는 code/ch07_memory.py 의 일부입니다 — 위에서 이미 붙여 넣었으니 다시 넣지 마세요.

def say(text: str, thread_id: str) -> str:
    config = {"configurable": {"thread_id": thread_id}}
    result = chatbot.invoke({"messages": [{"role": "user", "content": text}]}, config)
    return result["messages"][-1].text.strip()

주목할 점 두 가지.

넣는 것은 이번 말 하나뿐

아래는 개념 설명용 코드입니다 — 실습 파일에 넣지 않습니다.

{"messages": [{"role": "user", "content": text}]}

지금까지의 대화를 안 넣습니다. 이번에 손님이 한 말 하나만 넣습니다.

나머지는 체크포인터가 알아서 붙여 줍니다. 2장에서 history 를 통째로 넘기던 것과의 결정적인 차이입니다.

두 번째 인자가 설정

아래는 개념 설명용 코드입니다 — 실습 파일에 넣지 않습니다.

chatbot.invoke(inputs, config)

{"configurable": {"thread_id": ...}} 라는 형태가 낯설 텐데, LangGraph의 설정 전달 방식입니다. 7절에서 자세히 다룹니다.

5. 상태 안을 들여다보기

[1] 이 끝난 뒤의 상태를 직접 볼 수 있습니다.

직접 고쳐 보기 — agentic_ai/code/ch07_memory.py 를 열어 [1] 블록 아래에 아래를 덧붙인 뒤 다시 실행하세요.

    config = {"configurable": {"thread_id": "손님-오지우"}}
    snapshot = chatbot.get_state(config)
    print(f"\n  쌓인 메시지 {len(snapshot.values['messages'])}개:")
    for m in snapshot.values["messages"]:
        print(f"    {type(m).__name__:14s} {str(m.content)[:50]}")

  쌓인 메시지 6개:
    HumanMessage   제 이름은 오지우입니다. 스마트워치를 샀어요.
    AIMessage      오지우 고객님, 스마트워치 구매 건으로 연락 주셨군요...
    HumanMessage   제가 뭘 샀다고 했죠?
    AIMessage      고객님께서는 스마트워치를 구매하셨다고 말씀해주셨습니다...
    HumanMessage   제 이름도 기억하세요?
    AIMessage      네, 고객님의 성함은 오지우님이십니다...

2장에서 손으로 쌓았던 그 목록과 똑같습니다. 세 번 말했더니 6개가 쌓였습니다 — 손님 말 3개, 봇 답 3개.

.text 와 .content 가 섞여 나온 것을 눈치채셨나요? 위 코드는 m.content 인데 답을 꺼낼 때는 result["messages"][-1].text 였죠. .content 는 원본 그대로(문자열일 때도 있고 목록일 때도 있습니다), .text 는 항상 문자열입니다. 그래서 화면에 찍을 때는 str(m.content) 로 감쌌습니다. 이 과정에서 답을 꺼낼 때는 .text 를 씁니다.

get_state() 는 디버깅할 때 유용합니다. "대화가 이상한데 뭐가 쌓여 있지?" 를 확인하는 자리입니다.

스스로 해 보기

위의 get_state() 코드를 넣고 실행해 메시지가 어떻게 쌓이는지 보세요.
add_messages 를 빼고 실행해 보세요.
python class State(TypedDict): messages: list # Annotated 없이
어떻게 되나요? 왜 그런지 설명해 보세요.
노드에서 시스템 지시를 빼고 실행해 보세요. 답변의 말투가 어떻게 달라지나요?
노드가 이렇게 돌려주도록 바꿔 보세요.
python return {"messages": state["messages"] + [answer]}
메시지가 몇 개 쌓이나요? 예상과 다를 것입니다. 왜 그런지는 메시지의 id 를 찍어 보면 알 수 있습니다.
python print([m.id[:8] for m in snapshot.values["messages"]])

2번과 4번이 이 절의 핵심 실험입니다. add_messages 가 무슨 일을 하는지를 고장 내 보면서 확인하는 것입니다. 4번은 "중복되겠지" 하고 예상했다가 중복되지 않는 것을 보는 게 목적입니다 — 리듀서가 id 로 걸러 주기 때문입니다.

핵심 정리

네 단계 — 상태 정의 → 노드 작성 → 조립 → 컴파일. 열 줄 남짓입니다.
노드는 상태를 받아 바뀐 부분만 돌려주는 함수입니다. 2장의 chat_multi() 자리입니다.
START ──▶ [counselor] ──▶ END — 가장 단순한 그래프입니다.
부를 때는 이번 말 하나만 넣습니다. 나머지는 체크포인터가 붙여 줍니다.
2장에서는 history 를 통째로 넘겼습니다. 그 차이가 이번 장의 전부입니다.
get_state(config) 으로 쌓인 상태를 직접 볼 수 있습니다. 디버깅의 출발점입니다.
답을 꺼낼 때는 .text(항상 문자열)를 씁니다. .content 는 원본 그대로라 목록일 수도 있습니다.
같은 모델인데 결과가 다릅니다. 달라진 것은 구조입니다.