6. LCEL을 사용한 선언적 파이프라인

이전 챕터들에서 우리는 LLM 상호작용을 조율하기 위해 명령형 코드를 작성했습니다: 프롬프트를 생성하고, 모델을 호출하고, 응답을 파싱하는 방식이었죠. 이 방식은 작동하지만, AI 애플리케이션이 더 복잡해질수록 이 접근법은 장황해지고 유지보수하기 어려워집니다. 깊게 중첩된 함수 호출, 각 단계마다 수동으로 처리해야 하는 에러 핸들링, 그리고 전체 데이터 흐름을 이해하기 어려운 문제가 발생합니다.

LangChain Expression Language(LCEL)는 어떻게 만들지가 아니라 무엇을 원하는지를 선언할 수 있게 함으로써 이 문제를 해결합니다. 함수를 순차적으로 호출하는 절차적 코드를 작성하는 대신, 유닉스 파이프라인처럼 읽히는 간단한 파이프 연산자(|)를 사용하여 컴포넌트를 조합합니다. 그 결과는 AI 시스템을 통한 데이터 흐름을 명확하게 표현하는 더 깔끔하고 유지보수하기 쉬운 코드입니다.

이 챕터에서는 선형 워크플로우(linear workflows) 구축을 위한 LCEL을 소개합니다 - 데이터가 분기나 루프 없이 시작부터 끝까지 흐르는 작업 시퀀스입니다. LCEL을 언제 사용해야 하는지, 파이프라인을 조합하는 방법, 그리고 동기적으로 그리고 스트리밍 출력으로 실행하는 방법을 다룰 것입니다.

6.1) 왜 LCEL인가?

LCEL이 해결하는 문제

구체적인 예제로 시작해봅시다. 다음과 같은 작업이 필요한 고객 지원 어시스턴트를 구축한다고 가정해봅시다:

1. 사용자 질문을 받습니다
2. 시스템 지시사항과 사용자 메시지가 포함된 프롬프트로 포맷합니다
3. LLM에 전송합니다
4. 응답을 파싱하여 텍스트 콘텐츠만 추출합니다

다음은 명령형으로 작성하는 방법입니다(LCEL 없이):

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 친절한 고객 지원 어시스턴트입니다."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
def answer_question(question: str) -> str:
    # 단계 1: 질문으로 프롬프트 포맷
    messages = prompt.invoke({"question": question})
    
    # 단계 2: LLM 호출
    response = llm.invoke(messages)
    
    # 단계 3: 텍스트 콘텐츠 추출을 위해 출력 파싱
    result = parser.invoke(response)
    
    return result
 
# 사용
result = answer_question("비밀번호를 어떻게 재설정하나요?")
print(result)

출력:

비밀번호를 재설정하려면 다음 단계를 따르세요:
1. 로그인 페이지로 이동합니다
2. "비밀번호 찾기"를 클릭합니다
3. 이메일 주소를 입력합니다
4. 재설정 링크가 포함된 이메일을 확인합니다
5. 링크를 따라가서 새 비밀번호를 생성합니다

이 방식은 작동하지만, 다음과 같은 문제들이 있습니다:

장황함: 모든 단계가 명시적인 변수 할당과 함수 호출을 필요로 합니다. 실제 로직(포맷 → 호출 → 파싱)이 보일러플레이트에 묻혀 있습니다.

경직된 구조: 단계를 추가하려면(질문 검증이나 응답 로깅 같은), 함수 중간에 코드를 삽입해야 하므로 복잡도가 증가합니다.

내장 스트리밍 없음: 토큰이 도착하는 대로 스트리밍하려면, llm.stream()을 사용하고 비동기 반복을 수동으로 처리하도록 전체 함수를 다시 작성해야 합니다.

불명확한 데이터 흐름: 코드를 읽을 때, 이것이 단순한 파이프라인이라는 것이 즉시 명확하지 않습니다. 흐름을 이해하려면 변수 할당을 추적해야 합니다.

이제 LCEL 버전을 봅시다:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
# 컴포넌트 정의 (이전과 동일)
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 친절한 고객 지원 어시스턴트입니다."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
# 파이프 연산자를 사용하여 체인으로 조합
chain = prompt | llm | parser
 
# 사용
result = chain.invoke({"question": "비밀번호를 어떻게 재설정하나요?"})
print(result)

출력:

비밀번호를 재설정하려면 다음 단계를 따르세요:
1. 로그인 페이지로 이동합니다
2. "비밀번호 찾기"를 클릭합니다
3. 이메일 주소를 입력합니다
4. 재설정 링크가 포함된 이메일을 확인합니다
5. 링크를 따라가서 새 비밀번호를 생성합니다

출력은 동일하지만, 코드는 극적으로 다릅니다:

선언적: chain = prompt | llm | parser는 전체 흐름을 한 줄로 표현합니다. 왼쪽에서 오른쪽으로 읽으세요: 프롬프트 → LLM → 파서.

조합 가능: 각 컴포넌트(prompt, llm, parser)는 독립적이고 재사용 가능합니다. 파이프라인을 다시 작성하지 않고도 컴포넌트를 교체할 수 있습니다.

내장 스트리밍: 파이프라인 정의를 변경하지 않고 chain.invoke()와 chain.stream() 사이를 전환할 수 있습니다.

명확한 의도: | 연산자는 데이터 흐름을 한눈에 명확하게 보여줍니다.

LCEL이 적합한 곳: 선형 워크플로우

LCEL은 선형 워크플로우(linear workflows)를 위해 설계되었습니다 - 분기나 루프 없이 데이터가 시작부터 끝까지 한 방향으로 흐르는 시퀀스입니다.

이러한 선형 패턴은 많은 AI 애플리케이션을 커버합니다. 예를 들어, 문서 질의응답 시스템을 생각해보세요: 질문을 받고 → 관련 문서를 검색하고 → 프롬프트를 포맷하고 → LLM에 전송하고 → 답변을 파싱합니다. 각 단계는 이전 단계의 출력을 다음 단계의 입력으로 사용하는 명확한 시퀀스입니다.

하지만 워크플로우에 다음이 필요하다면 어떻게 될까요?

• LLM이 질문에 기반하여 어떤 도구를 호출할지 결정해야 함
• 도구를 호출하고, 결과를 관찰한 후, 다음에 무엇을 할지 결정해야 함
• 실패한 작업을 다른 접근법으로 재시도해야 함

이러한 시나리오는 루프와 조건부 분기가 필요합니다 - LCEL이 처리할 수 없는 것들입니다. 이것이 LangGraph가 필요한 이유입니다 (챕터 15에서 소개).

다음은 두 접근 방식의 시각적 비교입니다:

LCEL은 각 단계가 이전 단계의 출력을 처리하는 직선적인 파이프라인에 완벽합니다. 데이터는 한 방향으로만 흐릅니다.

LangGraph는 의사결정과 루프가 필요할 때 사용됩니다. 에이전트는 행동하고, 결과를 관찰하고, 다시 생각할 수 있습니다.

이 챕터에서는 LCEL에 집중합니다. 선형 파이프라인을 먼저 마스터해야 하는 이유:

• 기초부터: LCEL의 파이프 연산자(|)는 LangChain의 기본 문법입니다. 이를 이해해야 다른 것들도 쉽게 이해할 수 있습니다.
• LangGraph의 전제조건: LangGraph를 구성하는 에이전트 내에서도 LCEL 체인을 많이 사용합니다
• 실무 패턴: 복잡한 에이전트도 결국 LCEL 체인들을 조합한 것입니다

이 챕터의 나머지 부분에서는 파이프 연산자(|)를 사용하여 이러한 선형 파이프라인을 구축하는 방법을 보여줄 것입니다.

6.2) 파이프 연산자로 파이프라인 조합하기

LCEL 파이프의 동작 원리

LCEL의 파이프 연산자(|)는 Runnable 인터페이스 덕분에 작동합니다.

Runnable이란 무엇인가?

Runnable은 LangChain의 표준 인터페이스입니다. 컴포넌트가 Runnable 인터페이스를 구현하면, | 연산자로 다른 컴포넌트와 체이닝할 수 있습니다.

모든 Runnable은 다음 메서드를 제공합니다:

• .invoke(input) - 한 번 실행하고 완전한 결과를 받습니다
• .stream(input) - 실행하고 LLM이 단어를 생성할 때마다 즉시 받습니다
• .batch(inputs) - 여러 개의 입력을 한 번에 처리하고 모든 결과를 받습니다

이 챕터에서는 .invoke()와 .stream()에 집중할 것입니다 (.batch()는 필요할 때 나중에 다룰 것입니다).

왜 |가 작동하는가?

Runnable 클래스가 Python의 연산자 오버로딩 기능을 사용하여 | 연산자를 구현했기 때문입니다. prompt | llm을 작성하면, 두 컴포넌트를 순차적으로 실행하는 새로운 Runnable이 생성됩니다.

대부분의 LangChain 컴포넌트는 Runnable입니다

이것이 여러 다양한 컴포넌트를 체이닝할 수 있는 이유입니다:

• ChatPromptTemplate은 Runnable입니다
• ChatOpenAI는 Runnable입니다
• StrOutputParser는 Runnable입니다
• |로 생성한 커스텀 체인조차도 그 자체가 Runnable입니다!

이는 더 간단한 파이프라인을 조합하여 복잡한 파이프라인을 구축할 수 있다는 것을 의미합니다.

컴포넌트 연결하기: 입력/출력 타입

|로 컴포넌트를 연결할 때, 한 컴포넌트의 출력 타입이 다음 컴포넌트의 입력 타입과 일치해야 합니다.

주요 컴포넌트 시그니처:

예제 흐름:

chain = prompt | llm | parser

각 컴포넌트를 거치면서 데이터 타입이 어떻게 변환되는지 다이어그램으로 보면 다음과 같습니다:

• prompt: dict를 입력으로 받아 list[BaseMessage]로 변환
• llm: list[BaseMessage]를 입력으로 받아 AIMessage로 변환
• parser: AIMessage를 입력으로 받아 str로 변환

실제로 동작하는 모습을 봅시다:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 도움이 되는 어시스턴트입니다."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()
 
chain = prompt | llm | parser
 
result = chain.invoke({"question": "2+2는?"})
print(result)  # "2+2는 4입니다."

각 단계에서 일어나는 일:

타입이 일치하지 않으면 어떻게 되나요?

호환되지 않는 컴포넌트를 연결하려고 하면 오류가 발생합니다:

# 오류: 이것은 작동하지 않습니다
chain = llm | prompt  # LLM은 AIMessage를 출력하지만, prompt는 입력으로 dict를 받아야 합니다

오류 메시지는 뒤에 있는 컴포넌트가 기대하는 입력 타입과 실제로 받은 입력 타입을 알려줍니다.

6.3) 체인 실행하기: .invoke()와 .stream()

파이프라인 실행하기

체인을 조합한 후에는 .invoke() 메서드를 사용하여 실행합니다. 이것은 동기 방식으로 파이프라인을 실행합니다 - 전체 응답이 완료될 때까지 기다린 후 결과를 반환합니다.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 도움이 되는 어시스턴트입니다."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# 동기 실행
result = chain.invoke({"question": "2+2는 무엇인가요?"})
print(result)

출력:

2+2는 4입니다.

.invoke() 메서드는 간단합니다: 첫 번째 컴포넌트가 받는 입력 파라미터를 전달하면, 마지막 컴포넌트가 출력하는 결과값을 받습니다.

동일한 체인에서 토큰 스트리밍하기

.invoke()는 간단하지만, 사용자 대면 애플리케이션에서는 한계가 있습니다: 전체 응답이 완료될 때까지 사용자는 아무것도 볼 수 없습니다. 긴 응답(10-20초)의 경우, 이는 좋지 않은 사용자 경험을 만듭니다.

스트리밍은 최종 응답을 기다리는 대신, 토큰이 생성되는 대로 즉시 보여줍니다. ChatGPT에서 보는 타이핑 효과와 같습니다.

실제로 스트리밍이 동작하는 모습을 봅시다:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 도움이 되는 어시스턴트입니다."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# 청크가 도착하는 대로 스트리밍
for chunk in chain.stream({"question": "Python이 무엇인지 한 문장으로 설명해주세요"}):
    print(chunk, end="", flush=True)

출력 (토큰 단위로 실시간 표시):

Python은 간결하고 읽기 쉬운 문법으로 알려진 고수준 프로그래밍 언어로, 웹 개발, 데이터 과학, 자동화 등에 널리 사용됩니다.

위의 .invoke() 예제와 체인 정의가 동일하다는 점을 주목하세요. 체인을 다시 만들 필요 없이, .invoke() 대신 .stream()을 호출하기만 했습니다.

.stream() 메서드는 LLM이 토큰을 생성하는 대로 청크(chunk) 단위로 반환합니다. print(chunk, end="", flush=True)는 각 청크를 화면에 즉시 표시하여 실시간 타이핑 효과를 만듭니다.

.invoke() vs .stream() 언제 사용할까?

.invoke() 사용:

• 완성된 응답만 필요할 때 (분석, 번역, 분류)
• 응답이 짧아서 대기 시간이 문제되지 않을 때
• 다음 단계로 넘어가기 전에 전체 결과가 필요할 때

.stream() 사용:

• 사용자가 진행 상황을 봐야 할 때 (채팅, 콘텐츠 생성)
• 응답이 길어서 대기 시간이 긴 경우
• 실시간 피드백이 중요한 UI를 만들 때

두 메서드 모두 동일한 체인에서 작동합니다. 체인은 한 번만 정의하고, 상황에 따라 실행 방식을 선택하면 됩니다.

이 챕터에서는 LangChain의 선언적 파이프라인 문법인 LCEL을 배웠습니다:

• 파이프라인 구축: 파이프 연산자로 구축 - prompt | llm | parser
• 유연한 실행: 완전한 결과를 위해 .invoke() 사용 또는 실시간 출력을 위해 .stream() 사용
• 타입 안전성: 컴포넌트를 체이닝할 때 출력 타입을 입력 타입에 맞추기

동일한 체인이 두 실행 모드 모두에서 작동합니다 - 한 번 정의하고, 어디서나 사용하세요.

다음 챕터: 챕터 7에서는 Pydantic을 사용한 구조화된 출력을 다루며, LLM 응답에서 검증된 JSON 데이터를 추출할 수 있게 합니다.