[LangGraph] 기초부터 멀티 에이전트까지: 실전 개발 로드맵 총정리

LangGraph는 최근 LLM 애플리케이션 개발에서 가장 중요한 프레임워크로 자리 잡고 있습니다. 복잡해 보일 수 있는 개념들을 이해하기 쉬운 예시와 함께 정리하여, 기초부터 고급 멀티 에이전트 패턴까지의 로드맵을 공부하려 합니다.

1. LangGraph의 기초: StateGraph와 구조

LangGraph의 핵심은 상태(State)를 정의하고, 노드(Node)들이 이 상태를 주고받으며 작업을 수행하는 것입니다. 그래프는 크게 상태, 노드, 엣지(Edge)로 구성됩니다.

 

이를 공유 화이트보드에 비유할 수 있습니다. 회의실(Graph)에 여러 사람(Node)이 모여 있고, 화이트보드(State)에 적힌 내용을 보고 각자 할 일을 처리한 뒤 결과를 다시 화이트보드에 적는 방식입니다.

• 개념: 상태는 모든 노드가 공유하는 기억 공간이며, 노드는 실제 작업을 수행하는 일꾼입니다.
• 예시 상황: 챗봇이 사용자의 질문을 받으면, 먼저 처리 노드가 질문 내용을 분석해 상태에 저장하고, 답변 생성 노드가 그 내용을 보고 답변을 작성하여 다시 상태에 업데이트합니다.
• State (상태): 애플리케이션의 현재 상황(대화 기록, 변수 등)을 저장하는 공유 메모리입니다.
• Node (노드): 실제 로직(LLM 호출, 함수 실행)을 수행하고 상태를 업데이트하는 단위입니다.
• Edge (엣지): 노드 간의 연결 경로를 정의합니다.

Python

 

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

class GraphState(TypedDict):
    question: str
    answer: str

def chatbot_node(state: GraphState):
    return {"answer": "응답 메시지입니다."}

workflow = StateGraph(GraphState)
workflow.add_node("chatbot", chatbot_node)
workflow.set_entry_point("chatbot")
workflow.add_edge("chatbot", END)

app = workflow.compile()

2. 조건부 라우팅 (Conditional Routing)

모든 대화가 똑같은 순서로 진행되지는 않습니다. 마치 회사의 안내 데스크처럼, 방문자의 목적에 따라 다른 부서로 안내해 주는 역할이 필요합니다.

• 개념: 사용자의 입력이나 현재 상태에 따라 다음에 실행할 노드를 동적으로 결정하는 기술입니다.
• 예시 상황: 사용자가 "오늘 날씨 어때?"라고 물으면 검색 도구 노드로 보내고, "안녕, 반가워"라고 인사하면 일반 대화 노드로 보내는 방식입니다.

Python

 

def route_decision(state: GraphState):
    if "검색" in state["question"]:
        return "search_tool"
    else:
        return "chatbot"

workflow.add_conditional_edges(
    "router",
    route_decision,
    {
        "search_tool": "search_tool",
        "chatbot": "chatbot"
    }
)

3. 메모리 관리와 영속성 (Reducer & Checkpointer)

AI가 이전 대화를 기억하려면 뇌에 해당하는 메모리 기능이 필요합니다. 단순히 정보를 덮어쓰는 것이 아니라, 일기장처럼 내용을 계속 이어 적어야 합니다. 또한 게임의 세이브 기능처럼 프로그램이 종료되어도 기억을 유지해야 합니다.

• 개념: Reducer는 새로운 메시지를 기존 대화 목록에 추가(Append)하는 역할을 하며, Checkpointer는 이 대화 기록을 데이터베이스에 영구적으로 저장합니다.
• 예시 상황: 사용자가 어제 "나 내일 소개팅해"라고 말하고 앱을 껐다가 오늘 다시 켰을 때, AI가 "어제 말한 소개팅은 잘 했어?"라고 먼저 물어볼 수 있게 됩니다.

 

• Reducer: 상태를 덮어쓰지 않고 기존 리스트에 추가하는 로직입니다.
• Checkpointer: thread_id를 기준으로 대화 상태를 영구 저장하여 세션이 끊겨도 대화를 이어나갈 수 있게 합니다.

 

Python

 

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import MessagesState

workflow = StateGraph(MessagesState)

memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "user_session_1"}}
app.invoke({"messages": [HumanMessage(content="안녕하세요")]}, config=config)

4. 휴먼 인 더 루프 (Human-in-the-loop)와 스트리밍

중요한 결정을 내릴 때는 AI 혼자 처리하게 두지 않고 사람이 검토해야 할 때가 있습니다. 또한, 답변이 생성되는 과정을 실시간으로 보여주어 기다리는 지루함을 줄여야 합니다.

• 개념: 특정 단계 직전에 시스템을 일시 정지시켜 사람의 승인을 기다리거나 데이터를 수정할 수 있게 합니다. 스트리밍은 생성된 글자를 실시간으로 화면에 뿌려주는 기술입니다.
• 예시 상황: AI가 고객에게 보낼 사과 이메일 초안을 작성한 뒤, "전송" 버튼을 누르기 전에 담당자에게 팝업을 띄워 내용을 확인받습니다. 담당자가 "승인"을 눌러야만 실제 메일이 발송됩니다.

 

• interrupt_before: 특정 노드 실행 직전에 그래프를 멈추고 대기합니다.
• Streaming: astream_events를 사용하여 토큰 단위로 생성되는 응답을 실시간으로 보여줍니다.

 

Python

 

app = workflow.compile(
    checkpointer=memory,
    interrupt_before=["send_email"]
)

async for event in app.astream_events(inputs, version="v2"):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="")

5. 멀티 에이전트 패턴 1: Supervisor (감독관)

복잡한 프로젝트를 진행할 때 프로젝트 매니저(PM)가 팀원들에게 업무를 지시하는 것과 같습니다.

• 개념: 중앙 관리자(Supervisor) 에이전트가 사용자의 요청을 듣고, 식단 전문가, 운동 전문가 등 적절한 하위 에이전트에게 작업을 지시하고 결과를 보고받습니다.
• 예시 상황: 사용자가 "다이어트 계획 짜줘"라고 하면, 감독관이 먼저 식단 전문가에게 "식단표 짜와"라고 시키고, 그 다음 운동 전문가에게 "운동 루틴 짜와"라고 시킨 뒤, 마지막에 이를 종합하여 사용자에게 전달합니다.

Python

 

def supervisor_node(state: TeamState):
    response = llm.invoke(messages)
    return {"next": "DietExpert"} 

workflow.add_conditional_edges(
    "Supervisor",
    lambda state: state["next"],
    {
        "DietExpert": "DietExpert",
        "SleepExpert": "SleepExpert",
        "FINISH": END
    }
)

6. 멀티 에이전트 패턴 2: Handoff (업무 이관)

종합병원에서 진료를 보다가 의사가 "이건 제 분야가 아니니 저쪽 전문의에게 가보세요"라고 환자를 바로 연결해 주는 것과 비슷합니다.

• 개념: 중앙 관리자를 거치지 않고, 에이전트가 스스로 판단하여 다른 전문 에이전트에게 직접 제어권을 넘깁니다. 도구(Tool) 기능을 활용해 다른 에이전트를 호출합니다.
• 예시 상황: 고객센터 챗봇이 상담하다가 기술적인 문제가 나오면 "잠시만요, 기술 지원팀 연결해 드릴게요"라고 말하며 대화의 주도권을 기술 지원 에이전트에게 바로 넘깁니다.

Python

 

@tool
def transfer_to_sleep_expert():
    return "이동"

def should_continue(state):
    if tool_called:
        return "sleep_expert"
    return END

workflow.add_conditional_edges("diet_expert", should_continue)

7. 멀티 에이전트 패턴 3: Swarm (병렬 처리)

자동차 경주에서 피트인(Pit-in)을 했을 때, 여러 정비공이 달라붙어 타이어 교체와 주유를 동시에 진행하는 것과 같습니다.

• 개념: 서로 관련 없는 독립적인 작업들을 동시에 실행하여 전체 처리 시간을 단축하는 패턴입니다. 하나의 시작점에서 여러 갈래로 흩어졌다가(Fan-out) 나중에 하나로 합쳐집니다(Fan-in).
• 예시 상황: 사용자가 "내 건강 상태 종합 분석해줘"라고 요청하면, 식단 분석가와 수면 분석가가 동시에 각자 데이터를 분석합니다. 순차적으로 하면 10초 걸릴 일을 동시에 5초 만에 끝내고 결과를 합칠 수 있습니다.

Python

 

workflow.set_entry_point(["diet_expert", "sleep_expert"])

workflow.add_edge("diet_expert", "summarizer")
workflow.add_edge("sleep_expert", "summarizer")