18장. 그래프 기반 워크플로우

따라하기 — 첫 상태 그래프

목표

상담 흐름을 분류 → 응답 → 기록 세 단계의 직선 그래프로 만들어 실행합니다. 그래프의 구조를 글로 뽑아 보고, 상태가 세 노드를 지나며 채워지는 것을 log로 확인합니다. 그리고 답의 내용이 어느 노드에서, 무엇을 근거로 만들어졌는지 짚어 봅니다.

---

0. 실습 준비

VS Code에서 haru-market 폴더를 열고 터미널에서 환경을 켭니다.

$ conda activate myenv

이번 장의 코드는 4장에서 만든 분류기를 불러 씁니다. 코드 안의 이 줄입니다.

from lesson04_intent_classifier import classify   # 4장 분류기 재사용

그래서 lesson04_intent_classifier.py가 폴더 맨 위에 있어야 합니다.

$ ls lesson04_intent_classifier.py

파일 이름이 그대로 출력되면 준비된 것입니다. 없다고 나오면 4장으로 돌아가 먼저 만듭니다. 4장 파일의 실행 부분은 if __name__ == "__main__": 아래에 있어서, 불러 쓸 때 4장의 분류 실험이 다시 돌지는 않습니다.

LangGraph는 2장에서 pip install -r requirements.txt를 할 때 함께 설치되었습니다. 확인하려면 아래 명령을 실행합니다. 버전 번호가 나오면 됩니다.

$ python -c "import importlib.metadata as m; print(m.version('langgraph'))"

1.2.9

숫자는 설치한 때에 따라 다를 수 있습니다. 1.0 이상이면 됩니다.

---

1. 파일 만들기

haru-market 폴더 맨 위에 새 파일을 만듭니다.

lesson18_langgraph_basics.py

아래 코드 전체를 복사해 붙여 넣고 저장합니다.

# -*- coding: utf-8 -*-
"""[18장] LangGraph 기초 — 상담 흐름을 상태 그래프로

문제 상황: 분류하고, 조회하고, 예외 처리하고… if문이 뒤엉키기 시작했다.
LangGraph 는 흐름을 '그래프'로 선언한다.
  - State: 흐름 전체가 공유하는 데이터 (딕셔너리 스키마)
  - Node : 상태를 받아 상태 일부를 갱신해 반환하는 함수
  - Edge : 노드 사이의 이동 경로

오늘은 가장 단순한 직선 그래프로 상담 흐름을 옮긴다.
  분류 → 응답 생성 → 로그 기록

실행:  python lesson18_langgraph_basics.py
"""
from typing import TypedDict

from langgraph.graph import END, START, StateGraph

from config import MODEL, get_client
from lesson04_intent_classifier import classify   # 4장 분류기 재사용

client = get_client()

# 응답 노드가 근거로 삼을 정책 발췌 (반품·교환·환불 정책 제4조·제6조).
# 이번 장은 흐름의 구조를 보는 장이라 발췌를 직접 적어 둔다. 문서 검색은 19장에서 붙인다.
POLICY_NOTE = """- 단순 변심 반품: 편도 배송비 3,000원을 환불 금액에서 차감한다 (제6조)
- 단순 변심 교환: 왕복 배송비 6,000원을 고객이 부담한다 (제4조)"""


# ── 1. State 정의 — 이 그래프가 흐르는 동안 공유되는 데이터 ───────────
class CSState(TypedDict):
    question: str        # 고객 문의 (입력)
    intent: str          # 분류 결과
    urgency: str
    answer: str          # 최종 답변 (출력)
    log: list[str]       # 처리 과정 기록


# ── 2. Node 정의 — 상태를 받아 '바뀐 부분만' 반환한다 ─────────────────
def classify_node(state: CSState) -> dict:
    """4장 분류기를 그래프 노드로 감쌌다."""
    result = classify(state["question"])
    return {
        "intent": result.intent.value,
        "urgency": result.urgency.value,
        "log": state["log"] + [f"분류: {result.intent.value}/{result.urgency.value}"],
    }


def answer_node(state: CSState) -> dict:
    from google.genai import types
    r = client.models.generate_content(
        model=MODEL,
        contents=f"[정책 발췌]\n{POLICY_NOTE}\n\n"
                 f"[문의 유형: {state['intent']}]\n고객 문의: {state['question']}",
        config=types.GenerateContentConfig(
            system_instruction="하루마켓 상담원 '하루'. 존댓말 3문장 이내로 안내. "
                               "금액·기간은 [정책 발췌]에 있는 것만 말하고, "
                               "없는 내용은 확인 후 안내드린다고 답한다.",
            temperature=1.0),
    )
    return {"answer": r.text.strip(),
            "log": state["log"] + ["응답 생성 완료"]}


def log_node(state: CSState) -> dict:
    return {"log": state["log"] + ["상담 로그 저장 완료"]}


# ── 3. 그래프 조립: Node 등록 → Edge 연결 → 컴파일 ────────────────────
builder = StateGraph(CSState)
builder.add_node("classify", classify_node)
builder.add_node("answer", answer_node)
builder.add_node("log", log_node)

builder.add_edge(START, "classify")   # 시작 → 분류
builder.add_edge("classify", "answer")
builder.add_edge("answer", "log")
builder.add_edge("log", END)          # 로그 → 종료

graph = builder.compile()

if __name__ == "__main__":
    print("■ 실행")
    result = graph.invoke({
        "question": "단순 변심 반품 배송비 얼마예요?",
        "intent": "", "urgency": "", "answer": "", "log": [],
    })
    print(f"\n의도: {result['intent']} / 긴급도: {result['urgency']}")
    print(f"답변: {result['answer']}")
    print(f"로그: {result['log']}")

    print("""
──────────────────────────────────────────────────────────
if문 대신 그래프로 얻는 것
  - 흐름이 코드가 아니라 '구조'로 보인다 (노드와 엣지를 그림으로 그릴 수 있다)
  - 노드 단위로 테스트·교체 가능
  - 체크포인트를 붙이면 중단·재개 가능 (23장 HITL 의 기반)
지금은 직선 흐름뿐 — 문의 유형별로 갈라지려면? → 19장 조건 분기
──────────────────────────────────────────────────────────""")

---

2. 코드에서 볼 곳

파일은 주석의 번호대로 세 부분입니다.

세 노드를 나란히 놓고 봅니다.

answer_node를 눈여겨봅니다.

def answer_node(state: CSState) -> dict:
    from google.genai import types
    r = client.models.generate_content(
        model=MODEL,
        contents=f"[정책 발췌]\n{POLICY_NOTE}\n\n"
                 f"[문의 유형: {state['intent']}]\n고객 문의: {state['question']}",
        config=types.GenerateContentConfig(
            system_instruction="하루마켓 상담원 '하루'. 존댓말 3문장 이내로 안내. "
                               "금액·기간은 [정책 발췌]에 있는 것만 말하고, "
                               "없는 내용은 확인 후 안내드린다고 답한다.",
            temperature=1.0),
    )
    return {"answer": r.text.strip(),
            "log": state["log"] + ["응답 생성 완료"]}

앞 노드가 채운 intent를 읽어 문의 앞에 붙입니다. 앞 단계의 결과를 뒤 단계가 상태에서 꺼내 쓰는 것, 이것이 상태를 함께 본다는 말의 뜻입니다.

이 노드에는 도구도 정책 검색도 없습니다. 이번 장은 흐름의 구조를 보는 장이라 응답 단계를 가장 단순하게 두었습니다. 대신 답의 근거로 삼을 정책 발췌 두 줄(POLICY_NOTE) 을 파일 위쪽에 적어 두고 문의 앞에 붙입니다.

POLICY_NOTE = """- 단순 변심 반품: 편도 배송비 3,000원을 환불 금액에서 차감한다 (제6조)
- 단순 변심 교환: 왕복 배송비 6,000원을 고객이 부담한다 (제4조)"""

발췌를 손으로 적는 것은 이번 장까지입니다. 19장에서 이 자리를 정책 문서 검색으로 바꿉니다.

---

3. 실행하기

실행하기 전에 짐작해 보세요.

• 문의는 "단순 변심 반품 배송비 얼마예요?"입니다. answer 노드는 무엇을 보고 금액을 답할까요?
• 결과의 log에는 몇 줄이, 어떤 순서로 들어 있을까요?

$ python lesson18_langgraph_basics.py

LLM을 두 번 호출하므로 5~10초 걸립니다.

실행 결과 (답변 문장은 실행할 때마다 달라집니다)

■ 실행

의도: 환불교환 / 긴급도: 낮음
답변: 안녕하세요, 하루마켓 상담원 '하루'입니다.
단순 변심 반품 시 편도 배송비 3,000원이 환불 금액에서 차감됩니다.
다른 궁금한 점이 있으시면 언제든 편하게 말씀해 주세요.
로그: ['분류: 환불교환/낮음', '응답 생성 완료', '상담 로그 저장 완료']
(이하 생략)

---

4. 무엇을 관찰했나

그래프의 구조

방금 실행한 그래프를 그림으로 그리면 이렇습니다.

화살표 넷이 우리가 적은 add_edge 네 줄입니다.

이 구조는 compile()을 하는 순간, 모델을 한 번도 부르기 전에 정해집니다. 흐름은 실행하기 전에 이미 정해져 있습니다.

상태가 세 노드를 지나며 채워졌다

로그: ['분류: 환불교환/낮음', '응답 생성 완료', '상담 로그 저장 완료']

log는 빈 리스트로 시작해 노드를 하나 지날 때마다 한 줄씩 늘었습니다. 줄의 순서가 곧 노드가 실행된 순서입니다. intent와 urgency는 첫 노드가, answer는 둘째 노드가 채웠고, 우리는 마지막에 result에서 꺼내 출력했습니다.

• 문의를 "환불교환"으로 가려낸 것, 답 문장을 쓴 것은 모델입니다.
• 분류 다음에 응답, 응답 다음에 기록이라는 순서를 정한 것은 개발자이고, 그 순서대로 노드를 부른 것은 LangGraph입니다.

답의 금액은 answer 노드가 발췌에서 가져왔다

이번 실행의 답은 "편도 배송비 3,000원이 환불 금액에서 차감됩니다"입니다. POLICY_NOTE의 첫 줄 그대로입니다. 이 교재를 준비하며 세 번 실행했을 때 세 번 모두 3,000원으로 답했습니다.

각 노드가 한 일을 나눠 봅니다.

답의 내용은 answer라는 노드 하나가 정합니다. 그래서 답을 만드는 방식을 바꾸고 싶으면 answer라는 이름에 다른 함수를 등록하면 됩니다. 발췌를 손으로 적는 대신 정책 문서를 검색해 답하는 함수로 바꿔도, 다른 노드와 엣지는 한 줄도 바뀌지 않습니다. 실습문제 3에서 이 교체를 직접 해 봅니다.

그래프는 어느 단계가 무엇을 맡는지를 구조로 드러냅니다. 바꿀 곳이 노드 하나로 좁혀집니다.

if문 방식이었다면 "답을 만드는 방식을 바꾸자"에서 출발해 함수 전체를 읽으며 어느 갈래에서 답이 만들어지는지부터 찾아야 했을 것입니다.

긴급도는 채워졌지만 아무도 쓰지 않았다

urgency는 "낮음"으로 채워져 출력까지 되었지만, 어느 노드도 이 값을 읽지 않습니다. 긴급한 문의라고 흐름이 달라지지도 않습니다. 지금은 경로가 하나뿐이기 때문입니다. 상태의 값을 보고 경로를 나누는 일은 19장에서 합니다.

---

5. 지금 폴더의 모습

haru-market/
├── config.py
├── haru_tools.py
├── (lesson02 ~ lesson03 생략)
├── lesson04_intent_classifier.py  (이번 장이 불러 쓴다)
├── (lesson05 ~ lesson16 생략)
├── lesson17_memory.py
├── lesson18_langgraph_basics.py   ← 이번 장
└── memory_store/

---

핵심 정리

• 그래프는 CSState 선언 → 노드 세 개 → 엣지 네 줄 → compile() 로 만들었습니다.
• 그림의 화살표 넷이 add_edge 네 줄과 하나씩 맞아떨어집니다.
• log의 세 줄이 노드가 실행된 순서를 그대로 보여 줍니다.
• 답의 금액(3,000원)은 answer 노드가 정책 발췌에서 가져왔습니다. 답을 만드는 방식을 바꾸려면 그 노드 하나를 교체합니다.
• 이번 장의 산출물은 상담 흐름의 첫 상태 그래프입니다.