LangChain (4) - 조합하기

LangChain Runnable과 LCEL

프롬프트로 입력을 만들고, 모델에 전달하고, 나온 결과를 다시 파싱하려면 여러 컴포넌트를 순서대로 연결해야 합니다. 각각을 직접 함수 호출로 이어 붙일 수도 있지만, 그렇게 만들면 실행 방식과 오류 처리, streaming 같은 기능을 단계마다 따로 관리해야 합니다.

LangChain은 이 문제를 서로 다른 컴포넌트에 공통된 실행 인터페이스를 부여하고, 그 인터페이스를 만족하는 객체끼리 다시 조합하는 방식으로 해결합니다. 이 공통 추상화가 Runnable입니다. Prompt, model, parser 같은 객체를 같은 실행 단위로 다룰 수 있고, 조합된 결과 역시 다시 하나의 Runnable이 됩니다.

Runnable을 순서대로 연결할 때 가장 자주 사용하는 표현이 | 연산자입니다. 예를 들어 prompt | model | parser라고 작성하면 앞 단계의 출력이 다음 단계의 입력으로 전달되는 RunnableSequence가 만들어집니다. 이렇게 Runnable을 선언적으로 조합하는 방식을 LCEL(LangChain Expression Language)이라고 부릅니다.

중요한 점은 단순히 코드를 짧게 쓰기 위한 문법이 아니라는 것입니다. 조합된 chain도 하나의 Runnable이기 때문에 invoke, batch, stream 같은 공통 실행 방식을 그대로 사용할 수 있고, 필요하면 병렬 실행이나 retry, configuration 같은 동작도 같은 추상화 위에서 붙일 수 있습니다.

Runnable이 어떤 실행 규약을 제공하는지부터 살펴보고, prompt | model | parser가 실제로 어떤 입력과 출력을 주고받는지 따라갑니다. 이어서 직렬 실행인 RunnableSequence와 여러 경로를 동시에 실행하는 RunnableParallel이 어떻게 같은 조합 모델 안에서 동작하는지도 봅니다.

---

1장. 왜 공통 규약이 필요한가

앞의 세 편을 직접 이어 보면 이런 코드가 됩니다.

messages = prompt.format_messages(question=question)
response = model.invoke(messages)
text = response.content

동작 자체에는 문제가 없습니다. 하지만 프롬프트, 모델, 파서, 검색기 같은 부품마다 호출 방식이 다르면 조합이 커질수록 호출 코드를 매번 다시 작성해야 합니다.

LangChain은 이 문제를 모든 실행 가능한 부품을 같은 방식으로 호출한다는 방향으로 풀었습니다.

Prompt
  ↓
Model
  ↓
Parser

각 부품이 같은 규약을 따르면 다음처럼 하나의 실행 단위로 묶을 수 있습니다.

chain = prompt | model | parser

그리고 중요한 점은 합친 chain도 다시 Runnable이라는 것입니다.

Runnable + Runnable + Runnable
            ↓
         Runnable

작은 부품을 묶어 더 큰 부품을 만들고, 그 결과를 다시 다른 조합의 한 단계로 사용할 수 있습니다.

---

2장. Runnable

Runnable은 입력을 받아 출력을 만드는 LangChain의 공통 실행 인터페이스입니다.

대표적인 컴포넌트들이 이 규약을 따릅니다.

컴포넌트	예
프롬프트	ChatPromptTemplate
모델	ChatOpenAI, ChatAnthropic
출력 파서	StrOutputParser
검색기	Retriever 계열
조합된 체인	RunnableSequence, RunnableParallel

일반 파이썬 함수도 Runnable 흐름 안으로 넣을 수 있습니다. 필요하면 RunnableLambda로 명시적으로 감쌀 수 있고, | 연산자나 병렬 조합 안에 callable을 넣으면 LangChain이 Runnable로 자동 변환하기도 합니다.

from langchain_core.runnables import RunnableLambda

normalize = RunnableLambda(lambda text: text.strip())

이렇게 통일해 두면 부품마다 별도의 run(), predict(), __call__() 사용법을 외우는 대신 같은 실행 방식을 사용할 수 있습니다.

여기서는 조합 결과도 다시 Runnable이 된다는 성질을 기억하면 됩니다. 이 덕분에 체인을 계속 계층적으로 조립할 수 있습니다.

---

3장. 공통 실행 메서드

먼저 외울 것은 세 쌍입니다.

invoke   / ainvoke      입력 하나를 실행
batch    / abatch       입력 여러 개를 실행
stream   / astream      결과를 스트림으로 받음

2편에서 ChatPromptTemplate.invoke()를 썼고, 1편에서 ChatModel.invoke()를 썼습니다. 둘이 같은 메서드를 가진 이유가 바로 Runnable 규약입니다.

체인을 만들어도 사용법은 같습니다.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages([
    ("system", "한 문장으로 답하세요."),
    ("human", "{question}"),
])

model = ChatOpenAI(model="gpt-5.5")
parser = StrOutputParser()

chain = prompt | model | parser

한 건을 실행합니다.

chain.invoke({"question": "Runnable이 무엇인가요?"})

여러 입력을 실행합니다.

chain.batch([
    {"question": "Runnable이 무엇인가요?"},
    {"question": "LCEL이 무엇인가요?"},
])

비동기로도 같은 모양입니다.

await chain.ainvoke({"question": "LCEL이 무엇인가요?"})

이 공통 인터페이스가 조합의 기반입니다.

---

4장. 순차 조합

가장 기본적인 조합은 앞의 출력을 다음 입력으로 넘기는 것입니다.

chain = prompt | model | parser

내부적으로는 RunnableSequence가 만들어집니다.

입력 dict
   ↓
ChatPromptTemplate
   ↓ ChatPromptValue
ChatModel
   ↓ AIMessage
StrOutputParser
   ↓ str
최종 출력

각 단계의 출력 타입과 다음 단계의 입력 타입이 연결될 수 있어야 합니다.

3편에서 배운 구조화 출력도 같은 방식으로 연결됩니다.

structured_model = model.with_structured_output(Ticket)
chain = prompt | structured_model

result = chain.invoke({
    "text": "로그인하면 앱이 종료됩니다."
})

이 경우 마지막 출력은 문자열이 아니라 Ticket 같은 구조화된 객체가 됩니다.

즉 LCEL은 StrOutputParser만을 위한 문법이 아닙니다. Runnable이면 같은 방식으로 연결할 수 있다는 것이 핵심입니다.

---

5장. 병렬 조합

두 번째 핵심은 같은 입력을 여러 경로에 동시에 보내는 것입니다.

from langchain_core.runnables import RunnableParallel

parallel = RunnableParallel(
    summary=summary_chain,
    keywords=keyword_chain,
)

result = parallel.invoke({"text": "..."})

결과는 키별로 모입니다.

{
    "summary": "...",
    "keywords": ["...", "..."]
}

LCEL에서는 딕셔너리도 병렬 Runnable로 자동 변환할 수 있습니다.

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

이 구조가 RAG에서 자주 등장합니다.

from langchain_core.runnables import RunnablePassthrough

rag_chain = {
    "context": retriever | format_docs,
    "question": RunnablePassthrough(),
} | prompt | model | StrOutputParser()

입력 질문 하나가 두 갈래로 나뉩니다.

question
   ├─→ retriever → format_docs → context
   │
   └─→ 그대로 전달 ─────────────→ question

                 ↓
               prompt
                 ↓
               model

RunnablePassthrough는 입력을 그대로 다음 단계로 넘길 때 사용합니다.

format_docs처럼 일반 함수를 | 뒤에 직접 넣으면 callable이 Runnable로 변환됩니다. 명시적으로 다루고 싶다면 RunnableLambda(format_docs)라고 써도 됩니다.

---

6장. Runnable에 재시도와 폴백 붙이기

Runnable을 조합하는 이유는 | 문법이 짧아서만은 아닙니다. 같은 실행 규약을 사용하므로 공통 기능을 체인이나 특정 단계에 붙일 수 있습니다.

대표적으로 다음이 있습니다.

비동기 실행
배치 실행
스트리밍
재시도
폴백
실행 설정
추적용 tags / metadata

재시도

reliable_model = model.with_retry(
    stop_after_attempt=3,
)

chain = prompt | reliable_model | StrOutputParser()

재시도는 어느 범위에 붙이는지가 중요합니다.

# 모델 호출만 재시도
prompt | model.with_retry() | parser

# 체인 전체를 재시도
(prompt | model | parser).with_retry()

두 코드는 실패했을 때 다시 실행되는 범위가 다릅니다. 외부 API 호출처럼 실패 가능성이 높은 단계에 좁게 거는 편이 중복 실행을 줄이기 쉽습니다.

폴백

primary = ChatOpenAI(model="gpt-4o")
backup = ChatOpenAI(model="gpt-5.5")

model = primary.with_fallbacks([backup])
chain = prompt | model | StrOutputParser()

기본 모델이 처리하지 못한 예외가 발생하면 순서대로 fallback Runnable을 시도합니다.

실행 설정

chain.invoke(
    {"question": "LCEL이 무엇인가요?"},
    config={
        "tags": ["study"],
        "metadata": {"feature": "qa"},
    },
)

tags와 metadata는 실행 추적과 디버깅에서 구분값으로 사용할 수 있습니다.

---

7장. stream이 있다고 항상 토큰 스트리밍은 아니다

Runnable에는 stream()과 astream() 인터페이스가 있습니다.

for chunk in chain.stream({"question": "긴 설명을 해주세요"}):
    print(chunk, end="", flush=True)

하지만 여기서 중요한 구분이 있습니다.

stream() 메서드가 존재하는 것과 전체 체인이 입력부터 출력까지 점진적으로 스트리밍되는 것은 같은 말이 아닙니다.

RunnableSequence는 각 단계의 스트리밍 성질을 이어받습니다. 중간 컴포넌트가 입력을 전부 받은 뒤에야 결과를 만들 수 있다면 그 지점에서 스트림이 잠시 막힙니다.

스트리밍 가능한 단계
      ↓
스트리밍 가능한 단계
      ↓
        바로 전달 가능

반대로 중간 단계가 전체 결과를 모아야 한다면 다음 단계는 그 작업이 끝난 뒤 시작합니다.

Model stream
    ↓
전체 입력이 필요한 처리
    ↓
여기까지 모은 뒤 다음 출력 시작

특히 일반 RunnableLambda는 기본적으로 스트림 변환을 지원하는 컴포넌트가 아니므로, 스트리밍이 중요한 경로에서는 위치를 신경 써야 합니다.

따라서 실무에서는 단순히 chain.stream()이 호출되는지만 보지 말고 첫 결과가 실제로 언제 나오기 시작하는지 확인해야 합니다.

---

8장. LCEL이 잘하는 것과 그래프로 넘어가는 지점

LCEL의 핵심 조합은 두 가지입니다.

RunnableSequence
    앞 → 뒤로 순차 실행

RunnableParallel
    같은 입력을 여러 경로로 병렬 실행

조건 분기도 RunnableBranch 같은 컴포넌트로 만들 수 있습니다. 따라서 LCEL이 조건 분기를 전혀 못 하는 것은 아닙니다.

문제는 흐름이 복잡해질 때입니다.

결과가 부족하면 이전 단계로 돌아간다
같은 단계를 여러 번 반복한다
여러 단계가 하나의 상태를 계속 갱신한다
중간에 멈추고 나중에 다시 이어간다
실행 상태를 체크포인트로 남긴다

이런 요구가 늘어나면 파이프와 branch를 계속 중첩하는 것보다 State, Node, Edge로 실행 흐름을 명시하는 그래프가 읽기 쉽고 관리하기도 좋습니다.

단순한 데이터 흐름
    → LCEL

상태를 가진 반복·분기·중단·재개
    → LangGraph

LCEL이 사라지는 것은 아닙니다. 이후 LangGraph를 사용할 때도 노드 안에서는 다음처럼 지금 배운 체인을 그대로 사용할 수 있습니다.

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

즉 파이프를 버리고 그래프로 가는 것이 아니라, 파이프를 그래프의 한 노드 안에 넣는 것입니다.

다음 편에서는 이 경계를 넘어 State, Node, Edge로 흐름을 직접 제어합니다.

---

9장. 정리

이번 편의 핵심은 | 문법 자체가 아닙니다. Runnable이라는 공통 실행 규약이 있기 때문에 프롬프트, 모델, 파서, 검색기를 같은 방식으로 호출하고 서로 조합할 수 있다는 점이 중요합니다.

순차 연결은 앞 단계의 출력을 다음 단계로 넘기고, 병렬 연결은 같은 입력을 여러 경로에 보냅니다. 이렇게 조합한 결과도 다시 Runnable이므로 invoke, batch, stream으로 실행하거나 더 큰 체인의 한 부품으로 다시 사용할 수 있습니다.

앞의 세 편을 한 줄로 연결하면 다음과 같습니다.

chain = prompt | model | StrOutputParser()
result = chain.invoke({"question": "LangChain이 무엇인가요?"})

구조화 출력이 필요하면 마지막을 바꿉니다.

chain = prompt | model.with_structured_output(Ticket)

RAG처럼 여러 값을 동시에 만들고 싶으면 병렬 조합을 붙입니다.

chain = {
    "context": retriever | format_docs,
    "question": RunnablePassthrough(),
} | prompt | model | StrOutputParser()

그리고 흐름 자체가 반복되고 상태를 가져야 하기 시작하면 다음 단계는 LangGraph입니다.

---

용어 정리

용어	한 줄 뜻
LCEL	Runnable을 `
Runnable	입력을 받아 출력을 만들고 실행·조합할 수 있는 공통 인터페이스
RunnableSequence	Runnable을 앞에서 뒤로 순차 실행하는 조합
RunnableParallel	같은 입력을 여러 Runnable에 동시에 보내고 결과를 모으는 조합
RunnablePassthrough	입력을 그대로 통과시키는 Runnable
RunnableLambda	일반 파이썬 callable을 Runnable로 감싸는 클래스
RunnableBranch	조건에 따라 실행할 Runnable을 선택하는 컴포넌트
invoke / ainvoke	입력 하나를 동기 / 비동기로 실행
batch / abatch	여러 입력을 동기 / 비동기로 실행
stream / astream	실행 결과를 스트림 인터페이스로 받는 메서드
with_retry	실패한 Runnable을 다시 시도하도록 감싸는 메서드
with_fallbacks	실패하면 다른 Runnable을 순서대로 시도하게 하는 메서드
config	실행 시 tags, metadata 등 런타임 설정을 전달하는 값

---

참고자료

• LangChain Reference, Runnable
• LangChain Reference, RunnableSequence
• LangChain Reference, RunnableParallel
• LangChain Reference, RunnableLambda
• LangChain Reference, RunnablePassthrough
• LangChain Reference, RunnableBranch
• LangChain Reference, with_retry
• LangChain Reference, with_fallbacks
• LangChain PyPI
• langchain-core PyPI