d01 - 5. LangGraph

S4 · LangGraph 핵심 — 상태 · 노드 · 엣지 · 체크포인터

Agentic RAG 실전 (14시간) · Day 1

검색기를 에이전트의 도구로 쓰기 위한 LangGraph를 익힌다. 이 교시는 LLM·검색 없이 순수 파이썬 노드로 메커니즘만 다룬다(키 불필요). S0 프로젝트의 .venv 커널로 실행한다.

목표

• 그래프로 에이전트를 구성하는 이유를 이해한다.
• 상태(State)·노드(Node)·엣지(Edge)를 다룬다.
• 조건부 엣지(분기)와 체크포인터(대화 기억)를 구현한다.

 

1. 왜 그래프인가

일직선 코드는 간단한 작업에 충분하지만, 에이전트에는 두 가지가 더 필요하다.

• 분기: 상황을 보고 다음 행동을 바꾼다(검색할까 직답할까).
• 반복(루프): 결과가 부실하면 앞 단계로 되돌아간다(재검색).

if/while 로 욱여넣으면 금세 복잡해진다. LangGraph는 흐름을 순서도처럼 그린다 — 작업 단계는 노드, 흐름은 엣지, 갈림길은 조건부 엣지. 분기·반복이 명시적으로 드러나고, 상태 저장·재개는 프레임워크가 처리한다.

 

2. 상태 · 노드 · 엣지

• 상태(State): 실행 중 계속 전달되는 데이터 묶음. 각 노드가 읽고 일부를 갱신한다.
• 노드(Node): 상태를 받아 '바꿀 부분'만 dict로 반환하는 함수.
• 엣지(Edge): 노드 간 순서. 일반 엣지는 고정 순서, 조건부 엣지는 라우팅 함수가 다음 노드를 결정한다. 시작/끝은 START/END.

 

3. 최소 그래프

노드 하나짜리 그래프로 흐름을 확인한다. 상태를 TypedDict 로 선언하고, 노드는 변경분만 반환하며, START → 노드 → END 로 연결한 뒤 compile()·invoke() 로 실행한다.

from typing import Annotated, Sequence, TypedDict          # 상태의 타입을 선언하기 위한 도구들
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage  # 메시지 객체들
from langgraph.graph import StateGraph, START, END           # 그래프 핵심 부품
from langgraph.graph.message import add_messages             # 메시지 '누적' 리듀서(아래 4번에서 설명)

# ① 상태의 모양 선언: messages 라는 키에 '메시지 목록'을 담는다.
#    Annotated[..., add_messages] → "이 키는 덮어쓰지 말고 '누적(append)'하라"는 표시.
class State(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]

# ② 노드 함수: 상태를 받아 '바꿀 부분'(messages)만 dict로 반환한다.
def echo_node(state: State):
    last_text = state["messages"][-1].content    # 마지막 메시지의 내용을 꺼냄
    reply = AIMessage(content=f"메아리: {last_text}")  # 응답 메시지를 만든다
    return {"messages": [reply]}                  # 변경분만 반환 → 리듀서가 기존 목록에 누적

# ③ 그래프 조립
graph = StateGraph(State)            # 위에서 선언한 상태로 그래프 생성
graph.add_node("echo", echo_node)    # "echo" 라는 이름으로 노드 등록
graph.add_edge(START, "echo")        # 시작 → echo
graph.add_edge("echo", END)          # echo → 끝

# ④ 컴파일(실행 가능한 앱으로) 후 실행
app = graph.compile()
result = app.invoke({"messages": [HumanMessage(content="안녕하세요")]})  # 입력 메시지로 실행
for m in result["messages"]:         # 최종 상태의 메시지들을 출력
    print(f"{m.type:9s}| {m.content}")

human    | 안녕하세요
ai       | 메아리: 안녕하세요

 

🔍 관찰 포인트

• 노드는 {"messages": [reply]} 변경분만 반환했는데 결과엔 입력+응답이 모두 있다 → 리듀서(add_messages)가 누적했기 때문(다음 절).
• m.type: human / ai.

 

4. 상태 심화 — TypedDict · 리듀서

TypedDict 는 상태의 키·타입을 선언한다. 노드가 어떤 키를 반환하면 기본 동작은 덮어쓰기다. 그러나 대화 메시지는 누적되어야 한다. 이 "합치는 규칙"이 리듀서다. Annotated[목록, add_messages] 처럼 지정하면 새 메시지가 기존 목록에 누적된다. 이 패턴이 흔해서 LangGraph는 MessagesState 를 기본 제공한다(다음 절·S5에서 사용).

 

5. 조건부 엣지 — 분기

라우팅 함수가 상태를 보고 '다음 노드 이름'을 반환하면 add_conditional_edges 가 그에 따라 분기한다. S5에서는 이 자리에 LLM의 판단(도구 호출 여부)이 들어가며, 메커니즘은 동일하다.

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class S(TypedDict):     # 아주 단순한 상태: 문자열 하나와 결과
    text: str
    result: str

# 두 개의 '도착지' 노드
def short_node(state: S):
    return {"result": f"짧은 입력 처리: '{state['text']}'"}   # 짧을 때 할 일
def long_node(state: S):
    return {"result": f"긴 입력 처리: '{state['text']}'"}     # 길 때 할 일

# 라우팅 함수: 상태를 보고 '다음 노드 이름'을 반환 (이게 갈림길의 핵심)
def route(state: S):
    return "long" if len(state["text"]) > 5 else "short"     # 5자 초과면 long, 아니면 short

g = StateGraph(S)
g.add_node("short", short_node)        # 도착지 노드 등록
g.add_node("long", long_node)
# 시작점에서 곧바로 조건 분기: route()의 반환값 → 실제 노드 이름으로 연결
g.add_conditional_edges(START, route, {"short": "short", "long": "long"})
g.add_edge("short", END)                # 각 도착지 → 끝
g.add_edge("long", END)
app = g.compile()

for t in ["네", "아주 긴 문장입니다"]:    # 짧은 입력 / 긴 입력으로 분기 확인
    print(app.invoke({"text": t})["result"])

짧은 입력 처리: '네'
긴 입력 처리: '아주 긴 문장입니다'

 

🔍 관찰 포인트

• route 가 반환한 문자열이 매핑({"short":"short","long":"long"})을 통해 실제 노드로 연결된다. 이 분기 구조가 모든 라우팅의 기본이다.

 

6. 체크포인터 — 대화 기억

invoke 는 끝나면 상태가 사라진다. 체크포인터를 붙이면 상태를 thread_id 별로 저장해, 같은 thread_id 로 다시 부르면 이전 상태를 이어받는다.

• InMemorySaver: 메모리 저장(런타임 종료 시 소멸) — 실습용.
• 운영은 DB 기반 체크포인터(SqliteSaver/PostgresSaver, S10).

아래는 같은 thread_id 면 메시지가 누적되고, 다른 thread_id 면 초기화됨을 보여 준다.

from langgraph.graph import StateGraph, START, END, MessagesState  # 미리 만들어진 메시지 상태
from langgraph.checkpoint.memory import InMemorySaver                # 메모리 체크포인터
from langchain_core.messages import HumanMessage, AIMessage

# MessagesState 는 위에서 우리가 직접 만든 State 와 사실상 동일(messages + add_messages 리듀서).
def bot(state: MessagesState):
    n = len(state["messages"])                          # 지금까지 쌓인 메시지 개수
    last = state["messages"][-1].content                # 마지막 사용자 메시지
    return {"messages": [AIMessage(content=f"({n}번째 메시지) '{last}' 잘 받았어요")]}

g = StateGraph(MessagesState)
g.add_node("bot", bot)
g.add_edge(START, "bot"); g.add_edge("bot", END)
app = g.compile(checkpointer=InMemorySaver())           # ★ 체크포인터를 붙여 컴파일

# thread_id 는 'configurable' 안에 넣어 전달한다 — 이 값이 대화를 구분하는 열쇠.
cfg_a = {"configurable": {"thread_id": "대화A"}}

print(app.invoke({"messages": [HumanMessage(content="처음 인사")]}, cfg_a)["messages"][-1].content)
print(app.invoke({"messages": [HumanMessage(content="두 번째")]}, cfg_a)["messages"][-1].content)
# ↑ 같은 '대화A' 라서 이전 메시지가 기억되어 개수가 늘어난다.

cfg_b = {"configurable": {"thread_id": "대화B"}}        # 다른 대화 식별자
print(app.invoke({"messages": [HumanMessage(content="새 대화 시작")]}, cfg_b)["messages"][-1].content)
# ↑ '대화B' 는 처음이라 메시지 개수가 다시 작아진다(초기화).

(1번째 메시지) '처음 인사' 잘 받았어요
(3번째 메시지) '두 번째' 잘 받았어요
(1번째 메시지) '새 대화 시작' 잘 받았어요

 

왜 (3번째 메시지)가 나올까?

• 여기서 n = len(state["messages"]) 는 사용자 메시지 개수가 아니라, 현재 상태에 쌓여 있는 전체 메시지 개수를 센다.
• 같은 thread_id="대화A" 로 첫 번째 호출을 마치면 상태에는 이미 2개가 들어 있다.
  1. 사용자: 처음 인사
  2. AI: '(1번째 메시지) 처음 인사 잘 받았어요'
• 그다음 같은 thread_id="대화A" 로 두 번째를 보내면, bot() 이 실행되는 시점의 상태는 다음 3개를 포함한다.
  1. 사용자: 처음 인사
  2. AI: 첫 번째 응답
  3. 사용자: 두 번째
• 그래서 len(state["messages"]) == 3 이 되고, 응답이 (3번째 메시지) '두 번째' 잘 받았어요 로 출력된다.
• 즉, 이것은 세 번째 사용자 질문이라는 뜻이 아니라, 현재 state 안에 들어 있는 전체 메시지 기준 세 번째 항목 시점이라는 뜻이다.
• 만약 "사용자 몇 번째 질문인지"를 세고 싶다면, 전체 길이 대신 human 타입 메시지만 세도록 코드를 바꾸면 된다.

 

🔍 관찰 포인트

• 같은 thread는 메시지 개수가 늘어난다(기억). 다른 thread는 처음부터 시작한다(격리).
• checkpointer=InMemorySaver() 한 줄이 멀티턴 기억을 만든다.

 

7. 그래프 시각화 (선택)

구조를 그림으로 확인하면 이해가 빠르다.

from IPython.display import Image, display
try:
    display(Image(app.get_graph().draw_mermaid_png()))   # 그래프를 PNG 이미지로 렌더
except Exception:
    print(app.get_graph().draw_mermaid())                # 이미지가 안 되면 텍스트(머메이드)로 출력

✏️ 미니 실습

1. 5절 route 기준을 다른 조건(예: 물음표 포함)으로 바꾼다.
2. 6절에서 같은 thread로 세 번째 메시지를 보내 누적을 확인한다.
3. 라우팅 함수 자리에 'LLM 판단'을 넣으려면 무엇을 입력받아 무엇을 반환해야 할지 생각한다 → S5에서 확인.

# 미니 실습 1. route 기준을 '물음표 포함 여부'로 바꾸기
from langgraph.graph import StateGraph, START, END
from typing import TypedDict

# 실습 1 전용 상태입니다.
# 입력 문자열(text)과 최종 결과(result)를 저장합니다.
class QuestionState(TypedDict):
    text: str
    result: str

# 물음표가 있을 때 도착할 노드입니다.
def question_node(state: QuestionState):
    return {"result": f"질문으로 판단: '{state['text']}'"}

# 물음표가 없을 때 도착할 노드입니다.
def statement_node(state: QuestionState):
    return {"result": f"일반 문장으로 판단: '{state['text']}'"}

# 라우팅 함수는 상태를 입력받아 '다음 노드 이름'을 문자열로 반환해야 합니다.
# 여기서는 물음표가 포함되어 있으면 question, 아니면 statement 로 보냅니다.
def route_by_question_mark(state: QuestionState):
    return "question" if "?" in state["text"] else "statement"

# 조건부 엣지 예제를 새 그래프로 따로 구성합니다.
practice_graph_1 = StateGraph(QuestionState)
practice_graph_1.add_node("question", question_node)
practice_graph_1.add_node("statement", statement_node)
practice_graph_1.add_conditional_edges(
    START,
    route_by_question_mark,
    {"question": "question", "statement": "statement"},
)
practice_graph_1.add_edge("question", END)
practice_graph_1.add_edge("statement", END)
practice_app_1 = practice_graph_1.compile()

# 물음표가 있는 입력과 없는 입력을 각각 넣어 분기 결과를 확인합니다.
for text in ["안녕하세요", "오늘 회의인가요?"]:
    print(practice_app_1.invoke({"text": text})["result"])

일반 문장으로 판단: '안녕하세요'
질문으로 판단: '오늘 회의인가요?'

실습 1 해설

• 이 실습은 조건부 엣지에서 라우팅 함수가 문자열 하나를 반환하면 그 값이 다음 노드 이름으로 해석된다는 점을 확인하는 예제다.
• 기존에는 "문자열 길이"로 분기했지만, 여기서는 "물음표가 있는가"라는 더 직관적인 조건으로 바꿨다.
• 핵심은 조건 자체가 무엇이냐보다, 상태를 읽고 다음 노드 이름을 반환하는 구조가 LangGraph 라우팅의 기본이라는 점이다.

 

 

 

 

# 미니 실습 2. 같은 thread 로 세 번째 메시지를 보내 누적 확인하기
# 위 6절에서 만든 cfg_a 와 app 를 그대로 재사용합니다.
# 같은 thread_id='대화A' 로 다시 보내면 이전 대화 상태를 이어받습니다.

third_result = app.invoke(
    {"messages": [HumanMessage(content="세 번째 메시지입니다")]},
    cfg_a,
)

# 전체 메시지 개수와 사용자 메시지 개수를 각각 구합니다.
all_messages = third_result["messages"]
human_messages = [message for message in all_messages if message.type == "human"]

# 마지막 AI 응답은 '전체 메시지 기준'으로 번호를 매기고 있습니다.
print("마지막 AI 응답:", all_messages[-1].content)

# 하지만 학습용으로는 '사용자 몇 번째 질문인지'를 함께 보여 주는 편이 더 직관적입니다.
print(f"현재는 사용자 {len(human_messages)}번째 질문까지 진행된 상태입니다.")

# 현재 thread 안에 어떤 메시지들이 누적되어 있는지도 함께 확인합니다.
for idx, message in enumerate(all_messages, start=1):
    print(f"{idx}. {message.type}: {message.content}")

마지막 AI 응답: (7번째 메시지) '세 번째 메시지입니다' 잘 받았어요
현재는 사용자 4번째 질문까지 진행된 상태입니다.
1. human: 처음 인사
2. ai: (1번째 메시지) '처음 인사' 잘 받았어요
3. human: 두 번째
4. ai: (3번째 메시지) '두 번째' 잘 받았어요
5. human: 세 번째 메시지입니다
6. ai: (5번째 메시지) '세 번째 메시지입니다' 잘 받았어요
7. human: 세 번째 메시지입니다
8. ai: (7번째 메시지) '세 번째 메시지입니다' 잘 받았어요

실습 2 해설

• 체크포인터는 같은 thread_id 에 대해 이전 상태를 다시 불러오므로, 새 메시지를 보낼 때마다 대화 기록이 이어진다.
• 이 예제에서 (5번째 메시지)처럼 보이는 이유는 사용자 메시지와 AI 메시지를 모두 포함한 전체 목록 길이를 기준으로 세기 때문이다.
• 반면 현재는 사용자 3번째 질문까지 진행된 상태입니다. 라는 출력은 human 타입 메시지만 센 결과라서, 사람이 느끼는 턴 수와 더 가깝다.
• 즉, 같은 상태를 보더라도 "전체 메시지 수"와 "사용자 질문 수"는 다를 수 있다는 점을 확인하는 실습이다.

 

 

 

 

# 미니 실습 3. 라우팅 함수 자리에 'LLM 판단'을 넣는다면?
# 핵심은 동일합니다.
# 1) 입력: 현재 상태(state)에서 판단에 필요한 정보(예: 마지막 사용자 메시지)를 꺼낸다.
# 2) 출력: 다음에 갈 '노드 이름' 문자열을 반환한다.
# S5에서는 이 부분의 판단을 실제 LLM 이 수행하게 됩니다.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class AgentState(TypedDict):
    question: str
    next_step: str
    result: str

# 실제 LLM 대신 규칙 기반으로 흉내 내는 라우팅 함수입니다.
# 반환값은 반드시 그래프에 등록된 다음 노드 이름이어야 합니다.
def llm_like_router(state: AgentState):
    question = state["question"]

    # 보통 LLM 에게는 '검색이 필요한가, 바로 답할 수 있는가?' 같은 판단을 시킵니다.
    # 여기서는 키워드가 있으면 search, 없으면 answer 로 보내는 간단한 대체 예시를 사용합니다.
    if "찾아" in question or "검색" in question or "규정" in question:
        return "search"
    return "answer"

# 검색이 필요하다고 판단된 경우의 도착 노드입니다.
def search_node(state: AgentState):
    return {
        "next_step": "search",
        "result": f"검색 노드로 이동: '{state['question']}'",
    }

# 바로 답변이 가능하다고 판단된 경우의 도착 노드입니다.
def answer_node(state: AgentState):
    return {
        "next_step": "answer",
        "result": f"답변 노드로 이동: '{state['question']}'",
    }

practice_graph_3 = StateGraph(AgentState)
practice_graph_3.add_node("search", search_node)
practice_graph_3.add_node("answer", answer_node)
practice_graph_3.add_conditional_edges(
    START,
    llm_like_router,
    {"search": "search", "answer": "answer"},
)
practice_graph_3.add_edge("search", END)
practice_graph_3.add_edge("answer", END)
practice_app_3 = practice_graph_3.compile()

# 두 입력을 넣어, 라우팅 함수가 무엇을 받아 무엇을 반환하는지 확인합니다.
for question in ["휴가 규정을 찾아줘", "안녕"]:
    result = practice_app_3.invoke({"question": question, "next_step": "", "result": ""})
    print(result["next_step"], "|", result["result"])

search | 검색 노드로 이동: '휴가 규정을 찾아줘'
answer | 답변 노드로 이동: '안녕'

실습 3 해설

• S5에서 라우팅 함수 자리에 실제 LLM 판단이 들어가더라도, 함수의 입출력 형식 자체는 바뀌지 않는다.
• 입력은 현재 상태(state) 안의 정보이고, 출력은 결국 다음에 갈 노드 이름 문자열이다.
• 이 예제는 LLM 대신 규칙 기반 함수로 같은 구조를 먼저 보여 준다. 즉, LangGraph에서 LLM은 "그래프 밖의 마법"이 아니라 다음 노드를 고르는 판단기로 끼워 넣는 것에 가깝다.

 

 

 

 

정리

• 그래프(노드·엣지·상태)로 분기·반복을 표현하는 LangGraph 사고방식을 익혔다.
• 리듀서로 메시지 누적, 조건부 엣지로 분기, 체크포인터로 대화 기억을 키 없이 확인했다.
• 다음(S5): 이 무대에 LLM과 검색 도구를 올려 첫 Agentic RAG를 만든다.