LangChain (5) - 흐름 제어

LangChain 흐름 제어와 LangGraph

프롬프트, 모델, 파서를 Runnable로 연결한 파이프라인은 데이터가 한 방향으로 흐를 때 가장 이해하기 쉽습니다. 검색한 문서를 프롬프트에 넣고 모델을 호출하거나, 같은 입력을 여러 경로로 보내 병렬 처리한 뒤 결과를 모으는 작업이 대표적입니다.

조건에 따라 다른 경로를 선택하는 정도라면 RunnableBranch 같은 Runnable 조합으로도 처리할 수 있습니다. 문제는 실행이 진행되면서 여러 단계가 하나의 상태를 계속 갱신하고, 결과에 따라 이전 단계로 돌아가며, 실행 도중 멈춘 상태를 저장했다가 나중에 다시 이어야 할 때 생깁니다.

이런 흐름은 단순히 앞 단계의 출력을 다음 단계의 입력으로 넘기는 파이프라인보다 현재 상태와 다음에 실행할 단계를 함께 관리하는 실행 구조로 표현하는 편이 자연스럽습니다. LangChain 생태계에서 이 역할을 담당하는 low-level orchestration runtime이 LangGraph입니다.

LangGraph에서는 작업을 node와 edge로 구성하고, 각 node가 공유 state를 읽고 갱신하면서 다음 실행 경로를 결정할 수 있습니다. 여기에 checkpoint를 붙이면 실행 상태를 저장할 수 있어 human-in-the-loop, 중단 후 재개, fault recovery 같은 동작도 같은 실행 모델 안에서 처리할 수 있습니다.

Runnable과 LCEL로 표현하기 좋은 단방향 dataflow와 RunnableBranch를 이용한 간단한 조건 분기부터 살펴봅니다. 이어서 cycle, shared state, persistence가 필요한 흐름에서는 무엇이 달라지는지 보고, 언제 Runnable 조합을 유지하고 언제 LangGraph의 실행 모델로 넘어가는 것이 좋은지 구분합니다.

---

1장. LCEL이 잘 맞는 흐름

LCEL은 데이터가 앞에서 뒤로 이동하는 흐름을 표현하기 좋습니다.

chain = prompt | model | parser

같은 입력을 여러 작업에 보내는 병렬 처리도 자연스럽습니다.

chain = {
    "summary": summary_chain,
    "keywords": keyword_chain,
}

조건 분기도 가능합니다. RunnableBranch를 사용하면 입력이나 중간 결과에 따라 다른 Runnable을 선택할 수 있습니다.

따라서 LangGraph가 필요한 이유를 단순히 "LCEL은 분기를 못 한다"고 설명하면 정확하지 않습니다. 경계는 조건문 하나가 아니라 상태를 가진 반복적인 실행 흐름이 얼마나 복잡해졌는가에 있습니다.

예를 들어 다음 요구가 겹치기 시작하면 그래프가 더 잘 맞습니다.

• 평가 결과가 부족하면 생성 단계로 돌아가야 한다.
• 여러 단계가 같은 messages, draft, score를 읽고 수정해야 한다.
• 도구 호출 결과에 따라 다음 경로가 계속 달라진다.
• 실행을 저장해 두고 나중에 같은 지점에서 이어야 한다.

이런 요구에서는 "다음에 무엇을 실행할지"가 애플리케이션의 중요한 로직이 됩니다.

---

2장. 그래프에서는 흐름을 밖으로 꺼낸다

LangGraph의 Graph API는 실행 흐름을 크게 세 요소로 나눕니다.

요소	역할
State	여러 단계가 공유하는 현재 데이터
Node	실제 작업을 수행하고 State의 일부를 갱신하는 함수
Edge	다음에 실행할 Node를 정하는 연결

예를 들어 생성한 답을 평가하고 점수가 낮으면 다시 생성하는 흐름을 생각해 보겠습니다.

generate
   ↓
evaluate
   ├─ 점수 충분 → END
   └─ 점수 부족 → generate

파이프라인에서는 뒤 단계에서 앞 단계로 돌아가는 흐름이 구조 안에 잘 드러나지 않습니다. 그래프에서는 evaluate에서 다시 generate로 연결하는 Edge가 곧 반복 규칙이 됩니다.

State는 이 반복 동안 유지할 데이터를 담습니다.

from typing_extensions import TypedDict

class State(TypedDict):
    topic: str
    draft: str
    score: float
    iteration: int

각 Node는 State 전체를 읽고, 자신이 바꾼 부분만 반환합니다.

def generate(state: State):
    draft = writer.invoke(state["topic"])
    return {
        "draft": draft,
        "iteration": state["iteration"] + 1,
    }

이 구조를 이해하면 그래프를 별도의 신기한 실행 모델로 보기보다 상태와 다음 경로를 명시적으로 드러낸 워크플로로 볼 수 있습니다.

---

3장. 가장 작은 StateGraph

현재 LangGraph의 Graph API에서 중심 클래스는 StateGraph입니다.

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

class State(TypedDict):
    text: str
    length: int


def measure(state: State):
    return {"length": len(state["text"])}

builder = StateGraph(State)
builder.add_node("measure", measure)
builder.add_edge(START, "measure")
builder.add_edge("measure", END)

graph = builder.compile()

StateGraph 자체는 빌더입니다. Node와 Edge를 등록한 뒤 compile()해야 실행 가능한 그래프가 만들어집니다.

result = graph.invoke({"text": "hello", "length": 0})

컴파일된 그래프는 invoke, stream, ainvoke 같은 Runnable 계열의 실행 방식을 그대로 제공합니다. 앞에서 배운 실행 규약이 그래프에서도 이어지는 셈입니다.

---

4장. 조건과 반복은 Edge가 표현한다

고정된 순서는 일반 Edge로 연결합니다.

builder.add_edge(START, "generate")
builder.add_edge("generate", "evaluate")

다음 경로가 State에 따라 달라지면 조건부 Edge를 사용합니다.

def route(state: State):
    if state["score"] >= 0.8:
        return "finish"
    return "retry"

builder.add_conditional_edges(
    "evaluate",
    route,
    {
        "retry": "generate",
        "finish": END,
    },
)

여기서 중요한 것은 조건문 문법이 아닙니다. route()가 실행 흐름을 결정하는 규칙으로 분리되어 있다는 점입니다.

State
  ↓
evaluate
  ↓
route(State)
  ├─ retry  → generate
  └─ finish → END

반복을 만들 때는 반드시 종료 조건도 함께 설계해야 합니다. 점수, 반복 횟수, 시간 제한처럼 애플리케이션이 멈춰야 할 조건을 State나 실행 설정에 명시적으로 두는 편이 안전합니다.

---

5장. 표준 에이전트는 직접 그리지 않아도 된다

모델이 도구를 선택하고, 도구 결과를 다시 보고, 더 이상 도구 호출이 없을 때 종료하는 루프는 매우 흔합니다.

사용자 입력
   ↓
모델
   ├─ tool call 있음 → 도구 실행 → 모델로 돌아감
   └─ tool call 없음 → 종료

이런 표준 도구 호출 에이전트는 매번 StateGraph를 직접 그릴 필요가 없습니다. LangChain v1의 create_agent()가 이 반복 구조를 LangGraph 런타임 위에 만들어 줍니다.

from langchain.agents import create_agent

agent = create_agent(
    model="openai:gpt-5.5",
    tools=[search],
)

초기의 LangGraph에서는 langgraph.prebuilt.create_react_agent()가 이 역할을 맡았습니다. 현재 이 함수는 deprecated이며, 표준 에이전트는 langchain.agents.create_agent()를 사용하는 방향으로 정리됐습니다.

이 변화가 중요한 이유는 패키지 이름보다 추상화의 경계가 명확해졌기 때문입니다. 표준 에이전트 루프는 상위 LangChain API로 만들고, 흐름 자체가 제품 로직인 경우에는 LangGraph를 직접 사용합니다.

---

6장. 언제 직접 그래프를 그릴까

단일 모델이 도구를 선택하는 전형적인 에이전트라면 create_agent()가 출발점으로 적합합니다. 반대로 다음 단계와 종료 조건 자체가 업무 규칙이라면 StateGraph가 더 잘 맞습니다.

예를 들어 심사 시스템을 생각해 보겠습니다.

접수 → 자동 검토 → 위험도 판단
                  ├─ 낮음 → 승인
                  ├─ 중간 → 사람 검토
                  └─ 높음 → 거절

이 흐름에서는 "모델이 다음 도구를 알아서 고른다"보다 어느 조건에서 어느 단계로 이동하는지가 더 중요합니다. 이런 로직을 Node와 Edge로 직접 표현하면 실행 경로가 코드에 드러납니다.

생성-평가 루프도 마찬가지입니다.

초안 생성 → 평가
            ├─ 통과 → 종료
            └─ 실패 → 수정 → 다시 평가

여러 에이전트를 배치하거나 사람 승인을 특정 단계에 넣는 경우도 직접 그래프가 읽기 쉽습니다.

판단 기준은 간단합니다. 표준 도구 호출 루프를 원하는가, 아니면 흐름 자체가 애플리케이션의 요구사항인가를 보면 됩니다.

---

7장. 상태 저장은 별도의 문제다

그래프를 사용한다고 실행 상태가 자동으로 영구 저장되는 것은 아닙니다. 실행 사이에 State를 이어가거나, 중간에 멈췄다가 재개하려면 checkpointer를 연결합니다.

from langgraph.checkpoint.memory import InMemorySaver

graph = builder.compile(checkpointer=InMemorySaver())

실행할 때는 thread_id를 사용해 같은 실행 흐름을 구분합니다.

config = {"configurable": {"thread_id": "conversation-1"}}
graph.invoke(inputs, config)

여기서는 State와 흐름을 저장할 수 있다는 것만 잡으면 충분합니다. 체크포인트가 무엇을 저장하고, thread_id를 어떻게 설계하며, 중단과 재개가 어떻게 동작하는지는 LangGraph (3)에서 따로 다룹니다.

---

8장. 정리

LCEL과 LangGraph는 서로 대체 관계라기보다 서로 다른 크기의 문제를 맡습니다. 프롬프트, 모델, 파서, 검색기를 한 방향으로 연결하는 데이터 흐름은 LCEL이 간결합니다. 실행 중 상태가 계속 변하고, 그 상태에 따라 반복·분기·중단이 결정되면 그래프가 더 명확합니다.

LangGraph를 사용하더라도 Node 안에서는 기존 Runnable 체인을 그대로 사용할 수 있습니다.

def summarize(state: State):
    summary = summary_chain.invoke({"text": state["text"]})
    return {"summary": summary}

따라서 앞에서 배운 체인을 버리고 새로운 방식으로 갈아타는 것이 아닙니다. 작은 실행 단위는 Runnable로 만들고, 그 단위들의 상태 있는 흐름을 그래프로 묶는 것이 자연스러운 연결입니다.

다음 LangGraph 시리즈에서는 이 편에서 맛본 State, Node, Edge를 더 세밀하게 다룹니다.

---

참고자료

• LangChain 공식 문서, Agents
• LangChain Reference, create_agent
• LangGraph 공식 문서, Graph API
• LangGraph Reference, StateGraph
• LangGraph Reference, create_react_agent (deprecated)
• LangChain PyPI