실무 Multi-Agent 오케스트레이션 18장 · 그래프 기반 워크플로우 6 / 6 ← 이전목차다음 → TechLead Cro

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

실습문제와 해답

그래프의 노드를 떼어 보고, 하나 더 끼워 보며 18장 내용을 손에 익힙니다. 먼저 스스로 해 본 뒤 해답을 펼쳐 보세요. 해답은 새 파일로 만들어 실행합니다.

lesson18_langgraph_basics.py에서 노드 함수와 CSState를 가져와 씁니다. 가져올 때 그 파일의 그래프 조립까지는 실행되지만, LLM을 부르는 실행 부분은 if __name__ == "__main__": 아래에 있어 돌지 않습니다.


문제 1 — 노드를 그래프 없이 혼자 실행하기

노드는 평범한 함수입니다. 새 파일 ex18_node_test.py를 만들어 그래프를 거치지 않고 노드를 직접 불러 보세요.

  • classify_node에 상태 딕셔너리를 직접 넣어, 아래 세 문의에 대해 돌려준 키와 값을 출력한다. 1. "단순 변심 반품 배송비 얼마예요?" 2. "결제가 두 번 됐어요! 빨리 확인해 주세요." 3. "AI 말고 사람이랑 통화하고 싶어요." (상담원 연결 요청)
  • log_node에 log가 ["앞 단계 기록"]인 상태를 넣어 반환값을 출력하고, 넣었던 상태의 log가 바뀌었는지도 출력한다.
해답 보기
from lesson18_langgraph_basics import classify_node, log_node

questions = [
    "단순 변심 반품 배송비 얼마예요?",
    "결제가 두 번 됐어요! 빨리 확인해 주세요.",
    "AI 말고 사람이랑 통화하고 싶어요.",
]

print("■ classify_node 단독 실행")
for q in questions:
    state = {"question": q, "intent": "", "urgency": "", "answer": "", "log": []}
    update = classify_node(state)          # 그래프 없이 함수로 바로 부른다
    print(f"  {q}\n    → 돌려준 키: {list(update.keys())}")
    print(f"    → {update['intent']} / {update['urgency']} / {update['log']}")

print("\n■ log_node 단독 실행 (LLM 호출 없음)")
state = {"question": "", "intent": "", "urgency": "", "answer": "", "log": ["앞 단계 기록"]}
print(" ", log_node(state))
print("  원래 상태의 log:", state["log"])
$ python ex18_node_test.py

실행 결과

■ classify_node 단독 실행
  단순 변심 반품 배송비 얼마예요?
    → 돌려준 키: ['intent', 'urgency', 'log']
    → 환불교환 / 낮음 / ['분류: 환불교환/낮음']
  결제가 두 번 됐어요! 빨리 확인해 주세요.
    → 돌려준 키: ['intent', 'urgency', 'log']
    → 계정결제 / 높음 / ['분류: 계정결제/높음']
  AI 말고 사람이랑 통화하고 싶어요.
    → 돌려준 키: ['intent', 'urgency', 'log']
    → 상담원연결 / 높음 / ['분류: 상담원연결/높음']

■ log_node 단독 실행 (LLM 호출 없음)
  {'log': ['앞 단계 기록', '상담 로그 저장 완료']}
  원래 상태의 log: ['앞 단계 기록']

해설 — 세 가지를 봅니다.

첫째, 그래프 없이도 노드가 돌았습니다. 노드는 딕셔너리를 받아 딕셔너리를 돌려주는 함수일 뿐입니다. 그래서 흐름 전체를 돌리지 않고 한 단계만 따로 시험할 수 있습니다. if문으로 엮은 흐름에서는 하기 어려웠던 일입니다.

둘째, 돌려준 키가 세 개뿐입니다. question과 answer는 반환값에 없습니다. "바뀐 칸만 돌려준다"는 약속을 눈으로 확인한 것입니다.

셋째, 넣었던 상태는 바뀌지 않았습니다. log_node가 돌려준 log는 두 줄인데, 원래 상태의 log는 여전히 한 줄입니다. state["log"] + [...]는 기존 리스트를 고치지 않고 새 리스트를 만듭니다. 노드는 받은 상태를 직접 고치지 않고, 바뀐 내용을 돌려주기만 합니다. 합치는 일은 LangGraph의 몫입니다.

분류 결과는 이번 실행의 값입니다. 4장에서 본 대로 분류는 흔들림이 작지만, 긴급도는 실행에 따라 달라질 수 있습니다.


문제 2 — 노드 하나를 끼워 넣기

classify와 answer 사이에 긴급도가 "높음"이면 로그에 "우선 처리 대상"을 남기는 priority_node를 끼워 넣으세요. 새 파일 ex18_priority.py에 그래프를 새로 조립합니다.

  • CSState, classify_node, answer_node, log_node는 lesson18_langgraph_basics.py에서 가져온다. 이 함수들은 고치지 않는다.
  • "단순 변심 반품 배송비 얼마예요?"와 "결제가 두 번 됐어요! 빨리 확인해 주세요."를 실행해 log를 출력한다.

실행하기 전에 짐작해 보세요. 엣지는 몇 줄을 고치고 몇 줄을 보태야 할까요?

해답 보기
from langgraph.graph import END, START, StateGraph

from lesson18_langgraph_basics import CSState, answer_node, classify_node, log_node


def priority_node(state: CSState) -> dict:
    """긴급도가 '높음'이면 로그에 표시를 남긴다. LLM 을 부르지 않는 노드다."""
    if state["urgency"] == "높음":
        return {"log": state["log"] + ["우선 처리 대상"]}
    return {}                                # 바꿀 것이 없으면 빈 dict


builder = StateGraph(CSState)
builder.add_node("classify", classify_node)
builder.add_node("priority", priority_node)   # 새 노드
builder.add_node("answer", answer_node)
builder.add_node("log", log_node)

builder.add_edge(START, "classify")
builder.add_edge("classify", "priority")      # 바뀐 엣지
builder.add_edge("priority", "answer")        # 새 엣지
builder.add_edge("answer", "log")
builder.add_edge("log", END)

graph = builder.compile()

if __name__ == "__main__":
    for q in ["단순 변심 반품 배송비 얼마예요?",
              "결제가 두 번 됐어요! 빨리 확인해 주세요."]:
        result = graph.invoke({"question": q, "intent": "", "urgency": "",
                               "answer": "", "log": []})
        print(f"\n고객: {q}")
        print(f"로그: {result['log']}")
$ python ex18_priority.py

실행 결과

고객: 단순 변심 반품 배송비 얼마예요?
로그: ['분류: 환불교환/낮음', '응답 생성 완료', '상담 로그 저장 완료']

고객: 결제가 두 번 됐어요! 빨리 확인해 주세요.
로그: ['분류: 계정결제/높음', '우선 처리 대상', '응답 생성 완료', '상담 로그 저장 완료']

해설 — 바꾼 것을 세어 봅니다.

한 일 줄 수
새 노드 함수 작성 함수 하나
add_node 추가 1줄
엣지 수정 (classify → answer를 classify → priority로) 1줄
엣지 추가 (priority → answer) 1줄
기존 노드 함수 수정 0줄

classify_node, answer_node, log_node는 한 글자도 고치지 않았습니다. 노드 함수 안에 "다음에 어디로 가라"는 말이 없기 때문입니다. 순서는 엣지에만 적혀 있으니, 순서를 바꾸는 일은 엣지만 고치면 됩니다.

그림으로 그리면 화살표가 넷에서 다섯이 됩니다. classify와 answer 사이에 상자 하나가 끼어든 모양입니다.

priority_node가 바꿀 것이 없을 때 빈 딕셔너리를 돌려준 것도 봅니다. 첫 문의의 로그에는 "우선 처리 대상"이 없고, 나머지 줄은 그대로입니다. 아무 칸도 돌려주지 않으면 상태는 바뀌지 않습니다.

다만 이 노드는 표시만 남길 뿐 긴급한 문의를 실제로 먼저 처리하지는 않습니다. 두 문의 모두 같은 길로 answer에 갔습니다. 긴급도에 따라 가는 길 자체를 달리하려면 조건에 따라 갈리는 엣지가 필요합니다. 19장의 주제입니다.


문제 3 — answer 노드를 다른 함수로 바꿔 끼우기

"answer"라는 이름에 다른 함수를 등록해 보세요. 새 파일 ex18_swap_answer.py에 그래프를 새로 조립합니다.

  • 새 노드 answer_with_source_node를 만든다. answer_node와 약속은 같다(question을 읽고 answer와 log를 돌려준다). 다른 점은 답 끝에 (근거: 제N조) 를 적게 하는 것 하나다. 정책 발췌는 lesson18_langgraph_basics.py의 POLICY_NOTE를 가져다 쓴다.
  • CSState, classify_node, log_node는 가져와서 그대로 쓴다.
  • "단순 변심 반품 배송비 얼마예요?"와 "사이즈가 안 맞아서 교환하려는데 배송비가 드나요?"를 실행해 답과 log를 출력한다.

실행하기 전에 짐작해 보세요. add_node와 add_edge 가운데 몇 줄이 바뀔까요?

해답 보기
from google.genai import types
from langgraph.graph import END, START, StateGraph

from config import MODEL
from lesson18_langgraph_basics import (POLICY_NOTE, CSState, classify_node, client,
                                       log_node)


def answer_with_source_node(state: CSState) -> dict:
    """answer_node 와 같은 약속(question 을 읽고 answer·log 를 돌려준다)을 지키는 새 노드.
    답 끝에 근거 조항을 적게 한 점만 다르다."""
    r = client.models.generate_content(
        model=MODEL,
        contents=f"[정책 발췌]\n{POLICY_NOTE}\n\n고객 문의: {state['question']}",
        config=types.GenerateContentConfig(
            system_instruction="하루마켓 상담원 '하루'. 존댓말 2문장 이내. "
                               "[정책 발췌]에 있는 내용만 말하고, "
                               "답 끝에 (근거: 제N조) 를 적는다."),
    )
    return {"answer": r.text.strip(),
            "log": state["log"] + ["응답 생성 완료 (근거 표기)"]}


builder = StateGraph(CSState)
builder.add_node("classify", classify_node)
builder.add_node("answer", answer_with_source_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__":
    for q in ["단순 변심 반품 배송비 얼마예요?",
              "사이즈가 안 맞아서 교환하려는데 배송비가 드나요?"]:
        result = graph.invoke({"question": q, "intent": "", "urgency": "",
                               "answer": "", "log": []})
        print(f"\n고객: {q}")
        print(f"답변: {result['answer']}")
        print(f"로그: {result['log']}")
$ python ex18_swap_answer.py

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

고객: 단순 변심 반품 배송비 얼마예요?
답변: 단순 변심 반품 시 편도 배송비 3,000원이 환불 금액에서 차감됩니다. (근거: 제6조)
로그: ['분류: 환불교환/낮음', '응답 생성 완료 (근거 표기)', '상담 로그 저장 완료']

고객: 사이즈가 안 맞아서 교환하려는데 배송비가 드나요?
답변: 네, 단순 변심으로 인한 교환 시 왕복 배송비 6,000원을 고객님께서 부담하셔야 합니다. (근거: 제4조)
로그: ['분류: 환불교환/보통', '응답 생성 완료 (근거 표기)', '상담 로그 저장 완료']

해설 — 바뀐 것을 세어 봅니다.

한 일 줄 수
새 노드 함수 작성 함수 하나
add_node("answer", ...)에 등록하는 함수 교체 1줄
add_edge 수정 0줄
classify_node, log_node, CSState 수정 0줄

엣지는 함수가 아니라 "answer"라는 이름을 가리킵니다. 그 이름에 등록된 함수가 바뀌어도 엣지는 그대로입니다. 새 함수가 지킬 것은 약속 하나입니다. 상태에서 question을 읽고, answer와 log를 돌려주면 됩니다.

답은 두 문의 모두 발췌의 금액(3,000원, 6,000원)에 근거 조항이 붙어 나왔고, 로그의 둘째 줄도 새 노드가 남긴 글로 바뀌었습니다.

여기서 한 걸음 더 나갈 자리가 보입니다. 발췌 두 줄로 답할 수 있는 것은 배송비 문의까지입니다. 정책 질문에는 정책 문서를 검색하는 노드가, "제 주문 어디까지 왔어요?" 같은 문의에는 주문 도구를 쓰는 노드가 맞습니다. 문의 유형에 따라 다른 노드로 보내는 갈림길, 그것이 19장의 주제입니다.

← 이전 절정리와 체크리스트19장 →왜 조건 분기가 필요한가 — 문의마다 가야 할 길이 다르다
오명운 · macro@prag-ai.com