LangChain LCEL 완벽 가이드

LangChain Expression Language(LCEL)를 활용하여 AI 체인을 우아하게 구성하는 방법을 배웁니다. 파이프 연산자부터 커스텀 체인 개발까지, 실무에서 바로 활용할 수 있는 핵심 개념을 다룹니다.

---

목차

1. LangChain Expression Language
2. Chain 구성
3. Runnable 인터페이스
4. 실습: LCEL로 복잡한 체인
5. 실습: 커스텀 체인 개발

---

1. LangChain Expression Language

어느 날 김개발 씨는 회사에서 AI 챗봇 프로젝트를 맡게 되었습니다. 프롬프트를 만들고, LLM을 호출하고, 결과를 파싱하는 코드를 작성했는데, 어느새 코드가 스파게티처럼 엉켜버렸습니다.

"이걸 어떻게 유지보수하지?" 한숨이 절로 나왔습니다.

**LCEL(LangChain Expression Language)**은 LangChain의 핵심 구성 요소들을 파이프라인처럼 연결하는 선언적 문법입니다. 마치 레고 블록을 조립하듯이, 프롬프트, 모델, 파서 등을 파이프 연산자(|)로 깔끔하게 이어붙일 수 있습니다.

이를 통해 복잡한 AI 워크플로우도 한눈에 들어오는 코드로 표현할 수 있습니다.

다음 코드를 살펴봅시다.

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

# 프롬프트 템플릿 정의
prompt = ChatPromptTemplate.from_template(
    "다음 주제에 대해 한 문장으로 설명해주세요: {topic}"
)

# 모델과 파서 준비
model = ChatOpenAI(model="gpt-4")
parser = StrOutputParser()

# LCEL로 체인 구성 - 파이프 연산자로 연결
chain = prompt | model | parser

# 체인 실행
result = chain.invoke({"topic": "인공지능"})
print(result)

김개발 씨는 입사 6개월 차 백엔드 개발자입니다. 최근 회사에서 AI 기능을 도입하기로 했고, LangChain을 사용해 챗봇을 만드는 임무를 맡았습니다.

처음에는 의욕이 넘쳤습니다. 하지만 코드를 작성할수록 머리가 복잡해졌습니다.

프롬프트를 만드는 코드, LLM을 호출하는 코드, 결과를 처리하는 코드가 여기저기 흩어져 있었습니다. 한 기능을 수정하려면 여러 파일을 뒤져야 했고, 동료에게 코드를 설명하기도 어려웠습니다.

그때 선배 개발자 박시니어 씨가 다가왔습니다. "LCEL 써봤어요?

코드가 훨씬 깔끔해질 거예요." LCEL이란 무엇일까요? 쉽게 비유하자면, 마치 공장의 컨베이어 벨트와 같습니다.

원재료가 벨트 위에 올라가면, 각 작업대를 거치면서 가공되고, 최종 제품이 나오는 것처럼요. LCEL에서는 입력 데이터가 프롬프트, 모델, 파서를 차례로 거치면서 원하는 결과물로 변환됩니다.

LCEL이 없던 시절에는 어땠을까요? 개발자들은 각 단계를 직접 함수로 호출하고, 중간 결과를 변수에 저장하고, 다음 단계에 전달하는 코드를 일일이 작성해야 했습니다.

에러 처리도 각 단계마다 따로 해야 했고, 비동기 처리는 더욱 복잡했습니다. LCEL의 핵심은 **파이프 연산자(|)**입니다.

유닉스의 파이프라인에서 영감을 받은 이 문법은, 왼쪽 컴포넌트의 출력이 자동으로 오른쪽 컴포넌트의 입력이 됩니다. prompt | model | parser라고 쓰면, 프롬프트의 결과가 모델로, 모델의 결과가 파서로 자연스럽게 흘러갑니다.

위 코드를 살펴보겠습니다. 먼저 ChatPromptTemplate으로 프롬프트를 정의합니다.

{topic} 부분은 나중에 실제 값으로 채워질 자리입니다. 그 다음 ChatOpenAI로 사용할 모델을 지정하고, StrOutputParser로 결과를 문자열로 변환할 파서를 준비합니다.

가장 중요한 부분은 chain = prompt | model | parser입니다. 이 한 줄이 전체 파이프라인을 정의합니다.

마치 "프롬프트를 거쳐, 모델을 통과하고, 파서로 마무리한다"라고 읽을 수 있습니다. 실행은 invoke 메서드로 합니다.

딕셔너리 형태로 입력값을 전달하면, 체인이 순차적으로 실행되어 최종 결과가 반환됩니다. 스트리밍이 필요하면 stream, 비동기 처리가 필요하면 ainvoke를 사용할 수 있습니다.

LCEL의 진정한 장점은 가독성과 유지보수성입니다. 복잡한 로직도 파이프라인으로 표현하면 한눈에 흐름이 보입니다.

새로운 팀원이 와도 코드를 쉽게 이해할 수 있고, 중간에 단계를 추가하거나 제거하기도 간편합니다. 박시니어 씨의 조언을 따라 코드를 리팩토링한 김개발 씨는 감탄했습니다.

"와, 코드가 절반으로 줄었어요! 그리고 뭘 하는 코드인지 바로 보이네요." 이제 김개발 씨의 챗봇 프로젝트는 한결 관리하기 쉬워졌습니다.

실전 팁

💡 - 파이프 연산자는 왼쪽에서 오른쪽으로 데이터가 흐른다고 생각하면 이해하기 쉽습니다

• 복잡한 체인은 중간 변수로 나눠서 가독성을 높일 수 있습니다
• invoke 외에도 batch, stream, ainvoke 등 다양한 실행 방법을 상황에 맞게 선택하세요

---

2. Chain 구성

김개발 씨가 LCEL의 기본을 익힌 후, 좀 더 복잡한 요구사항이 들어왔습니다. "사용자 질문을 분석해서 카테고리를 분류하고, 카테고리에 맞는 답변을 생성해주세요." 단순히 컴포넌트를 연결하는 것만으로는 부족해 보였습니다.

Chain은 여러 컴포넌트를 논리적으로 묶어 하나의 작업 단위로 만든 것입니다. 마치 요리 레시피처럼, 재료 손질부터 완성까지의 단계를 정해놓은 것과 같습니다.

LCEL에서는 다양한 방식으로 체인을 구성할 수 있으며, 조건 분기나 병렬 처리도 가능합니다.

다음 코드를 살펴봅시다.

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

# 분류 체인
classify_prompt = ChatPromptTemplate.from_template(
    "다음 질문의 카테고리를 '기술', '일반', '비즈니스' 중 하나로 분류하세요: {question}"
)

# 답변 체인
answer_prompt = ChatPromptTemplate.from_template(
    "카테고리: {category}\n질문: {question}\n위 카테고리에 맞게 답변해주세요."
)

model = ChatOpenAI(model="gpt-4")
parser = StrOutputParser()

# 체인 조합 - 분류 결과를 다음 단계로 전달
chain = (
    {"category": classify_prompt | model | parser, "question": RunnablePassthrough()}
    | answer_prompt
    | model
    | parser
)

result = chain.invoke("Python의 GIL이 뭔가요?")

김개발 씨는 LCEL의 파이프 연산자에 익숙해졌습니다. 하지만 새로운 과제가 주어졌습니다.

사용자의 질문을 먼저 분석하고, 그 분석 결과에 따라 다른 방식으로 답변해야 하는 상황이었습니다. "단순히 prompt | model | parser만으로는 안 되겠는데..." 김개발 씨는 고민에 빠졌습니다.

분류 결과와 원본 질문, 두 가지 정보를 동시에 다음 단계로 넘겨야 했기 때문입니다. 박시니어 씨가 화이트보드에 그림을 그리며 설명했습니다.

"체인 구성의 핵심은 데이터 흐름을 설계하는 거예요. 마치 물이 여러 갈래로 나뉘었다가 다시 합쳐지는 것처럼요." LCEL에서 체인을 구성하는 방법은 여러 가지입니다.

가장 강력한 기법 중 하나는 딕셔너리를 사용한 병렬 처리입니다. 딕셔너리의 각 키에 다른 체인이나 값을 할당하면, 그것들이 병렬로 실행되고 결과가 하나의 딕셔너리로 합쳐집니다.

위 코드에서 핵심 부분을 살펴보겠습니다. {"category": classify_prompt | model | parser, "question": RunnablePassthrough()}라는 구조가 보입니다.

여기서 category 키에는 분류 체인이 연결되어 있고, question 키에는 RunnablePassthrough()가 연결되어 있습니다. RunnablePassthrough는 입력을 그대로 통과시키는 특별한 컴포넌트입니다.

마치 고속도로의 통과 차선처럼, 데이터를 변형 없이 그대로 다음 단계로 전달합니다. 덕분에 원본 질문을 보존하면서 동시에 분류 작업도 수행할 수 있습니다.

이 구조가 실행되면 어떤 일이 일어날까요? 입력으로 "Python의 GIL이 뭔가요?"라는 질문이 들어오면, 분류 체인은 이를 "기술"로 분류합니다.

동시에 RunnablePassthrough는 원본 질문을 그대로 유지합니다. 결과적으로 {category: "기술", question: "Python의 GIL이 뭔가요?"}라는 딕셔너리가 만들어집니다.

이 딕셔너리는 answer_prompt로 전달됩니다. 프롬프트 템플릿의 {category}와 {question} 자리가 딕셔너리의 값으로 채워지고, 완성된 프롬프트가 모델로 전달됩니다.

최종적으로 카테고리에 맞춤화된 답변이 생성됩니다. 실무에서 이런 패턴은 매우 유용합니다.

예를 들어 고객 문의 시스템에서 질문 유형을 먼저 파악하고, 유형에 따라 다른 전문가 프롬프트를 적용할 수 있습니다. 또는 감정 분석을 먼저 수행하고, 부정적인 감정이 감지되면 더 공감적인 톤으로 응답하도록 설계할 수도 있습니다.

주의할 점이 있습니다. 체인이 복잡해질수록 디버깅이 어려워집니다.

중간 결과를 확인하고 싶다면 RunnableLambda를 사용해 로깅을 추가하거나, 체인을 작은 단위로 나눠서 테스트하는 것이 좋습니다. 김개발 씨는 이제 복잡한 워크플로우도 깔끔하게 표현할 수 있게 되었습니다.

"체인 구성이 이렇게 유연한 줄 몰랐어요. 마치 레고 블록 조립하는 것 같네요!"

실전 팁

💡 - 딕셔너리를 사용하면 여러 데이터를 병렬로 처리하고 합칠 수 있습니다

• RunnablePassthrough는 원본 데이터를 보존해야 할 때 유용합니다
• 복잡한 체인은 작은 단위로 나눠서 개별 테스트 후 조합하세요

---

3. Runnable 인터페이스

김개발 씨가 체인을 구성하다 보니 문득 궁금해졌습니다. "왜 모든 컴포넌트가 파이프로 연결될 수 있는 거지?

프롬프트도, 모델도, 파서도 전혀 다른 건데..." 그 비밀은 바로 Runnable 인터페이스에 있었습니다.

Runnable은 LCEL의 모든 컴포넌트가 구현하는 공통 인터페이스입니다. 마치 USB 포트처럼, 서로 다른 기기도 같은 규격을 따르면 연결할 수 있듯이, Runnable 인터페이스를 구현한 컴포넌트들은 자유롭게 연결될 수 있습니다.

invoke, batch, stream 등 표준화된 메서드를 제공합니다.

다음 코드를 살펴봅시다.

from langchain_core.runnables import RunnableLambda, RunnableParallel
from langchain_openai import ChatOpenAI

# 커스텀 함수를 Runnable로 변환
def add_greeting(text: str) -> str:
    return f"안녕하세요! {text}"

def add_emoji(text: str) -> str:
    return f"{text} :)"

# RunnableLambda로 일반 함수를 Runnable로 변환
greeting_runnable = RunnableLambda(add_greeting)
emoji_runnable = RunnableLambda(add_emoji)

# Runnable 체인 구성
chain = greeting_runnable | emoji_runnable

# 다양한 실행 방식
print(chain.invoke("반갑습니다"))  # 단일 실행
print(chain.batch(["첫번째", "두번째"]))  # 배치 실행

김개발 씨는 LCEL을 사용하면서 신기한 점을 발견했습니다. ChatPromptTemplate, ChatOpenAI, StrOutputParser는 완전히 다른 클래스인데, 어떻게 파이프 연산자로 자연스럽게 연결될까요?

박시니어 씨가 설명했습니다. "그건 모두 Runnable이라는 공통 인터페이스를 구현하고 있기 때문이에요.

마치 모든 전자제품이 220V 콘센트에 꽂히는 것처럼요." Runnable 인터페이스는 LangChain의 핵심 설계 철학입니다. 이 인터페이스를 구현한 모든 컴포넌트는 동일한 방식으로 실행할 수 있습니다.

가장 중요한 메서드는 invoke입니다. 입력을 받아 처리하고 출력을 반환하는 단순한 구조입니다.

하지만 Runnable의 진정한 힘은 다양한 실행 방식에 있습니다. invoke는 단일 입력을 처리하고, batch는 여러 입력을 한 번에 처리합니다.

stream은 결과를 조금씩 흘려보내고, ainvoke, abatch, astream은 비동기 버전입니다. 위 코드에서 주목할 부분은 RunnableLambda입니다.

이것은 일반 파이썬 함수를 Runnable로 변환해주는 래퍼입니다. add_greeting과 add_emoji는 평범한 함수지만, RunnableLambda로 감싸면 체인에 연결할 수 있게 됩니다.

이것이 왜 중요할까요? 실무에서는 LLM 호출 전후로 데이터를 가공해야 하는 경우가 많습니다.

입력 데이터를 정규화하거나, 출력 결과를 후처리하거나, 외부 API를 호출해야 할 수도 있습니다. RunnableLambda를 사용하면 이런 커스텀 로직도 체인의 일부로 깔끔하게 통합할 수 있습니다.

RunnableParallel도 알아두면 좋습니다. 여러 Runnable을 병렬로 실행하고 결과를 딕셔너리로 합쳐줍니다.

앞서 배운 딕셔너리 문법이 사실 RunnableParallel의 축약형입니다. batch 메서드는 성능 최적화에 핵심입니다.

여러 입력을 개별적으로 invoke하는 것보다 batch로 한 번에 처리하면 훨씬 효율적입니다. 특히 외부 API 호출이 포함된 경우, 네트워크 오버헤드를 크게 줄일 수 있습니다.

stream 메서드는 사용자 경험에 중요합니다. LLM의 응답은 생성에 시간이 걸리는데, stream을 사용하면 토큰이 생성되는 대로 바로 보여줄 수 있습니다.

ChatGPT처럼 글자가 하나씩 나타나는 효과를 구현할 수 있는 것입니다. 주의할 점이 있습니다.

모든 Runnable이 모든 메서드를 완벽하게 지원하는 것은 아닙니다. 예를 들어 일부 컴포넌트는 스트리밍을 지원하지 않을 수 있습니다.

체인을 구성할 때는 각 컴포넌트의 지원 범위를 확인하는 것이 좋습니다. 김개발 씨는 이제 Runnable의 개념을 이해했습니다.

"아, 그래서 뭐든 파이프로 연결할 수 있었군요. 제가 만든 함수도 체인에 넣을 수 있다니 정말 유연하네요!"

실전 팁

💡 - RunnableLambda로 어떤 파이썬 함수든 체인에 통합할 수 있습니다

• batch를 활용하면 여러 입력을 효율적으로 처리할 수 있습니다
• stream은 LLM 응답을 실시간으로 보여줄 때 필수입니다

---

4. 실습: LCEL로 복잡한 체인

김개발 씨에게 도전적인 과제가 주어졌습니다. "사용자 질문을 받아서, 관련 문서를 검색하고, 검색 결과를 바탕으로 답변을 생성해주세요.

아, 그리고 답변의 신뢰도 점수도 함께 출력해야 해요." 여러 단계가 복잡하게 얽힌 체인을 구성해야 했습니다.

실제 프로덕션 환경에서는 단순한 체인으로는 부족합니다. RAG(Retrieval-Augmented Generation) 패턴처럼 검색과 생성을 결합하거나, 여러 모델의 결과를 종합해야 하는 경우가 많습니다.

LCEL의 다양한 기능을 조합하면 이런 복잡한 워크플로우도 우아하게 표현할 수 있습니다.

다음 코드를 살펴봅시다.

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda

# 가상의 문서 검색 함수
def search_documents(query: str) -> str:
    docs = {"Python": "Python은 1991년 만들어진 프로그래밍 언어입니다."}
    return docs.get(query.split()[0], "관련 문서를 찾을 수 없습니다.")

# 신뢰도 평가 프롬프트
confidence_prompt = ChatPromptTemplate.from_template(
    "다음 답변의 신뢰도를 0-100 사이 숫자로만 답하세요: {answer}"
)

# 답변 생성 프롬프트
answer_prompt = ChatPromptTemplate.from_template(
    "문서: {context}\n질문: {question}\n문서를 참고하여 답변하세요."
)

model = ChatOpenAI(model="gpt-4")
parser = StrOutputParser()

# 복잡한 체인 구성
chain = (
    {"context": RunnableLambda(search_documents), "question": RunnablePassthrough()}
    | answer_prompt | model | parser
    | {"answer": RunnablePassthrough(), "confidence": confidence_prompt | model | parser}
)

result = chain.invoke("Python이란 무엇인가요?")

김개발 씨는 이제 LCEL의 기본기를 갖췄습니다. 하지만 실제 프로덕션 환경은 훨씬 복잡했습니다.

단순히 질문을 받아 답변하는 것이 아니라, 검색도 하고, 분석도 하고, 평가도 해야 했습니다. 박시니어 씨가 화이트보드에 흐름도를 그렸습니다.

"실무에서 가장 많이 쓰는 패턴이 RAG예요. Retrieval-Augmented Generation, 즉 검색으로 보강된 생성이라는 뜻이죠." RAG의 핵심 아이디어는 간단합니다.

LLM은 학습 데이터에 없는 최신 정보나 특정 도메인 지식을 모릅니다. 그래서 먼저 관련 문서를 검색하고, 그 문서를 컨텍스트로 제공하면서 질문에 답하게 하는 것입니다.

마치 오픈북 시험처럼요. 위 코드의 구조를 단계별로 살펴보겠습니다.

첫 번째 딕셔너리 {"context": RunnableLambda(search_documents), "question": RunnablePassthrough()}에서 두 가지 일이 동시에 일어납니다. search_documents 함수가 질문을 받아 관련 문서를 검색하고, 원본 질문은 그대로 유지됩니다.

검색 결과와 질문이 answer_prompt로 전달됩니다. 프롬프트는 "문서를 참고하여 답변하세요"라고 모델에게 지시합니다.

모델은 검색된 문서를 바탕으로 더 정확한 답변을 생성할 수 있습니다. 여기서 끝이 아닙니다.

답변이 생성된 후, 두 번째 딕셔너리가 또 다른 병렬 처리를 수행합니다. answer 키에는 생성된 답변이 그대로 저장되고, confidence 키에서는 별도의 체인이 답변의 신뢰도를 평가합니다.

이 구조의 장점은 모듈화입니다. 검색 함수를 교체하고 싶다면 search_documents만 바꾸면 됩니다.

신뢰도 평가 방식을 변경하고 싶다면 confidence_prompt만 수정하면 됩니다. 각 부분이 독립적이면서도 유기적으로 연결되어 있습니다.

실무에서는 이보다 더 복잡한 체인도 흔합니다. 예를 들어 질문을 여러 언어로 번역해서 검색하고, 결과를 합치고, 중복을 제거하고, 순위를 매기고, 상위 결과만 선택해서 답변을 생성하는 식입니다.

LCEL은 이런 복잡한 흐름도 선언적으로 표현할 수 있게 해줍니다. 디버깅 팁을 하나 드리겠습니다.

복잡한 체인에서 문제가 생기면 중간 단계의 출력을 확인하고 싶을 것입니다. RunnableLambda로 print 함수를 감싸서 체인 중간에 끼워넣으면 데이터 흐름을 추적할 수 있습니다.

주의할 점도 있습니다. 체인이 복잡해질수록 에러 처리가 어려워집니다.

검색이 실패하면? 모델이 이상한 답변을 내놓으면?

각 상황에 대한 폴백 로직을 고려해야 합니다. RunnableWithFallback을 사용하면 에러 시 대체 체인을 실행할 수 있습니다.

김개발 씨는 복잡한 체인을 완성하고 뿌듯해했습니다. "예전 같았으면 함수가 열 개는 필요했을 텐데, LCEL로 하니까 흐름이 한눈에 보이네요!"

실전 팁

💡 - RAG 패턴은 LLM의 한계를 극복하는 가장 효과적인 방법 중 하나입니다

• 복잡한 체인은 중간에 로깅을 추가해서 디버깅하세요
• RunnableWithFallback으로 에러 상황에 대비하세요

---

5. 실습: 커스텀 체인 개발

김개발 씨가 팀의 핵심 개발자로 성장했습니다. 이제는 단순히 체인을 사용하는 것을 넘어, 팀 전체가 재사용할 수 있는 커스텀 컴포넌트를 만들어야 할 때가 왔습니다.

"우리만의 특별한 Runnable을 만들어볼까요?"

커스텀 체인은 반복되는 패턴을 재사용 가능한 컴포넌트로 캡슐화하는 것입니다. Runnable 클래스를 상속받아 invoke 메서드를 구현하면, 표준 LCEL 체인과 완벽하게 호환되는 나만의 컴포넌트를 만들 수 있습니다.

이를 통해 팀 전체의 생산성을 높일 수 있습니다.

다음 코드를 살펴봅시다.

from langchain_core.runnables import Runnable, RunnableConfig
from typing import Any, Optional
import time

class TimedRunnable(Runnable):
    """실행 시간을 측정하는 커스텀 Runnable"""

    def __init__(self, runnable: Runnable, name: str = "chain"):
        self.runnable = runnable
        self.name = name

    def invoke(self, input: Any, config: Optional[RunnableConfig] = None) -> dict:
        start = time.time()
        result = self.runnable.invoke(input, config)
        elapsed = time.time() - start
        return {"result": result, "elapsed_time": f"{elapsed:.2f}s", "name": self.name}

# 사용 예시
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("Tell me about {topic}")
model = ChatOpenAI(model="gpt-4")
chain = prompt | model

# 커스텀 Runnable로 감싸기
timed_chain = TimedRunnable(chain, name="topic_explainer")
result = timed_chain.invoke({"topic": "LCEL"})

김개발 씨는 이제 팀에서 LCEL 전문가로 통합니다. 그런데 비슷한 패턴의 코드가 여러 프로젝트에서 반복되는 것이 눈에 띄었습니다.

매번 실행 시간을 측정하고 로깅하는 코드, 에러를 잡아서 재시도하는 코드, 결과를 캐싱하는 코드... "이런 공통 패턴을 한 번만 만들어두면 모두가 편해지겠는데요." 김개발 씨는 박시니어 씨에게 조언을 구했습니다.

박시니어 씨가 말했습니다. "좋은 생각이에요.

커스텀 Runnable을 만들면 됩니다. 한 번 잘 만들어두면 팀 전체가 레고 블록처럼 가져다 쓸 수 있어요." 커스텀 Runnable을 만드는 방법은 간단합니다.

Runnable 클래스를 상속받고, invoke 메서드를 구현하면 됩니다. 이렇게 만든 컴포넌트는 기존 LCEL 체인과 완벽하게 호환됩니다.

위 코드의 TimedRunnable을 살펴보겠습니다. 이 클래스는 다른 Runnable을 감싸서 실행 시간을 측정합니다.

생성자에서 감쌀 대상 runnable과 이름을 받고, invoke 메서드에서 시간 측정 로직을 추가합니다. invoke 메서드 내부를 보면, 먼저 시작 시간을 기록합니다.

그 다음 내부 runnable의 invoke를 호출하고, 완료 후 경과 시간을 계산합니다. 최종적으로 결과와 함께 실행 시간 정보를 딕셔너리로 반환합니다.

이 패턴의 장점은 투명성입니다. TimedRunnable로 감싼 체인은 기존과 똑같이 동작하면서, 추가로 성능 정보를 제공합니다.

기존 코드를 전혀 수정하지 않고도 모니터링 기능을 추가할 수 있는 것입니다. 실무에서 유용한 커스텀 Runnable 아이디어를 몇 가지 소개하겠습니다.

RetryRunnable은 실패 시 자동으로 재시도합니다. CachingRunnable은 동일한 입력에 대해 캐시된 결과를 반환합니다.

LoggingRunnable은 입출력을 로깅합니다. RateLimitRunnable은 API 호출 속도를 제한합니다.

더 고급 기능도 구현할 수 있습니다. batch, stream, ainvoke 등의 메서드도 오버라이드하면 배치 처리나 스트리밍에서도 커스텀 로직이 동작합니다.

다만 모든 메서드를 구현하는 것은 복잡할 수 있으니, 필요한 것만 선택적으로 구현하세요. 주의할 점이 있습니다.

커스텀 Runnable을 만들 때는 config 매개변수를 반드시 내부 runnable에 전달해야 합니다. 이 config에는 콜백, 태그 등 중요한 런타임 정보가 담겨 있습니다.

전달하지 않으면 LangSmith 같은 도구에서 추적이 제대로 되지 않을 수 있습니다. 김개발 씨는 팀을 위한 커스텀 Runnable 라이브러리를 만들었습니다.

이제 팀원들은 공통 패턴을 매번 작성하는 대신, 검증된 컴포넌트를 가져다 쓸 수 있게 되었습니다. 코드 품질은 올라가고, 개발 속도도 빨라졌습니다.

"LCEL의 진정한 힘은 확장성에 있었군요." 김개발 씨는 만족스럽게 웃었습니다. 이제 그는 단순한 사용자가 아니라, 도구를 만드는 개발자가 된 것입니다.

실전 팁

💡 - 커스텀 Runnable은 팀 전체가 재사용할 공통 패턴에 적합합니다

• invoke 구현 시 config 매개변수를 내부 runnable에 반드시 전달하세요
• 필요한 메서드만 선택적으로 구현해도 충분합니다

---

이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!