LangChain Expression Language(LCEL) 사용 방법 및 예시

LangChain Expression Language(LCEL)은 LangChain에서 여러 컴포넌트(prompt, LLM, 파서, 후처리 함수 등)를 “파이프(pipe)” 형태로 간결하게 연결해 주는 표현 방식입니다. 2023년 하반기부터 LangChain에서 체인(chain) 구성 방식에서 LCEL을 활용하는 경향이 커지고 있습니다.

 

이번 포스팅에서는 LCEL의 주요 개념과 예제에 대해서 정리해보도록 하겠습니다.

---

주요 개념

• Runnable: 입력(input)을 받아 출력(output)을 내는 구성 요소
• RunnableSequence: 여러 Runnable을 직렬(sequence) 로 연결한 것
• RunnableLambda: 단순한 Python 함수를 Runnable로 감싸서 파이프에서 쓸 수 있게 함
• RunnableParallel: 여러 Runnable을 병렬(parallel)으로 실행하고, 결과를 병합(예: 딕셔너리 구조)해서 넘김
• RunnablePassthrough: 현재 입력 값을 “넘겨주는(pass through)” 역할. 병렬 구조에서 입력을 일부 다음 단계에서도 쓰고 싶을 때 유용
• | 연산자 (파이프): Runnable들을 직렬(sequence)로 연결. 입력 → 컴포넌트1 → 컴포넌트2 → … → 출력 흐름이 생김
• invoke / batch / stream
  • invoke(input) : 단일 입력에 대해 동기(synchronous) 실행
  • batch([input1, input2, …]) : 여러 입력을 한 번에 처리
  • stream(input) : 출력이 길거나 점진적으로 생성되는 경우 청크 단위로 결과 반환

예제 및 실행 결과

Runnable

LCEL에서 모든 구성 요소는 Runnable 인터페이스를 따릅니다.

아래 예제 코드에 등장하는 prompt, llm, parser 모두 Runnable입니다.

from dotenv import load_dotenv
import os
import openai

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

load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")

# Runnable 예시: Prompt, LLM, Parser 모두 Runnable
prompt = ChatPromptTemplate.from_template("Translate to English: {text}")
llm = ChatOpenAI(model="gpt-3.5-turbo")
parser = StrOutputParser()

# 체인 구성
chain = prompt | llm | parser

print(chain.invoke({"text": "안녕하세요"}))

 

Hello

 

RunnableSequence

RunnableSequence는 여러 Runnable을 직렬(sequence) 로 연결하는 방법을 코드로 직접 정의하는 방식입니다.
보통은 a | b | c 같은 파이프(|) 연산자를 더 자주 쓰지만, 내부적으로는 RunnableSequence를 생성합니다.

위의 예제 코드를 RunnableSequence로 변경하면 아래와 같습니다.

from dotenv import load_dotenv
import os
import openai

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

load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")

# Runnable 예시: Prompt, LLM, Parser 모두 Runnable
prompt = ChatPromptTemplate.from_template("Translate to English: {text}")
llm = ChatOpenAI(model="gpt-3.5-turbo")
parser = StrOutputParser()

# RunnableSequence로 체인 구성
chain = RunnableSequence(first=prompt, last=llm | parser)

# 실행
print(chain.invoke({"text": "안녕하세요"}))

Hello

 

RunnableLambda

직접 만든 Python 함수를 LCEL 파이프라인에 포함하려면 RunnableLambda를 씁니다.

일반 Python 함수를 체인 속에 넣을 때 유용합니다.

from langchain_core.runnables import RunnableLambda

# 간단한 함수
def double_number(x: int) -> int:
    return x * 2

# Runnable로 감싸기
double = RunnableLambda(double_number)

print(double.invoke(5))

10

 

RunnableParallel

여러 Runnable을 동시에 실행하고, 결과를 딕셔너리 형태로 모아줍니다.

입력값을 동시에 여러 방식으로 가공하고 싶을 때 사용합니다.

from langchain_core.runnables import RunnableParallel, RunnableLambda

# 간단한 Runnable 두 개
add_one = RunnableLambda(lambda x: x + 1)
square = RunnableLambda(lambda x: x * x)

# 병렬 실행: 같은 입력 x에 대해 add_one과 square 동시에 실행
parallel = RunnableParallel({
    "plus_one": add_one,
    "squared": square
})

print(parallel.invoke(3))

{'plus_one': 4, 'squared': 9}

 

RunnableParallel를 사용하지 않고 병렬적으로 사용하는 방법도 있습니다.

from langchain_core.runnables import RunnableParallel, RunnableLambda

# 간단한 Runnable 두 개
add_one = RunnableLambda(lambda x: x + 1)
square = RunnableLambda(lambda x: x * x)

# 병렬 실행: 같은 입력 x에 대해 add_one과 square 동시에 실행
parallel = {
    "plus_one": add_one,
    "squared": square
}

print(parallel.invoke(3))

 

RunnablePassthrough

입력 dict 전체를 출력으로 보존합니다.
특히 병렬 실행할 때, 원본 입력을 결과에 같이 포함시키고 싶을 때 자주 사용합니다.

from langchain_core.runnables import RunnablePassthrough, RunnableParallel, RunnableLambda

# 입력값을 두 배로 만드는 Runnable
double = RunnableLambda(lambda x: x * 2)

# 원본 입력과 가공된 값을 모두 반환
chain = RunnableParallel({
    "original": RunnablePassthrough(),
    "doubled": double
})

print(chain.invoke(7))

주의 사항

1. 입출력 형태(shape) 주의
   각 Runnable이 어떤 입력을 받고 어떤 출력을 내는지 명확히 해야 downstream 에러를 줄일 수 있습니다.
   타입 힌트나 input_schema, output_schema를 활용하면 안정성이 높아집니다.
2. 병렬 작업 시 비용/지연 고려
   병렬 실행은 네트워크 호출 수를 늘리므로 latency 증가나 rate limit 문제를 유발할 수 있습니다.
3. 스트리밍 지원 여부 확인
   .stream()/.astream() 사용 시, 모델이 스트리밍을 지원하는지 확인해야 합니다.
4. 비동기 실행
   서버 환경에서는 .ainvoke(), .abatch() 같은 비동기 메서드를 적극 활용하는 것이 좋습니다.
5. 디버깅 & 추적
   간단히는 RunnableLambda(lambda x: print(x) or x) 같은 방식으로 중간 출력 로깅이 가능합니다.
   본격적인 추적은 LangSmith 같은 툴과 연동해 각 단계의 입력/출력을 살펴보는 것이 좋습니다.
6. 조건 분기 / Fallback
   LCEL은 조건에 따라 다른 체인을 실행하거나, 실패 시 대체 경로(fallback)를 두는 방식도 지원합니다.
   (예: 입력이 영어면 그대로 출력, 아니면 번역 LLM 호출)
7. 버전 호환성
   LCEL API는 langchain-core 버전별로 일부 차이가 있을 수 있으므로, 공식 문서의 Cheat Sheet 를 반드시 참고하세요.