[LLM] Chain - LCEL

1. Chain 개념

 : 여러 컴포넌트를 정해진 순서대로 연결해 복잡한 AI 작업을 단계별로 자동화하는 구조
 - 각 컴포넌트는 입력으로 이전 처리 결과를 받아 처리 후 다음 단계로 전달
 • 예시 흐름 : 질문 → 검색 → 요약 → 응답 생성
 • 장점 : 작업 흐름의 명확성, 코드 재사용성, 유지보수 및 확장성 향상
 • 한계 : 순차 실행 구조 → 조건 분기, 반복 등 복잡한 흐름은 LangGraph 필요

 
2. Chain 구성 방식 비교

방식	설명	상태
Off-the-shelf	LLMChain, SequentialChain	Deprecated
LCEL	| 연산자로 컴포넌트를 선언적 연결	권장

# Off-the-shelf(구 방식)
chain = LLMChain(prompt=prompt, llm=model, output_parser=parser)

# LCEL(신 방식)
chain = prompt | model | parser

 
3. LCEL과 Runnable 인터페이스

 : LCEL로 연결되는 모든 컴포넌트는 Runnable 인터페이스 상속

메소드	설명
invoke(input)	단일 입력 처리
batch(inputs)	다수 입력 한 번에 처리
stream(input)	스트리밍 방식으로 응답 반환
ainvoke() / abatch() / astream()	비동기 버전
assign(**kwargs)	앞 단계 출력 dict에 새 필드 추가해 전달

 
4. 주요 Runnable 구현체

4-1. RunnableSequence
 : 여러 Runnable을 순차 실행
 -  LCEL( | )로 체인 구성 시 자동 생성됨

chain = prompt | model | parser
# 내부적으로 RunnableSequence(prompt, model, parser)와 동일

 
4-2. RunnableLambda
 : 일반 함수 or lambda 표현식 → Runnable
 - 일반 함수는 chain에 직접 포함시켜도 자동 변환(굳이 명시 필요 X)

# 1. lambda
c1 = RunnableLambda(lambda x: f"{x}에 대해 한 문장으로 설명해줘.")

# 2. 일반 함수 - 직접 포함 가능
def get_value_len(value: str):
    return value, len(value)
    
chain = prompt | model | parser | get_value_len

 

4-3. RunnablePassthrough
 : 입력 데이터를 변경 없이 그대로 다음 단계로 전달
 - .assign(key=Runnable, ...) : 기존 dict에 새 필드 추가해 전달

# 그대로 통과
rp = RunnablePassthrough()

# 필드 추가
rp2 = RunnablePassthrough.assign(
    address = RunnableLambda(lambda x: "서울시 금천구"),
    tel_no = RunnableLambda(lambda x: "010-1111-2222")
)
rp2.invoke({"name":"홍길동"})
# -> {"name": "홍길동", "address": "서울시 금천구", "tel_no": "010-1111-2222"}

 

4-4. RunnableParallel
 : 여러 Runnable을 병렬 실행 후 결과를 dict로 반환
 - LCEL에서 chain 내 {} dict로 정의 시 자동으로 RunnableParallel로 처리

parallel = RunnableParallel({
    "value1": RunnableLambda(lambda x: x + 10),
    "value2": RunnableLmabda(lambda x: x - 10),
    "org_value": RunnablePassthroungh()
})
parallel.invoke(200)
# -> {"value1": 210, "value2": 190, "org_value": 200}

 

5. Chain간 연결

 : Chain도 Runnable 타입으로, 다른 chain 구성 요소로 포함 가능
 ex. itemgetter로 입력 dict에서 특정 키 값만 추출해 전달

from operator import itemgetter

# recipe_chain + translate_chain 연결
chain = {
    "content": recipe_chain  # 음식명으로 레시피 생성
    "language": itemgetter("language"_)  # 입력에서 언어값 추출
} | translate_chain

chain.invoke({"food": "김치찌개", "language": "독일어"})

 
6. @chain 데코레이터

 : 표현하기 어려운 조건 분기, 반복, 복합 로직 필요 시 사용 일반 함수를 RunnableLambda 타입으로 만듦

from language_core.runnables import chain

@chain
def multi_language_recipe_chain(input_data: dict) -> dict:
    food = input_data['food']
    language = input_data['language']
    is_korean = input_data['is_korean']
    
    korean_recipe = recipe_chain.invoke({"food": food})
    result = translate_chain.invoke({"content": korean_recipe, "language": language})
    
    final_result = {"recipe": result}
    if is_korean:
        final_result["korean_recipe"] = korean_recipe
      return final_result

LCEL : 단순 파이프라인, @chain : 흐름 제어 필요한 복잡한 에이전트 로직에 적합

 
7. LLM Cache

 : 동일한 질문에 대해 LLM 재호출이 아닌, 저장된 응답 반환해 처리 속도 및 비용 절감

from langchain_core.globals import set_llm_cache
from langchain_community.cache import InMemoryCache, SQLiteCache

set_llm_cache(InMemoryCache())  # 메모리 캐시(휘발성)
set_llm_cache(SQLiteCache("cache.sqlite"))  # 파일 캐시(영속성)

 • 설정 한 번만 하면 이후 모든 LLM 호출에 작용 저동
 • Chatbot처럼 유사 질문이 반복되는 상황에 효과적