6장. 랭그래프로 설계하는 RAG 파이프라인

출처: 『RAG 마스터: 랭체인으로 완성하는 LLM 서비스』(브라이스 유·조경아·박수진·김재웅, 프리렉 2025) | 공식: https://langchain-ai.github.io/langgraph/

LCEL의 선형 체인은 순서대로 실행된다. 하지만 "문서가 부적절하면 재검색하라", "코드에 오류가 있으면 재생성하라" 같은 분기·루프가 필요한 순간, 체인은 한계를 드러낸다. 랭그래프는 RAG 파이프라인을 그래프로 선언해 이 복잡성을 다룬다 — 노드는 작업을 수행하고, 에지는 흐름을 결정하며, 상태는 노드 간 정보를 공유한다.

실습 — 책 공식 노트북: Ch06. LangGraph 폴더. 코랩 바로 열기: LangGraph 기초·코드 어시스트 챗봇·Corrective RAG.

학습 목표

이 장을 끝내면 다음을 할 수 있다.

• 랭그래프의 세 구성요소(State·Node·Edge)의 역할을 설명하고, LCEL 체인과의 구조적 차이를 비교한다.
• TypedDict와 리듀서(add_messages)를 사용해 상태를 정의하고, 노드 간 정보 공유 방식을 구현한다.
• add_edge와 add_conditional_edges로 고정 연결과 조건부 분기를 구성하고, 라우팅 함수를 작성한다.
• MemorySaver 체크포인터와 thread_id를 결합해 다중 세션 대화 지속성을 구현한다.
• Corrective-RAG 패턴(검색→평가→쿼리 재작성→웹 검색→생성)을 랭그래프로 설계하고 조립한다.
• 코드 어시스트 챗봇의 생성→검사→반영→재생성 루프를 decide_to_finish 에지로 제어한다.

전체 흐름도

랭그래프 RAG 파이프라인의 핵심 흐름: 상태 정의 → 노드 구현 → 에지 연결 → 컴파일 → 실행.

[ State 정의 (TypedDict + 리듀서) ]
   │
   ▼
[ Node 구현 ]
   ├─ retrieve        (문서 검색)
   ├─ grade_documents (연관성 평가)
   ├─ transform_query (쿼리 재작성)
   ├─ web_search_node (웹 검색 보완)
   └─ generate        (답변 생성)
   │
   ▼
[ Edge 연결 ]
   ├─ 일반 에지   : add_edge(A, B)          — 고정 순서
   └─ 조건부 에지 : add_conditional_edges   — 상태 기반 분기
   │
   ▼
[ 그래프 컴파일 : workflow.compile(checkpointer=...) ]
   │
   ▼
[ 실행 : app.stream(inputs) ]
   ├─ stream_mode="values"  → 전체 상태 반환
   └─ stream_mode="updates" → 변경분만 반환

0. 사전 필수 용어

• 그래프(Graph) — 객체 간 관계를 나타내는 자료구조. 노드(정점)와 에지(간선)로 구성된다.
• 노드(Node) — 그래프에서 실제 작업을 수행하는 실행 단위. 파이썬 함수가 곧 노드다. 상태를 입력으로 받아 업데이트된 상태를 반환한다.
• 에지(Edge) — 노드 간 실행 흐름을 연결하는 요소. 고정 연결(일반 에지)과 조건 분기(조건부 에지) 두 종류가 있다.
• 상태(State) — 그래프 내 모든 노드가 공유하는 변수의 집합. 주로 TypedDict로 정의하며, 노드마다 부분 업데이트가 가능하다.
• StateGraph — 사용자 정의 상태를 매개변수로 활용하는 범용 그래프 클래스. 복잡한 워크플로우에 사용된다.
• MessageGraph — 메시지 목록만으로 구성된 특수 유형. 단순 대화형 시스템에 적합하다.
• 리듀서(Reducer) — Annotated[list, add_messages] 처럼 상태 필드에 붙여, 기존 값을 덮어쓰는 대신 병합하는 함수.
• checkpointer(체크포인터) — 각 실행 단계의 상태를 저장소에 보존하는 컴포넌트. MemorySaver가 가장 단순한 인메모리 구현이다.
• 슈퍼스텝(Superstep) — 여러 노드가 병렬로 동시에 작업하는 그래프 처리 단위. 구글 Pregel에서 영감.
• interrupt_before — 지정 노드 실행 직전에 그래프를 멈춰 인간이 상태를 검토·수정하게 하는 파라미터.

1. 랭그래프의 구성요소

LCEL 체인 vs LangGraph를 먼저 비교하고 나면, 각 구성요소의 필요성이 명확해진다.

1.1 그래프

랭그래프 그래프는 구글 Pregel의 슈퍼스텝 방식으로 동작한다. 슈퍼스텝 하나에서 여러 노드가 병렬 실행되고, 입력이 없는 노드는 비활성화된다. 모든 노드가 비활성화되면 그래프가 종료된다.

두 가지 그래프 클래스: - StateGraph — 범용. 사용자가 정의한 State를 매개변수로 받는다. 이 장에서 주로 사용. - MessageGraph — 메시지 목록만으로 구성된 특수 유형. 단순 대화형 시스템에 적합.

1.2 상태(State)

상태는 그래프 내 모든 노드가 공유하는 전역 변수 묶음이다. TypedDict로 선언하며, 각 노드는 상태를 전부 반환할 필요 없이 변경한 필드만 반환해도 된다.

기본 상태 정의 — 덮어쓰기 방식:

from typing import TypedDict

class State(TypedDict):
    count: int
    messages: list[str]

노드가 {"count": 2}를 반환하면, count만 업데이트되고 messages는 유지된다.

리듀서 방식 — 병합(누적):

from typing import TypedDict, Annotated
from operator import add

class State(TypedDict):
    count: int
    messages: Annotated[list[str], add]  # 새 메시지를 리스트에 append

messages에 add 리듀서가 붙으면, 노드가 ["bye"]를 반환해도 기존 ["hi"]가 보존되어 ["hi", "bye"]가 된다.

채팅 전용 — add_messages 리듀서:

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

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

add_messages는 LangChain 메시지 객체를 id 기준으로 관리해준다. 메시지 추가·수정 모두 처리.

1.3 노드(Node)

노드는 파이썬 함수다. 첫 번째 인자로 state(현재 상태), 두 번째로 config(선택, 설정값)를 받고, 업데이트된 상태 딕셔너리를 반환한다.

from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, START, END

builder = StateGraph(dict)

def my_node(state: dict, config: RunnableConfig):
    user_id = config["configurable"]["user_id"]
    return {"result": f"Hello, {state['input']}! (user: {user_id})"}

def my_other_node(state: dict):
    return state  # 상태 변경 없이 그대로 전달

builder.add_node("my_node", my_node)
builder.add_node("other_node", my_other_node)

START / END 노드: - START — 그래프 실행의 시작점. 사용자 입력을 처음 받아 첫 노드로 전달. - END — 실행 완료를 나타내는 종료점. 더 이상 처리할 작업이 없을 때 연결.

from langgraph.graph import START, END

graph.add_edge(START, "node_a")
graph.add_edge("node_a", END)

1.4 에지(Edge)

에지는 노드 실행 후 다음 노드를 결정하는 흐름 제어 요소다.

일반 에지 — 고정 연결:

graph.add_edge("node_a", "node_b")

조건부 에지 — 상태에 따른 분기:

from typing import Literal

def routing_function(state: State) -> Literal["node_b", "node_c"]:
    if state["some_flag"]:
        return "node_b"
    return "node_c"

graph.add_conditional_edges(
    "node_a",
    routing_function,
    {"node_b": "node_b", "node_c": "node_c"}
)

조건부 진입 지점 — START에서 조건 분기:

from langgraph.graph import START

graph.add_conditional_edges(
    START,
    routing_function,
    {True: "node_b", False: "node_c"}
)

2. 랭그래프 활용

기본 챗봇(루프) → 도구 연동(조건부 에지) → 스트리밍 → 상태 저장 → 인간 개입 순서로 기능을 쌓아간다.

2.1 루프 구현

가장 단순한 챗봇 그래프. 사용자 입력 → LLM 응답의 단일 루프.

from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

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

llm = ChatOpenAI(model="gpt-4o-mini")

def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
graph = graph_builder.compile()

while True:
    user_input = input("User: ")
    if user_input.lower() in ["quit", "exit", "q"]:
        print("Goodbye!")
        break
    for event in graph.stream({"messages": ("user", user_input)}):
        for value in event.values():
            print("Assistant:", value["messages"][-1].content)

2.2 조건문(조건부 에지) 구현

Tavily 검색 도구를 연동해, LLM이 도구 호출이 필요하다고 판단하면 tools 노드로, 단순 답변이면 END로 분기한다.

from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.prebuilt import ToolNode, tools_condition

tool = TavilySearchResults(max_results=2)
tools = [tool]

llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)

def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)

# 미리 빌드된 도구 노드 (BasicToolNode 직접 구현의 대안)
tool_node = ToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)

# 미리 빌드된 조건부 에지 (tools_condition: tool_calls 존재하면 tools, 없으면 END)
graph_builder.add_conditional_edges("chatbot", tools_condition)
graph_builder.add_edge("tools", "chatbot")  # 도구 실행 후 다시 chatbot으로
graph_builder.add_edge(START, "chatbot")

graph = graph_builder.compile()

PrebuiltComponents 참고 — ToolNode(도구 노드)와 tools_condition(조건부 에지 함수)은 랭그래프가 제공하는 미리 빌드된 컴포넌트다. from langgraph.prebuilt import ToolNode, tools_condition으로 불러와 직접 구현 없이 재사용 가능하다.

2.3 스트리밍

graph.stream() 호출 시 stream_mode 파라미터로 동작 방식을 선택한다.

from langchain_core.messages import BaseMessage

events = graph.stream(
    input={"messages": [("user", user_input)]},
    stream_mode="updates"
)
for event in events:
    for value in event.values():
        if isinstance(value["messages"][-1], BaseMessage):
            print("Assistant:", value["messages"][-1].content)

2.4 상태 저장(Persistence)

MemorySaver 체크포인터를 컴파일 시 지정하면, thread_id 별로 대화 상태가 자동 저장·복원된다.

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()

# compile 시 checkpointer 지정 — 이것만으로 상태 저장 활성화
graph = graph_builder.compile(checkpointer=memory)

# thread_id로 대화 세션 구분
config = {"configurable": {"thread_id": "session_1"}}

events = graph.stream(
    {"messages": [("user", "안녕! 내 이름은 오해원이야.")]},
    config,
    stream_mode="values"
)
for event in events:
    event["messages"][-1].pretty_print()

# 같은 thread_id → 이전 맥락 유지
events = graph.stream(
    {"messages": [("user", "내 이름을 기억하니?")]},
    config,
    stream_mode="values"
)
# → "네, 오해원님!" 처럼 맥락 유지된 답변

# 현재 상태 스냅샷 조회
snapshot = graph.get_state(config)
print(snapshot.values["messages"])  # 전체 대화 이력

thread_id 격리 — thread_id가 다르면 맥락이 분리된다. thread_id: "session_1"에서 한 대화는 thread_id: "session_2"에서 기억하지 못한다.

2.5 루프 개입(Human-in-the-Loop)

interrupt_before 파라미터에 노드를 지정하면, 해당 노드 실행 직전에 그래프가 멈춘다. 이 시점에 상태를 확인하고 수정한 뒤 재개할 수 있다.

# tools 노드 실행 직전에 개입
graph = graph_builder.compile(
    checkpointer=memory,
    interrupt_before=["tools"]
)

config = {"configurable": {"thread_id": "2"}}
events = graph.stream(
    {"messages": [("user", "지금 서울 날씨 어때?")]},
    config,
    stream_mode="values"
)
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()
# → LLM이 tool_calls 정보를 생성했지만 실행 전에 멈춤

# 다음 실행 노드 확인
snapshot = graph.get_state(config)
print(snapshot.next)  # ('tools',)

# 상태를 직접 수정 (실제 도구 호출 없이 강제 응답 주입)
from langchain_core.messages import ToolMessage, AIMessage

existing_message = snapshot.values["messages"][-1]
tool_call_id = existing_message.tool_calls[0]["id"]
answer = "서울의 날씨는 매우 맑아요."

graph.update_state(
    config,
    {"messages": [
        ToolMessage(content=answer, tool_call_id=tool_call_id),
        AIMessage(content=answer),
    ]}
)

루프 개입 활용 시나리오: 현재 상태 편집·과거 기록 탐색·상태 수정·특정 시점 메시지 추가. AI 응답을 정밀하게 제어해 신뢰도를 높인다.

3. 랭그래프 실습

3.1 자체교정 RAG(Corrective-RAG)

기본 RAG의 한계: 검색된 문서가 질문과 무관하면 답변 품질이 크게 떨어진다. 자체교정 RAG는 검색 → 연관성 평가 → 부적절하면 쿼리 재작성 → 웹 검색 보완 → 생성의 루프를 구성해 이를 극복한다.

흐름: START → retrieve → grade_documents → (연관 있음 → generate → END | 연관 없음 → transform_query → web_search_node → generate → END)

1단계: 상태 정의

from typing import List
from typing_extensions import TypedDict

class GraphState(TypedDict):
    question: str        # 사용자 질문 (재작성 후 덮어씀)
    generation: str      # LLM 생성 답변
    web_search: str      # 웹 검색 필요 여부 ("예"/"아니오")
    documents: List[str] # 검색된 문서 리스트

2단계: 문서 연관성 평가 노드

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

class GradeDocuments(BaseModel):
    binary_score: str = Field(description="문서와 질문의 연관성 여부 (예 or 아니오)")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm_grader = llm.with_structured_output(GradeDocuments)

system = """당신은 사용자의 질문에 대해 검색된 문서의 관련성을 평가하는 전문가입니다.
문서에 질문과 관련된 키워드나 의미가 담겨 있으면 '관련 있음'으로 평가하세요.
'예' 또는 '아니오'로 표시해 주세요."""

grade_prompt = ChatPromptTemplate.from_messages([
    ("system", system),
    ("human", "검색된 문서: \n\n {document} \n\n 사용자 질문: {question}"),
])
retrieval_grader = grade_prompt | structured_llm_grader

3단계: 노드 함수들

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_core.output_parsers import StrOutputParser
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain.schema import Document

# 문서 인덱싱 (사전 준비)
urls = [
    "https://google.github.io/styleguide/pyguide.html",
    "https://google.github.io/styleguide/javaguide.html",
]
docs = [WebBaseLoader(url).load() for url in urls]
docs_list = [item for sublist in docs for item in sublist]
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=250, chunk_overlap=0
)
doc_splits = text_splitter.split_documents(docs_list)
vectorstore = Chroma.from_documents(
    documents=doc_splits,
    collection_name="rag-chroma",
    embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever()

# RAG 체인 (답변 생성용)
system_rag = """당신은 질문에 답변하는 도우미입니다.
제공된 문맥을 바탕으로 질문에 답변하세요. 모르면 모른다고 말하세요.
세 문장을 넘지 않도록 간결하게 작성하세요."""
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", system_rag),
    ("human", "질문: {question} \n문맥: {context} \n답변:"),
])
rag_chain = rag_prompt | llm | StrOutputParser()

# 질문 재작성 체인
rewrite_system = """당신은 입력된 질문을 웹 검색에 최적화된 형태로 만드는 질문 생성기입니다.
입력된 질문의 이면에 있는 의도를 파악해주세요."""
rewrite_prompt = ChatPromptTemplate.from_messages([
    ("system", rewrite_system),
    ("human", "질문: \n\n {question} \n 더 나은 질문으로 바꿔주세요."),
])
question_rewriter = rewrite_prompt | llm | StrOutputParser()

# 웹 검색 도구
web_search_tool = TavilySearchResults(k=3)

def retrieve(state):
    """문서 검색 노드"""
    question = state["question"]
    documents = retriever.invoke(question)
    return {"documents": documents, "question": question}

def generate(state):
    """답변 생성 노드"""
    question = state["question"]
    documents = state["documents"]
    generation = rag_chain.invoke({"context": documents, "question": question})
    return {"documents": documents, "question": question, "generation": generation}

def grade_documents(state):
    """문서 연관성 평가 노드"""
    question = state["question"]
    documents = state["documents"]
    filtered_docs = []
    web_search = "아니오"
    for d in documents:
        score = retrieval_grader.invoke(
            {"question": question, "document": d.page_content}
        )
        if score.binary_score == "예":
            filtered_docs.append(d)
        else:
            web_search = "예"
    return {"documents": filtered_docs, "question": question, "web_search": web_search}

def transform_query(state):
    """쿼리 재작성 노드"""
    question = state["question"]
    documents = state["documents"]
    better_question = question_rewriter.invoke({"question": question})
    return {"documents": documents, "question": better_question}

def web_search_node(state):
    """웹 검색 노드"""
    question = state["question"]
    documents = state["documents"]
    docs = web_search_tool.invoke({"query": question})
    web_results = "\n".join([d["content"] for d in docs])
    web_results = Document(page_content=web_results)
    documents.append(web_results)
    return {"documents": documents, "question": question}

# 에지 라우팅 함수
def decide_to_generate(state):
    """웹 검색 필요 여부로 다음 노드 결정"""
    if state["web_search"] == "예":
        return "transform_query"
    return "generate"

4단계: 그래프 조립

from langgraph.graph import END, StateGraph, START

workflow = StateGraph(GraphState)

# 노드 등록
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("transform_query", transform_query)
workflow.add_node("web_search_node", web_search_node)

# 에지 연결
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {"transform_query": "transform_query", "generate": "generate"},
)
workflow.add_edge("transform_query", "web_search_node")
workflow.add_edge("web_search_node", "generate")
workflow.add_edge("generate", END)

app = workflow.compile()

# 실행
inputs = {"question": "구글의 코드 작성 가이드"}
for output in app.stream(inputs):
    for key, value in output.items():
        print(f"Node '{key}':")
        if "generation" in value:
            print(value["generation"])

3.2 코드 어시스트 챗봇

LLM으로 코드를 생성하고, exec()로 직접 실행 테스트를 통과할 때까지 재시도하는 그래프. 코드 생성 → import 검사 → 코드 실행 검사 → (오류 있으면 반영 후 재생성) → 종료의 루프다.

동작 흐름:

사용자 질문 + 코드 맥락
   │
   ▼
generate (코드 생성)
   │
   ▼
check_code (import 실행 → 전체 코드 실행)
   ├─ 오류 없음 또는 최대 시도(3회) → END
   └─ 오류 있음 → reflect (오류 반영) → generate (재생성)

상태와 코드 생성 체인:

from typing import List, TypedDict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

class GraphState(TypedDict):
    error: str        # 테스트 오류 여부 ("yes"/"no")
    messages: List    # 사용자 질문·오류 메시지 등 포함
    generation: str   # 생성된 코드
    iterations: int   # 시도 횟수

class Code(BaseModel):
    prefix: str = Field(description="문제와 접근 방식 설명")
    imports: str = Field(description="import 문")
    code: str = Field(description="import 제외 코드 블록")
    description: str = Field(description="코드 스키마 설명")

system = """당신은 LCEL 전문가인 코딩 어시스턴트입니다.
다음은 필요한 LCEL 문서 전문입니다: {context}
제공하는 코드는 실행 가능해야 하며, 필요한 모든 import와 변수가 정의되어 있어야 합니다."""

code_gen_prompt = ChatPromptTemplate.from_messages([
    ("system", system),
    ("placeholder", "{messages}"),
])
llm = ChatOpenAI(temperature=0, model="gpt-4o-mini")
code_gen_chain = code_gen_prompt | llm.with_structured_output(Code)

노드 함수들:

def generate(state: GraphState):
    """코드 생성 노드"""
    messages = state["messages"]
    iterations = state["iterations"]
    error = state.get("error", "no")
    if error == "yes":
        messages += [("user", "다시 시도해보세요. prefix, imports, code 블록으로 구조화하세요.")]
    code_solution = code_gen_chain.invoke(
        {"context": concatenated_content, "messages": messages}
    )
    messages += [("assistant", f"{code_solution.prefix}\nImports:\n{code_solution.imports}\nCode:\n{code_solution.code}")]
    return {"generation": code_solution, "messages": messages, "iterations": iterations + 1}

def code_check(state: GraphState):
    """코드 검사 노드 — import 실행 → 전체 코드 실행 두 단계"""
    messages = state["messages"]
    code_solution = state["generation"]
    iterations = state["iterations"]
    imports = code_solution.imports
    code = code_solution.code
    try:
        exec(imports)
    except Exception as e:
        error_message = [("user", f"import 테스트 실패: {e}")]
        return {"generation": code_solution, "messages": messages + error_message,
                "iterations": iterations, "error": "yes"}
    try:
        exec(imports + "\n" + code)
    except Exception as e:
        error_message = [("user", f"코드 실행 테스트 실패: {e}")]
        return {"generation": code_solution, "messages": messages + error_message,
                "iterations": iterations, "error": "yes"}
    return {"generation": code_solution, "messages": messages, "iterations": iterations, "error": "no"}

def reflect(state: GraphState):
    """오류 반영 후 재생성 준비 노드"""
    messages = state["messages"]
    iterations = state["iterations"]
    code_solution = state["generation"]
    reflections = code_gen_chain.invoke({"context": concatenated_content, "messages": messages})
    messages += [("assistant", f"오류를 반영한 코드입니다: {reflections}")]
    return {"generation": code_solution, "messages": messages, "iterations": iterations}

MAX_ITERATIONS = 3

def decide_to_finish(state: GraphState):
    """오류 없거나 최대 시도 횟수 도달 시 종료, 아니면 재시도"""
    error = state["error"]
    iterations = state["iterations"]
    if error == "no" or iterations >= MAX_ITERATIONS:
        return "end"
    return "reflect"

그래프 조립:

from langgraph.graph import END, StateGraph, START

workflow = StateGraph(GraphState)
workflow.add_node("generate", generate)
workflow.add_node("check_code", code_check)
workflow.add_node("reflect", reflect)

workflow.add_edge(START, "generate")
workflow.add_edge("generate", "check_code")
workflow.add_conditional_edges(
    "check_code",
    decide_to_finish,
    {"end": END, "reflect": "reflect", "generate": "generate"},
)
workflow.add_edge("reflect", "generate")

app = workflow.compile()

# 실행
question = "LCEL로 RAG 체인을 어떻게 만들어?"
app.invoke({
    "messages": [("user", question)],
    "iterations": 0
})

핵심 개념 정리

한 문장 요약 — 랭그래프는 노드(함수)·에지(흐름)·상태(공유 변수)를 그래프로 조직해, LCEL의 선형 체인이 다루지 못하는 분기·루프·상태 저장·인간 개입이 있는 복잡한 RAG 파이프라인을 체계적으로 구현하게 해준다.

실무 체크리스트

• [ ] StateGraph(State)에서 State는 TypedDict로 정의하고, 메시지 누적에는 Annotated[list, add_messages]를 쓴다.
• [ ] 도구 연동이 반복적인 패턴이면 ToolNode + tools_condition 프리빌트 컴포넌트를 활용해 코드를 줄인다.
• [ ] compile(checkpointer=MemorySaver())와 thread_id를 세트로 기억한다 — 둘 중 하나만으론 지속성이 안 된다.
• [ ] interrupt_before=[노드명]으로 개입 지점을 만들고, graph.update_state(config, ...)로 상태를 수동 수정 후 재개한다.
• [ ] Corrective-RAG의 decide_to_generate 같은 에지 함수는 반드시 Literal["노드명", ...] 타입 힌트를 붙여 라우팅 버그를 방지한다.
• [ ] 코드 어시스트처럼 exec()로 실행 검증하는 패턴은 운영 환경에서 샌드박싱을 반드시 고려한다.
• [ ] 노드 함수는 변경한 필드만 반환해도 된다. 전체 상태를 반환할 의무가 없음을 기억한다.
• [ ] graph.get_state(config)로 StateSnapshot을 조회하면 전체 대화 이력·next 노드·체크포인트 ID를 확인할 수 있다.

연습문제

1. 개념. LCEL 체인과 LangGraph의 가장 큰 구조적 차이를 한 문장으로 설명하고, LangGraph가 반드시 필요한 시나리오를 한 가지 들어라.
2. 비교. Annotated[list, add] 리듀서와 Annotated[list, add_messages] 리듀서의 차이는 무엇인가? 각각 어떤 상황에 쓰이는가?
3. 코드. StateGraph로 두 노드(A, B)를 만들고, A의 상태값 score가 70 이상이면 B로, 미만이면 END로 분기하는 그래프를 작성하라.
4. 적용. MemorySaver를 사용하는 챗봇에서 두 사용자(Alice, Bob)가 동시에 대화할 때, 각자의 맥락을 분리하려면 어떻게 config를 구성해야 하는가?
5. 설계. Corrective-RAG를 확장해 "LLM이 생성한 답변이 환각(hallucination)인지 평가하는 노드"를 추가하고 싶다. 이 노드를 그래프의 어느 위치에 어떤 에지로 연결할지 설계하라.

최신 동향 (2026-05 기준)

최신 동향 (검증 2026-05-21) — 책은 2025년 기준이므로 아래를 보완해 익히면 된다.

• LangGraph 독립 패키지로 성숙. 책 집필 시점 이후 LangGraph는 독립 패키지로 성숙했다. 복잡한 분기·루프·에이전트 파이프라인의 표준으로 자리잡아, 단순 LCEL 체인 대신 LangGraph 권장 범위가 크게 확대됐다. (출처: https://langchain-ai.github.io/langgraph/)
• StateGraph + checkpointer 조합이 대화 메모리 표준. 책 1장의 RunnableWithMessageHistory·ConversationBufferMemory는 LangChain 0.3에서 deprecated. 대화 상태 관리는 LangGraph StateGraph + MemorySaver/SqliteSaver checkpointer로 이전 권장. (출처: https://python.langchain.com/docs/versions/migrating_memory/)
• 모델·가격은 수시 변동. 책의 gpt-4o-mini 표기는 2025년 기준 예시다. 실제 사용 시 공식 모델 목록 확인: OpenAI(https://platform.openai.com/docs/models) · Anthropic(https://docs.anthropic.com/en/docs/about-claude/models) · Google(https://ai.google.dev/gemini-api/docs/models).

부록 A. 용어 사전

부록 B. 핵심 비교표

LCEL 체인(선형) vs LangGraph(그래프)

실습 패턴 비교

부록 C. 추천 참고 자료

검증된 외부 자료 (Tier 1 공식, 생존 확인 2026-05-21)

부록 D. 연습문제 풀이

1. (LCEL은 선형, LangGraph는 그래프) LCEL 체인은 노드를 좌에서 우로 순서대로만 실행하는 선형 파이프라인인 반면, LangGraph는 분기·루프·조건부 흐름이 가능한 그래프 구조다. LangGraph가 반드시 필요한 시나리오는 "검색된 문서가 부적절하면 쿼리를 재작성하고 웹 검색을 추가로 수행하는" Corrective-RAG처럼 상태에 따라 실행 경로가 달라지는 루프형 파이프라인이다.

2. (리듀서 두 종류의 차이) Annotated[list, add]는 파이썬 기본 operator.add를 사용해 단순히 리스트를 이어 붙이는 반면, Annotated[list, add_messages]는 LangChain 메시지 객체의 id 필드를 기준으로 추가 또는 수정 처리를 한다. 전자는 일반 문자열·문서 등 단순 목록 누적에, 후자는 HumanMessage·AIMessage 같은 LangChain 메시지 객체를 관리하는 대화 상태 필드에 사용한다.

3. (조건부 에지 그래프 작성) score 값을 기준으로 B 노드 또는 END로 분기하는 최소 그래프는 다음과 같다.

   ```python from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END

   class State(TypedDict): score: int result: str

   def node_a(state: State): return {"score": state["score"]}

   def node_b(state: State): return {"result": "B 노드 처리 완료"}

   def routing(state: State) -> Literal["node_b", "end"]: if state["score"] >= 70: return "node_b" return END

   builder = StateGraph(State) builder.add_node("node_a", node_a) builder.add_node("node_b", node_b) builder.add_edge(START, "node_a") builder.add_conditional_edges("node_a", routing, {"node_b": "node_b", END: END}) builder.add_edge("node_b", END) graph = builder.compile() ```

4. (다중 세션 맥락 분리) MemorySaver는 thread_id 단위로 대화 상태를 저장하므로, Alice와 Bob 각자에게 고유한 thread_id를 할당한 config 딕셔너리를 사용하면 된다. 예를 들어 Alice는 config = {"configurable": {"thread_id": "alice"}}, Bob은 config = {"configurable": {"thread_id": "bob"}}으로 구성한다. 서로 다른 thread_id는 완전히 격리된 상태 저장소로 동작하기 때문에 두 사용자의 맥락이 섞이지 않는다.

5. (환각 평가 노드 설계) 환각 평가 노드는 generate 노드 직후, 즉 generate → grade_hallucination으로 일반 에지를 연결한다. grade_hallucination 노드는 생성된 답변(generation)과 검색 문서(documents)를 비교해 환각 여부를 판정하고, 상태에 예컨대 hallucination: str 필드를 "예"/"아니오"로 업데이트한다. 이후 add_conditional_edges로 "아니오"이면 END로, "예"이면 transform_query(쿼리 재작성)로 돌아가 재검색·재생성 루프를 수행하도록 연결한다. 이렇게 하면 Corrective-RAG의 기존 흐름(grade_documents → decide_to_generate)을 건드리지 않고 생성 단계 이후에 품질 게이트를 추가할 수 있다.