LCEL로 체인 구성하기

지난 장에서 우리는 프롬프트 템플릿으로 입력을 정돈하고, 출력 파서로 모델의 응답을 애플리케이션이 쓰기 좋은 형태로 바꾸는 법을 배웠습니다. 이제 그 조각들을 하나의 흐름으로 연결할 차례입니다.

LangChain에서 이 연결 방식을 가장 간결하게 표현하는 도구가 LCEL, 즉 LangChain Expression Language입니다. 이름은 조금 거창하지만, 핵심은 단순합니다.

입력을 받아 처리하는 작은 부품들을 | 연산자로 이어 붙여 하나의 실행 가능한 파이프라인으로 만든다.

예를 들어 다음 한 줄은 LangChain 앱의 가장 기본적인 골격입니다.

chain = prompt | model | parser

처음 보면 "파이썬에서 이런 문법을 써도 되나?" 싶을 수 있습니다. 하지만 이 한 줄 안에는 프롬프트 생성, 모델 호출, 응답 파싱이라는 LLM 애플리케이션의 기본 흐름이 모두 들어 있습니다. LCEL은 LangChain이 체인을 선언적으로 조합하기 위해 도입한 방식이며, 이렇게 만든 체인은 기본적으로 단일 실행, 스트리밍, 배치, 비동기 실행을 같은 인터페이스로 다룰 수 있습니다.¹

LCEL은 Runnable을 연결하는 언어다

LCEL을 이해하려면 먼저 Runnable이라는 단어와 친해져야 합니다. Runnable은 말 그대로 "실행할 수 있는 것"입니다. 어떤 입력을 받아 어떤 출력을 내는 객체라면 Runnable처럼 다룰 수 있습니다.

LangChain에서는 프롬프트 템플릿, 채팅 모델, 출력 파서, 검색기, 직접 만든 함수까지 다양한 구성 요소가 Runnable 인터페이스를 따릅니다. 그래서 서로 종류가 달라도 같은 방식으로 실행하고, 같은 방식으로 연결할 수 있습니다.

다음 세 가지는 모두 Runnable입니다.

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 어려운 기술 개념을 쉽게 설명하는 튜터입니다."),
    ("human", "{topic}을 초보자에게 설명해 주세요."),
])

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

prompt는 입력 딕셔너리를 받아 메시지 목록으로 바꿉니다. model은 메시지를 받아 AI 응답 메시지를 만듭니다. parser는 AI 응답 메시지에서 실제 텍스트만 꺼냅니다. 역할은 서로 다르지만 모두 "입력 → 출력"이라는 공통 구조를 갖습니다.

이제 이들을 연결하면 하나의 체인이 됩니다.

chain = prompt | model | parser

이 체인도 다시 Runnable입니다. 그래서 다음처럼 실행할 수 있습니다.

result = chain.invoke({"topic": "LCEL"})
print(result)

여기서 중요한 점은 prompt | model | parser가 단순히 보기 좋은 축약 표현이 아니라는 것입니다. 왼쪽 컴포넌트의 출력이 오른쪽 컴포넌트의 입력으로 전달되는 실행 흐름이 만들어집니다. LangChain 내부에서는 이런 순차 연결이 Runnable 시퀀스로 구성됩니다.

prompt | model | parser를 단계별로 추적하기

LCEL을 처음 배울 때는 한 번쯤 파이프라인을 손으로 따라가 보는 것이 좋습니다. 체인이 마법처럼 동작하는 것처럼 보이지만, 실제로는 각 단계의 입출력이 차례로 이어질 뿐입니다.

먼저 전체 코드를 보겠습니다.

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 개발자를 위한 친절한 기술 코치입니다."),
    ("human", "{concept}을 한 문단으로 설명하고, 마지막에 짧은 비유를 덧붙여 주세요."),
])

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

chain = prompt | model | parser

answer = chain.invoke({"concept": "LangChain Expression Language"})
print(answer)

이제 체인 전체가 아니라 각 단계를 따로 실행해 보겠습니다.

input_data = {
    "concept": "LangChain Expression Language"
}

prompt_value = prompt.invoke(input_data)
print(prompt_value)

prompt.invoke()의 출력은 단순 문자열이 아니라 모델에 전달할 수 있는 프롬프트 값입니다. 채팅 모델을 사용할 때는 보통 시스템 메시지와 사용자 메시지 형태로 정리됩니다.

개념적으로는 다음과 비슷한 구조가 만들어집니다.

System: 당신은 개발자를 위한 친절한 기술 코치입니다.
Human: LangChain Expression Language을 한 문단으로 설명하고, 마지막에 짧은 비유를 덧붙여 주세요.

이 결과가 다음 단계인 모델의 입력이 됩니다.

ai_message = model.invoke(prompt_value)
print(type(ai_message))
print(ai_message.content)

채팅 모델의 출력은 보통 AIMessage입니다. 여기에는 실제 답변인 content뿐 아니라 모델 이름, 토큰 사용량, 응답 메타데이터 같은 정보가 함께 들어갈 수 있습니다.

마지막으로 파서가 이 메시지를 애플리케이션에서 바로 쓰기 쉬운 문자열로 바꿉니다.

parsed = parser.invoke(ai_message)
print(type(parsed))
print(parsed)

결국 다음 두 코드는 같은 흐름을 표현합니다.

# 단계별 실행
prompt_value = prompt.invoke(input_data)
ai_message = model.invoke(prompt_value)
parsed = parser.invoke(ai_message)

# LCEL 실행
parsed = (prompt | model | parser).invoke(input_data)

LCEL의 장점은 이 흐름을 한눈에 읽을 수 있게 만든다는 데 있습니다. 왼쪽에서 오른쪽으로 읽으면 데이터가 어디서 만들어지고, 어디서 변환되고, 어디서 최종 형태가 되는지 자연스럽게 보입니다.

파이프 연산자는 Unix 파이프처럼 읽는다

LCEL의 | 연산자는 Unix 셸의 파이프를 떠올리면 이해하기 쉽습니다. Unix에서 다음 명령은 앞 명령의 출력을 뒤 명령의 입력으로 넘깁니다.

cat app.log | grep ERROR | sort

LCEL도 같은 느낌입니다.

chain = prompt | model | parser

prompt가 만든 메시지를 model이 받고, model이 만든 응답을 parser가 받습니다. 파이썬 입장에서는 | 연산자가 객체의 특수 메서드를 통해 동작하지만, 개발자가 기억해야 할 감각은 간단합니다. "왼쪽 결과가 오른쪽으로 흐른다"입니다. LCEL을 설명하는 여러 자료에서도 이 파이프 연산자를 Runnable을 이어 붙이는 핵심 문법으로 다룹니다.²

과거 LangChain에서는 LLMChain, SequentialChain 같은 클래스를 사용해 체인을 만들었습니다. 간단한 예제에서는 잘 동작했지만, 체인이 복잡해질수록 문제가 생겼습니다. 어떤 체인은 스트리밍을 지원하고, 어떤 체인은 배치 실행이 어색하며, 여러 체인을 합칠 때 입력 키와 출력 키를 맞추느라 코드가 장황해지곤 했습니다.

LCEL은 이 문제를 Runnable이라는 공통 인터페이스로 정리합니다. 프롬프트도 Runnable, 모델도 Runnable, 파서도 Runnable, 검색기도 Runnable, 사용자 정의 함수도 Runnable로 다룰 수 있습니다. 덕분에 체인을 만들 때 "이 클래스는 어떤 실행 메서드를 쓰지?"라고 고민하는 시간이 줄어듭니다.

LCEL 체인의 네 가지 실행 방식

LCEL로 만든 체인은 같은 객체를 여러 방식으로 실행할 수 있습니다. 가장 자주 쓰는 것은 invoke, stream, batch, 비동기 메서드입니다.

chain = prompt | model | parser

이 체인을 기준으로 하나씩 살펴보겠습니다.

invoke: 단일 입력을 처리한다

invoke는 가장 기본적인 실행 방식입니다. 하나의 입력을 넣고 하나의 출력을 받습니다.

result = chain.invoke({
    "concept": "Runnable"
})

print(result)

API 서버에서 사용자의 질문 하나를 받아 답변 하나를 반환하는 상황이라면 invoke만으로 충분합니다. 노트북에서 실험하거나, 체인의 각 단계를 테스트할 때도 가장 먼저 사용하게 됩니다.

stream: 응답을 조금씩 보여준다

LLM 응답이 길어질 때 사용자는 빈 화면을 오래 보고 싶어 하지 않습니다. stream을 사용하면 모델이 토큰 또는 청크를 생성하는 대로 화면에 표시할 수 있습니다.

for chunk in chain.stream({"concept": "LCEL 스트리밍"}):
    print(chunk, end="", flush=True)

사용자 경험은 크게 달라집니다. 전체 답변이 완성될 때까지 기다린 뒤 한 번에 보여주는 대신, 답변이 생성되는 과정을 실시간으로 보여줄 수 있습니다. 챗봇, 문서 작성 도우미, 코드 생성기처럼 응답이 긴 앱에서는 스트리밍이 거의 필수에 가깝습니다.

다만 모든 컴포넌트가 스트리밍에 똑같이 유리한 것은 아닙니다. 예를 들어 문자열 파서는 청크를 그대로 흘려보내기 쉽지만, 완성된 JSON 전체를 검증해야 하는 파서는 최종 결과가 나올 때까지 기다려야 할 수 있습니다. 스트리밍 UX를 설계할 때는 체인 끝의 파서가 어떤 방식으로 동작하는지도 함께 고려해야 합니다.

batch: 여러 입력을 한꺼번에 처리한다

batch는 여러 입력을 리스트로 넘기고 결과 리스트를 받습니다.

inputs = [
    {"concept": "프롬프트 템플릿"},
    {"concept": "출력 파서"},
    {"concept": "RunnableParallel"},
]

results = chain.batch(inputs)

for result in results:
    print("---")
    print(result)

게시글 100개의 요약을 만들거나, 고객 리뷰 1,000개의 감성을 분류하거나, FAQ 후보 질문을 대량 생성하는 작업에서는 batch가 편리합니다. 내부적으로 병렬 처리나 제공자별 최적화가 적용될 수 있기 때문에 단순히 for문으로 하나씩 호출하는 것보다 처리량을 높이기 좋습니다.

물론 배치 크기는 신중하게 정해야 합니다. 너무 큰 배치를 한 번에 보내면 모델 제공자의 rate limit에 걸리거나, 실패했을 때 재처리 범위가 커질 수 있습니다. 실무에서는 작은 배치로 시작한 뒤 지연 시간, 비용, 오류율을 보며 조정하는 편이 안전합니다.

ainvoke, astream, abatch: 기다리는 시간을 낭비하지 않는다

웹 서버에서는 한 사용자의 모델 응답을 기다리는 동안 다른 요청도 처리해야 합니다. 이때 비동기 실행이 중요해집니다.

result = await chain.ainvoke({
    "concept": "비동기 LCEL"
})

비동기 스트리밍도 가능합니다.

async for chunk in chain.astream({"concept": "비동기 스트리밍"}):
    print(chunk, end="", flush=True)

비동기 방식은 모델 호출, 검색기 호출, 외부 API 호출처럼 I/O 대기 시간이 큰 작업에서 특히 효과적입니다. CPU가 무거운 계산을 직접 수행하는 것이 아니라 네트워크 응답을 기다리는 상황이라면, 비동기 실행으로 서버 자원을 더 효율적으로 사용할 수 있습니다.

정리하면 invoke는 단일 요청, stream은 실시간 사용자 경험, batch는 대량 처리, 비동기 메서드는 동시성이 중요한 서버 환경에 어울립니다.

이전 장의 구조화 파싱 예제를 LCEL로 완성하기

지난 장에서는 제품 리뷰에서 액션 아이템을 추출하는 예제를 다루었습니다. 이번에는 그 흐름을 LCEL로 완성해 보겠습니다. 이 예제는 프롬프트, 모델, Pydantic 기반 출력 파서가 어떻게 한 줄의 체인으로 이어지는지 보여줍니다.

from typing import List
from pydantic import BaseModel, Field

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

class ActionItem(BaseModel):
    owner: str = Field(description="담당자. 알 수 없으면 '미정'")
    task: str = Field(description="수행해야 할 작업")
    priority: str = Field(description="high, medium, low 중 하나")

class ReviewActions(BaseModel):
    summary: str = Field(description="리뷰의 핵심 요약")
    action_items: List[ActionItem] = Field(description="실행해야 할 작업 목록")

parser = PydanticOutputParser(pydantic_object=ReviewActions)

prompt = ChatPromptTemplate.from_template("""
다음 제품 리뷰를 읽고 개선 액션 아이템을 추출하세요.

리뷰:
{review}

반드시 아래 형식 지침을 따르세요.
{format_instructions}
""").partial(
    format_instructions=parser.get_format_instructions()
)

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

chain = prompt | model | parser

result = chain.invoke({
    "review": "앱 디자인은 깔끔하지만 결제 버튼을 찾기 어렵습니다. 배송 알림도 하루 늦게 왔어요. 고객센터 답변은 친절했습니다."
})

print(result.summary)
for item in result.action_items:
    print(item.owner, item.task, item.priority)

이 체인의 출력은 단순 문자열이 아니라 ReviewActions 객체입니다. 즉, 애플리케이션 코드에서는 result.action_items[0].priority처럼 타입이 있는 필드로 접근할 수 있습니다.

다만 최신 모델 제공자 중 일부는 LangChain의 with_structured_output()을 통해 모델 API 수준에서 구조화 출력을 강제할 수 있습니다. 이 방식은 프롬프트에 형식 지침을 넣고 파서가 후처리하는 방식보다 안정적일 때가 많습니다. 그래도 PydanticOutputParser는 LCEL 흐름에서 "프롬프트가 형식을 지시하고, 모델이 답하고, 파서가 검증한다"는 구조를 이해하기에 매우 좋은 예제입니다.

입력을 여러 갈래로 나누는 RunnableParallel

지금까지의 체인은 일직선이었습니다.

입력 → 프롬프트 → 모델 → 파서 → 출력

하지만 실제 앱에서는 하나의 입력을 여러 방식으로 동시에 처리해야 할 때가 많습니다. 예를 들어 긴 고객 문의를 받았다고 해 봅시다. 우리는 이 문의를 요약하고, 감성을 분류하고, 키워드를 뽑고 싶습니다. 순서대로 세 번 호출할 수도 있지만, 서로 의존하지 않는 작업이라면 동시에 실행하는 편이 낫습니다.

이때 RunnableParallel을 사용합니다.

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

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

summary_chain = (
    ChatPromptTemplate.from_template("다음 글을 두 문장으로 요약하세요:nn{text}")
    | model
    | parser
)

sentiment_chain = (
    ChatPromptTemplate.from_template("다음 글의 감성을 positive, neutral, negative 중 하나로 분류하세요:nn{text}")
    | model
    | parser
)

keywords_chain = (
    ChatPromptTemplate.from_template("다음 글에서 핵심 키워드 5개를 쉼표로 구분해 추출하세요:nn{text}")
    | model
    | parser
)

analysis_chain = RunnableParallel(
    summary=summary_chain,
    sentiment=sentiment_chain,
    keywords=keywords_chain,
)

result = analysis_chain.invoke({
    "text": "결제는 빨랐지만 배송이 늦었고, 고객센터는 문제를 빠르게 해결해 주었습니다."
})

print(result)

결과는 딕셔너리 형태입니다.

{
    "summary": "...",
    "sentiment": "...",
    "keywords": "..."
}

세 작업은 같은 입력을 받지만 서로 다른 체인을 통과합니다. 병렬 실행이 가능한 구조에서는 전체 지연 시간이 각 작업 시간의 합이 아니라 가장 오래 걸린 작업에 가까워질 수 있습니다. RAG에서도 이 패턴은 자주 등장합니다. 같은 질문을 벡터 검색, 키워드 검색, 데이터베이스 조회로 동시에 보내고, 결과를 합쳐 최종 답변을 만드는 식입니다.³

딕셔너리 매핑은 RunnableParallel의 짧은 표현이다

LCEL에서는 딕셔너리를 파이프 안에 넣는 방식도 자주 사용합니다. 다음 코드는 RAG 체인에서 특히 많이 볼 수 있는 모양입니다.

from langchain_core.runnables import RunnablePassthrough

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

여기서 딕셔너리는 사실상 병렬 매핑입니다. 같은 입력이 context 경로와 question 경로로 나뉩니다. context 경로에서는 검색기가 관련 문서를 찾고, question 경로에서는 원래 질문을 그대로 보존합니다.

예를 들어 사용자가 다음 질문을 입력했다고 합시다.

"환불 정책은 어떻게 되나요?"

체인의 첫 단계는 개념적으로 다음 딕셔너리를 만듭니다.

{
    "context": [검색된 문서들],
    "question": "환불 정책은 어떻게 되나요?"
}

이 딕셔너리가 다음 프롬프트의 입력이 됩니다.

prompt = ChatPromptTemplate.from_template("""
다음 문맥만 사용해 질문에 답하세요.

문맥:
{context}

질문:
{question}
""")

이 패턴은 뒤에서 RAG를 만들 때 계속 보게 됩니다. 지금은 RunnablePassthrough()가 "원본 입력을 잃어버리지 않게 옆에 붙여 두는 도구"라는 점만 기억하면 충분합니다.

RunnablePassthrough로 원본 입력과 결과를 함께 들고 가기

체인을 만들다 보면 모델의 결과뿐 아니라 원본 입력도 함께 다음 단계로 넘기고 싶을 때가 있습니다. 예를 들어 사용자의 문장을 번역한 뒤, 원문과 번역문을 함께 비교 프롬프트에 넣고 싶을 수 있습니다.

from langchain_core.runnables import RunnablePassthrough

translate_prompt = ChatPromptTemplate.from_template(
    "다음 문장을 영어로 번역하세요:nn{text}"
)

translate_chain = translate_prompt | model | parser

chain = RunnableParallel(
    original=RunnablePassthrough(),
    translated=translate_chain,
)

result = chain.invoke({
    "text": "LCEL을 사용하면 체인을 읽기 쉽게 만들 수 있습니다."
})

print(result["original"])
print(result["translated"])

주의할 점이 하나 있습니다. 이 예제에서 RunnablePassthrough()는 입력 딕셔너리 전체를 그대로 보존합니다. 만약 원본 문장 문자열만 꺼내고 싶다면 뒤에서 볼 RunnableLambda로 형태를 정리하면 됩니다.

lambda 변환으로 데이터 모양 맞추기

LCEL을 쓰다 보면 대부분의 어려움은 "모델이 어렵다"가 아니라 "다음 단계가 원하는 입력 모양과 지금 출력 모양이 다르다"에서 생깁니다. 이때 작은 변환 함수를 체인 중간에 넣으면 흐름이 깔끔해집니다.

LangChain에서는 일반 파이썬 함수를 Runnable로 감싸기 위해 RunnableLambda를 사용합니다.

from langchain_core.runnables import RunnableLambda

pick_text = RunnableLambda(lambda x: x["text"])

result = pick_text.invoke({"text": "안녕하세요", "user_id": 123})
print(result)

이제 원본 문장만 보존하고 번역 결과와 합쳐 보겠습니다.

from langchain_core.runnables import RunnableParallel, RunnableLambda

pick_text = RunnableLambda(lambda x: x["text"])

translation_bundle = RunnableParallel(
    original=pick_text,
    translated=translate_chain,
)

result = translation_bundle.invoke({
    "text": "프롬프트와 모델과 파서를 연결합니다."
})

print(result)

RunnableLambda는 아주 강력하지만 남용하면 체인의 가독성이 떨어질 수 있습니다. 한 줄짜리 키 추출이나 간단한 포맷팅에는 좋지만, 복잡한 비즈니스 로직은 이름 있는 함수로 빼는 편이 낫습니다.

def format_translation_pair(data: dict) -> str:
    return f"원문: {data['original']}n번역: {data['translated']}"

format_pair = RunnableLambda(format_translation_pair)

final_chain = translation_bundle | format_pair

print(final_chain.invoke({
    "text": "복잡한 흐름도 작은 조각으로 나누면 테스트하기 쉽습니다."
}))

작은 변환 함수를 잘 배치하면 LCEL 체인의 각 단계가 명확해집니다. 입력 모양을 바꾸고, 필요한 키를 뽑고, 다음 프롬프트가 원하는 형태로 재구성할 수 있습니다.

요약 후 번역: 다단계 체인의 첫 번째 예

이제 여러 단계를 가진 워크플로를 만들어 보겠습니다. 첫 번째 예제는 "긴 글을 요약한 뒤 영어로 번역하기"입니다.

summary_prompt = ChatPromptTemplate.from_template("""
다음 글을 한국어로 3문장 이내로 요약하세요.

글:
{text}
""")

summary_chain = summary_prompt | model | parser

translate_prompt = ChatPromptTemplate.from_template("""
다음 한국어 요약문을 자연스러운 영어로 번역하세요.

요약문:
{summary}
""")

translate_chain = translate_prompt | model | parser

여기서 summary_chain의 출력은 문자열입니다. 그런데 translate_prompt는 {summary}라는 키를 가진 딕셔너리를 기대합니다. 따라서 중간에 문자열을 딕셔너리로 바꿔 주어야 합니다.

wrap_summary = RunnableLambda(lambda summary: {"summary": summary})

summary_then_translate = summary_chain | wrap_summary | translate_chain

result = summary_then_translate.invoke({
    "text": "LangChain은 LLM 애플리케이션을 구성하기 위한 다양한 추상화를 제공한다. LCEL은 그중 체인을 선언적으로 조합하는 방식이며, 프롬프트, 모델, 파서, 검색기 등을 Runnable로 연결할 수 있게 해 준다. 이를 통해 개발자는 복잡한 흐름을 작은 단위로 나누고, 각 단계를 독립적으로 테스트할 수 있다."
})

print(result)

이 예제에서 중요한 것은 체인을 작은 단위로 분리했다는 점입니다. summary_chain만 따로 실행해 요약 품질을 확인할 수 있고, translate_chain만 따로 실행해 번역 품질을 점검할 수도 있습니다. 전체 체인이 잘못된 답을 냈을 때 어느 단계가 문제인지 찾기 쉬워집니다.

의도 분류 후 라우팅하기

두 번째 예제는 사용자 문의의 의도를 분류한 뒤 다른 체인으로 보내는 흐름입니다. 쇼핑몰 고객센터 챗봇을 만든다고 생각해 봅시다. 사용자의 질문은 배송, 환불, 제품 정보, 기타 문의로 나뉠 수 있습니다.

먼저 의도 분류 체인을 만듭니다.

intent_prompt = ChatPromptTemplate.from_template("""
다음 사용자 문의의 의도를 하나로 분류하세요.
가능한 값: shipping, refund, product, other

문의:
{question}

분류값만 출력하세요.
""")

intent_chain = intent_prompt | model | parser

다음으로 각 의도별 답변 체인을 만듭니다.

shipping_chain = (
    ChatPromptTemplate.from_template("배송 문의에 친절히 답하세요. 문의: {question}")
    | model
    | parser
)

refund_chain = (
    ChatPromptTemplate.from_template("환불 정책을 기준으로 답하세요. 문의: {question}")
    | model
    | parser
)

product_chain = (
    ChatPromptTemplate.from_template("제품 정보를 설명하세요. 문의: {question}")
    | model
    | parser
)

other_chain = (
    ChatPromptTemplate.from_template("일반 고객센터 상담원처럼 답하세요. 문의: {question}")
    | model
    | parser
)

라우팅은 파이썬 함수로 명확하게 작성할 수 있습니다.

def route_by_intent(data: dict):
    intent = data["intent"].strip().lower()

    if intent == "shipping":
        return shipping_chain
    if intent == "refund":
        return refund_chain
    if intent == "product":
        return product_chain
    return other_chain

이제 원래 질문과 분류 결과를 함께 만든 뒤, 적절한 체인을 선택해 실행합니다.

from langchain_core.runnables import RunnableLambda, RunnableParallel, RunnablePassthrough

prepare_route = RunnableParallel(
    question=RunnableLambda(lambda x: x["question"]),
    intent=intent_chain,
)

router = RunnableLambda(lambda data: route_by_intent(data).invoke({"question": data["question"]}))

support_chain = prepare_route | router

answer = support_chain.invoke({
    "question": "주문한 상품이 아직 도착하지 않았는데 배송 조회는 어디서 하나요?"
})

print(answer)

이 코드는 의도적으로 단순하게 작성했습니다. 실무에서는 RunnableBranch 같은 분기 전용 도구를 사용할 수도 있고, 라우팅 결과를 LangSmith로 추적하며 분류 정확도를 평가할 수도 있습니다. 핵심은 "분류 체인"과 "답변 체인"을 분리했다는 점입니다. 분류가 잘못되면 분류 체인만 개선하면 되고, 배송 답변 톤이 어색하면 배송 체인만 수정하면 됩니다.

초안 작성 후 검토하기

세 번째 예제는 LLM을 한 번만 믿지 않는 패턴입니다. 먼저 초안을 작성하고, 다음 단계에서 검토한 뒤, 필요하면 개선안을 만듭니다. LLM은 훌륭한 초안을 만들 수 있지만, 한 번의 응답이 항상 최선은 아닙니다. 그래서 여러 호출을 작은 단계로 나누면 품질을 높일 수 있습니다.

draft_prompt = ChatPromptTemplate.from_template("""
다음 주제로 블로그 글 도입부 초안을 작성하세요.
독자가 계속 읽고 싶도록 흥미롭게 시작하세요.

주제:
{topic}
""")

draft_chain = draft_prompt | model | parser

검토 체인은 초안과 원래 주제를 함께 받아야 합니다.

review_prompt = ChatPromptTemplate.from_template("""
다음 블로그 도입부 초안을 검토하세요.

주제:
{topic}

초안:
{draft}

검토 기준:
- 첫 문장이 독자의 관심을 끄는가
- 주제가 분명한가
- 과장된 표현이 없는가

개선할 점을 구체적으로 제안하세요.
""")

review_chain = review_prompt | model | parser

초안과 원래 주제를 함께 보존하기 위해 병렬 매핑을 사용합니다.

make_draft_bundle = RunnableParallel(
    topic=RunnableLambda(lambda x: x["topic"]),
    draft=draft_chain,
)

review_workflow = make_draft_bundle | RunnableParallel(
    topic=RunnableLambda(lambda x: x["topic"]),
    draft=RunnableLambda(lambda x: x["draft"]),
    review=review_chain,
)

result = review_workflow.invoke({
    "topic": "개발자가 LCEL을 배워야 하는 이유"
})

print("초안:")
print(result["draft"])
print("n검토:")
print(result["review"])

여기서 한 단계 더 나아가 개선 체인을 붙일 수도 있습니다.

revise_prompt = ChatPromptTemplate.from_template("""
아래 초안과 검토 의견을 바탕으로 블로그 도입부를 개선하세요.

초안:
{draft}

검토 의견:
{review}

개선된 도입부만 출력하세요.
""")

revise_chain = revise_prompt | model | parser

full_writing_workflow = review_workflow | revise_chain

final_intro = full_writing_workflow.invoke({
    "topic": "개발자가 LCEL을 배워야 하는 이유"
})

print(final_intro)

이런 워크플로는 글쓰기뿐 아니라 코드 생성, 정책 답변, 보고서 작성, SQL 생성에도 적용할 수 있습니다. "생성 → 검토 → 개선"은 LLM 애플리케이션에서 매우 강력한 기본 패턴입니다.

각 단계를 독립적으로 테스트하기

LCEL 체인의 가장 큰 장점 중 하나는 복잡한 흐름을 작은 조각으로 나누고, 각 조각을 따로 실행할 수 있다는 점입니다.

요약 후 번역 체인을 다시 떠올려 봅시다.

summary_chain = summary_prompt | model | parser
translate_chain = translate_prompt | model | parser
summary_then_translate = summary_chain | wrap_summary | translate_chain

이 구조에서는 다음 테스트가 모두 가능합니다.

# 요약만 테스트
summary = summary_chain.invoke({"text": long_text})
print(summary)

# 번역만 테스트
translation = translate_chain.invoke({"summary": "LCEL은 체인을 쉽게 조합하는 방식입니다."})
print(translation)

# 중간 변환만 테스트
wrapped = wrap_summary.invoke("LCEL은 Runnable을 연결합니다.")
print(wrapped)

# 전체 흐름 테스트
final = summary_then_translate.invoke({"text": long_text})
print(final)

LLM 앱은 일반적인 함수보다 디버깅이 어렵습니다. 출력이 매번 조금씩 달라질 수 있고, 실패 원인이 프롬프트인지, 모델 설정인지, 파서인지, 입력 데이터인지 헷갈릴 때가 많습니다. LCEL은 체인을 잘게 쪼갤 수 있게 해 주므로 문제를 좁혀 가기 쉽습니다.

실무에서는 각 체인의 입력과 출력을 예제 데이터로 고정해 두고 회귀 테스트처럼 활용할 수 있습니다. "이 프롬프트는 최소한 이런 구조의 답을 내야 한다", "이 파서는 이 JSON을 반드시 객체로 바꿔야 한다", "이 라우터는 배송 문의를 shipping으로 보내야 한다" 같은 기준을 만들면 체인의 유지보수성이 크게 올라갑니다.

LCEL을 쓸 때 자주 생기는 실수

LCEL은 간결하지만, 데이터 흐름을 대충 생각하면 오류가 납니다. 특히 입출력 모양을 놓치면 체인이 갑자기 이해하기 어려워집니다.

자주 만나는 실수는 다음과 같습니다.

• 프롬프트가 {question}을 요구하는데 체인 입력에는 {"input": ...}을 넘긴다.

• 앞 단계가 문자열을 출력하는데 다음 단계는 딕셔너리를 기대한다.

• RunnablePassthrough()가 문자열 하나가 아니라 입력 딕셔너리 전체를 넘긴다는 점을 잊는다.

• 병렬 체인에서 여러 모델 호출이 동시에 발생해 비용과 rate limit이 늘어나는 것을 고려하지 않는다.

• 구조화 파서를 붙여 놓고 프롬프트에 형식 지침을 넣지 않아 파싱 실패가 발생한다.

해결 방법은 단순합니다. 체인을 만들 때 각 단계 옆에 입출력 모양을 주석으로 적어 보세요.

# 입력: {"text": str}
# 출력: str
summary_chain = summary_prompt | model | parser

# 입력: str
# 출력: {"summary": str}
wrap_summary = RunnableLambda(lambda summary: {"summary": summary})

# 입력: {"summary": str}
# 출력: str
translate_chain = translate_prompt | model | parser

# 입력: {"text": str}
# 출력: str
summary_then_translate = summary_chain | wrap_summary | translate_chain

이렇게만 해도 체인 오류의 절반 이상은 줄어듭니다. LCEL을 잘 쓰는 개발자는 파이프를 많이 쓰는 개발자가 아니라, 각 파이프 사이에서 데이터가 어떤 모양으로 흐르는지 정확히 아는 개발자입니다.

LCEL은 복잡함을 숨기지 않고 정리한다

좋은 추상화는 복잡함을 없애는 것이 아니라, 복잡함을 다룰 수 있는 모양으로 정리합니다. LCEL이 바로 그렇습니다.

LLM 애플리케이션은 결국 여러 단계를 거칩니다. 사용자 입력을 정리하고, 필요한 정보를 검색하고, 프롬프트를 만들고, 모델을 호출하고, 결과를 파싱하고, 다시 검토하거나 다음 도구로 넘깁니다. 이 모든 과정을 거대한 함수 하나에 몰아넣으면 처음에는 빠르게 만들 수 있지만, 조금만 커져도 수정하기 어려운 코드가 됩니다.

LCEL은 이 흐름을 작은 Runnable 조각으로 나눕니다. 그리고 | 연산자로 조각들을 읽기 쉬운 방향으로 연결합니다. 단일 요청은 invoke, 실시간 응답은 stream, 대량 처리는 batch, 서버 동시성은 비동기 메서드로 처리합니다. 하나의 입력을 여러 경로로 나눌 때는 RunnableParallel, 원본을 보존할 때는 RunnablePassthrough, 데이터 모양을 바꿀 때는 RunnableLambda를 사용합니다.

이 장에서 익힌 감각은 뒤의 모든 장에서 반복됩니다. 대화 메모리도, 문서 검색도, RAG도, 도구 호출도, 에이전트도 결국 Runnable 조각을 어떻게 연결하느냐의 문제로 돌아옵니다. LCEL을 읽을 수 있게 되면 LangChain 코드는 더 이상 낯선 마법 주문이 아니라, 왼쪽에서 오른쪽으로 흐르는 데이터 파이프라인으로 보이기 시작합니다.

다음 장에서는 이 파이프라인에 시간의 흐름을 더합니다. 사용자의 이전 발화를 기억하고, 대화 상태를 관리하며, "한 번의 질문과 답변"을 넘어서는 LangChain 애플리케이션을 만들어 보겠습니다.

참고

¹LangChain Expression Language

²LangChain Expression Language Explained

³LangChain LCEL Tutorial — Build Chains the Modern Way (2026)