LangChain 고급 LCEL 컴포넌트: RunnablePassthrough, RunnableParallel, RunnableLambda

LangChain Expression Language(LCEL)에서는 다양한 Runnable 계열의 유틸리티 클래스들을 제공한다. 이 글에서는 그 중에서도 실무에서 자주 활용되는 RunnablePassthrough, RunnableParallel, RunnableLambda 세 가지 컴포넌트를 개별적으로 정리한다.

1. RunnablePassthrough

정의: 입력을 그대로 출력으로 반환하는 가장 단순한 형태의 Runnable 구현체이다.

용도

체인 구조 테스트 시 플레이스홀더 역할
전처리 또는 후처리를 생략하고 기본 입력값을 유지할 때
병렬 실행 시 특정 경로에서 처리가 필요 없는 경우

예제 코드

from langchain.schema.runnable import RunnablePassthrough

passthrough = RunnablePassthrough()
result = passthrough.invoke("LCEL 테스트")  # 결과: "LCEL 테스트"

비동기 실행

result = await passthrough.ainvoke("비동기 실행 테스트")

---

2. RunnableParallel

정의: 여러 개의 Runnable을 병렬로 실행하고 결과를 병합하여 반환하는 구조이다. 입력은 동일하게 전달되며, 결과는 dict 형식으로 구성된다.

용도

프롬프트 버전 비교 (A/B 테스트)
동일 입력에 대한 멀티 모델 응답 수집
다양한 파이프라인 실행 결과의 병렬 처리

예제 코드

from langchain.prompts import PromptTemplate
from langchain.chat_models import ChatOpenAI
from langchain.output_parsers import StrOutputParser
from langchain.schema.runnable import RunnableParallel

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

prompt1 = PromptTemplate.from_template("질문1: {question}")
prompt2 = PromptTemplate.from_template("질문2: {question}")

chain1 = prompt1 | llm | StrOutputParser()
chain2 = prompt2 | llm | StrOutputParser()

parallel = RunnableParallel({
    "ver1": chain1,
    "ver2": chain2
})

result = parallel.invoke({"question": "LangChain이란?"})

# 결과: {"ver1": "...", "ver2": "..."}

비동기 실행

result = await parallel.ainvoke({"question": "LCEL 비교"})

---

3. RunnableLambda

정의: Python의 람다 함수 또는 커스텀 함수를 Runnable로 감싸 체인에 삽입할 수 있게 해주는 유틸리티 컴포넌트이다.

용도

입력값 전처리 또는 출력값 후처리
간단한 수식 계산, 텍스트 포맷 변환
조건 분기 및 커스텀 로직 삽입

예제 코드

from langchain.schema.runnable import RunnableLambda

multiply = RunnableLambda(lambda x: x * 10)
result = multiply.invoke(3)  # 결과: 30

체인 내 삽입 예시

chain = prompt | RunnableLambda(lambda x: {"prompt": x["question"].upper()}) | llm

비동기 실행

result = await multiply.ainvoke(5)

728x90