LCEL이란?

LangChain에서 chain을 구성하는 방법입니다. 프롬프트나 LLM을 | 로 연결하여 작성하고 처리의 연쇄를 구성한다.

LCEL에서는 왜 프롬프트, 모델, 아웃풋 파서등을 |(체인)으로 연결해서 실행할 수 있을까?

모든 클래스가 langchain의 Runnable 클래스를 상속받고 있기 때문이다. Runnable을 연결하면 RunnableSequence가 된다. 이 시퀀스도 Runnable의 일종이다. RunnableSequence를 invoke하면 runnable이 순서대로 invoke된다.

LCEL 내부 구현

LCEL은 어떻게 구현되었길래 체인으로 모델과 프롬프트를 연결할 수 있을까? __or__과 __ror__가 이렇게 구현되어 있다:

def __or__(self,
    other: Union[Runnable[Any, Other],
                Callable[[Any], Other]],
                Callable[Iterator[Any], Iterator[Other]],
                Mapping[str, Union[Runnable[Any,Other]], Callable[[Any],Other],Any]) -> RunnableSerializable[Input, Other]:
        return RunnableSequence(first=self, last=coerce_to_runnable(other))

def __ror__(self,
    other: Union[Runnable[Other, Any],
                Callable[[Other], Any]],
                Callable[Iterator[Other], Iterator[Any]],
                Mapping[str, Union[Runnable[Other,Any]], Callable[[Other],Any],Any]) -> RunnableSerializable[Other, Output]:
        return RunnableSequence(first=coerce_to_runnable(other), last=self)

이 연산자들을 통해 체인을 오버로드할 수 있다.

그러니까 원래라면 각각의 러너블을 실행(호출)하는 방법이 달라, 이 컴포넌트들을 연결하는 처리를 개별적으로 구현해야하는데, LCEL을 사용하면 모든 컴포넌트를 통일된 인터페이스로 호출할 수 있게 된다. 이게 엄청 편리함!!

LCEL이 편리한 유스케이스

프롬프트 템플릿을 채우고, 그 결과를 챗 모델에 제공하고 결과를 파이썬 객체로 변환하고 싶다.
프롬프팅으로 단계별로 생각하게 하고 그 결과를 요약하게 하고 싶다
출력을 얻은 후에 서비스 정책에 위반되지 않는지 확인하고 싶다
출력 결과를 바탕으로 SQL을 실행하여 데이터를 분석하고 싶다.

사용 방법

1. Prompt와 Model 연결하기

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant that can answer questions."),
    ("human", "{input}")
])

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

# 체인 생성
chain = prompt | model

ai_message = chain.invoke({"input": "What is the capital of France?"})
print(ai_message.content)

랭체인을 안쓰는 경우와 크게 달라보이지 않는다. 그렇지만 프롬프트가 다양해지고, 워크플로우가 복잡해지는 경우에는 이 구조가 아주 편리해진다.

예를 들어보자.

예시) Prompt, Model, Pydantic Output Parser 연결하기

from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field, List

# 응답 형식 모델 설정
class Recipe(BaseModel):
    dish: str = Field(description = "먹고싶은 음식의 이름")
    recipe: str = Field(description = "음식의 요리법")
    ingredient: List[str] = Field(description = "필요한 재료의 목록")

# 연쇄 처리할 부분
prompt = ChatPromptTemplate.from_messages([...])

output_parser = PydanticOutputParser(pydantic_object = Recipe)
prompt_input = prompt.partial(
    format_instructions = output_parser.get_format_instructions()
)

model = ChatOpenAI(model = "gpt-4o-mini", temperature = 0.0).bind(
    response_format = {"type": "json_object"}
)

chain = prompt_input | model | output_parser

output = chain.invoke({"input": "당근 카레"})

이렇게 하면 여러개의 러너블도 간단하게 한줄로 연결할 수 있다.

# 동일한 기능을 하는 간단한 다른 방법
chain2 = prompt | model.with_structured_output(Recipe)

output2 = chain2.invoke({"input": "양파 카레"})

2. LCEL의 고급 연산자들

RunnableParallel (병렬 실행):

from langchain_core.runnables import RunnableParallel

# 여러 작업을 동시에 실행
parallel_chain = RunnableParallel(
    summary=prompt | model | StrOutputParser(),
    keywords=keyword_prompt | model | StrOutputParser(),
    sentiment=sentiment_prompt | model | StrOutputParser()
)

RunnableLambda (함수 변환):

from langchain_core.runnables import RunnableLambda

def format_input(x):
    return {"input": x.upper()}

chain = RunnableLambda(format_input) | prompt | model

RunnableBranch (조건부 분기):

from langchain_core.runnables import RunnableBranch

chain = RunnableBranch(
    (lambda x: "question" in x.lower(), question_chain),
    (lambda x: "summary" in x.lower(), summary_chain),
    default_chain  # 기본값
)

주의할 점 : 입출력 타입이 일치할 것!

# 타입 불일치 예시 (에러 발생)
prompt_returns_messages = ChatPromptTemplate.from_template("Hello {name}")
parser_expects_string = StrOutputParser()

# 올바른 연결
chain = prompt_returns_messages | model | parser_expects_string

주의사항:

ChatPromptTemplate → AIMessage 객체 출력
StrOutputParser → AIMessage.content (string) 추출
중간에 모델이 AIMessage 생성해야 함