[LangChain] LCEL

LangChain Expression Language (LCEL)

• Runnable 객체들을 파이프 (|) 로 연결하는 문법적 표현 (Runnable 객체 연결. Runnable 결과 연결이 아님)
  • 파이프는 Runnable 객들을 연결시킴. 예외적으로 Runnable 이 아닌 callable 객체가 오면 LCEL 파이프에서는 해당 callable 을 내부적으로 RunnableLambda 로 감싸줘 Runnable 객체로 만들어줌
  • | 와 RunnableSequence 는 동일. 내부적으로 | 연산자가 RunnableSequence 를 생성하는 syntactic sugar 표현
    • e.g. prompt | llm | StrOutputParser() 와 RunnableSequence(prompt, llm, StrOutputParser()) 는 동일
• Runnable 객체들을 직관적이고 선언적으로 연결할 수 있도록 설계
• 과거에는 Chain 구성을 위해 명시적인 Chain 클래스를 사용했지만 현재는 LCEL (Runnable 기반 체이닝) 이 표준이자 권장 방식
  • LLMChain 또한 Runnable 객체. 따라서 .invoke 와 같은 Runnable 의 주요 메서드 사용 가능
  • SeuquentialChain 은 Runnable 객체 아님. .stream 같은 메서드는 있지만 형식적일뿐 제대로 동작하지 않음

callable: 호출 가능한 객체 입니다. 즉, 함수이거나 __call__() 을 구현한 객체를 말합니다. callable(obj) 는 obj() 형태로 호출이 가능한지를 판별하는 내장 함수입니다. obj 가 함수 (Function) 자체이거나 객체 (instance) 인데 그 객체의 클래스가 __call__() 메서드를 구현한 경우 True 를 반환합니다.

Runnable

• LCEL 의 기본 단위
• LangChain 생태계에서 실행 가능한 최소 단위이자 핵심 인터페이스
• 모든 체인 구성요소가 이 인터페이스를 공통으로 따름
• 이 인터페이스를 상속받아 구현된 다양한 실행 단위 클래스들 존재
  • e.g. RunnableMap, RunnableLambda, RunnableSequence, RunnableParallel, RunnablePassthrough 및 prompt, llm, parser 등
• 주요 메서드
  • .invoke: 단일 입력 → 단일 출력
  • .batch: 여러 입력을 병렬 실행
    • 내부적으로는 .invoke 를 여러 번 실행하지만 성능상 이점을 위해 비동기/병렬 처리에 최적화되어 있음
  • .stream: 스트리밍 방식으로 토큰 단위 출력
    • 제너레이터 (generator) 를 반환하며 for 문으로 하나씩 토큰을 받을 수 있음
    • 실시간 토큰 단위 출력
    • 토큰이 생성될 때마다 yield
    • OpenAI API 의 stream=True 옵션처럼 동작
    • 사용자 인터페이스나 콘솔에서 실시간 출력 보여줄 때 사용
    • 실시간 출력 가능, 응답 지연 최소화, 메모리 절약, 토큰 단위 제어 가능
    • ChatGPT, Claude 등에서 실제 해당 방식을 사용
    • SeuquentialChain 은 .stream 메서드를 사용 가능하지만 제대로 동작하지 않는데 이는 내부 LLMChain 이 완전 실행된 후 결과만 리턴하기 때문이기도 하며 SeuquentialChain 이 Runnable 이 아니며 이에 구현된 .stream 도 최신 LangChain 과 호환시키기 위한 임시 조치일 뿐이기 때문
  • .ainvoke: invoke 의 비동기 버전
  • .abatch: batch 의 비동기 버전
  • .astream: stream 의 비동기 버전

# 과거 Chain 클래스 방식

prompt1 = PromptTemplate(
    input_variables=["topic"],
    template="Suggest a research question about {topic}.",
)
chain1 = LLMChain(llm=llm, prompt=prompt1, output_key='question')
# print(chain1.invoke(input={'topic': 'soccer'}))

prompt2 = PromptTemplate(
    input_variables=["topic"],
    template="name famous korean {topic} player.",
)
chain2 = LLMChain(llm=llm, prompt=prompt2, output_key='player')

promtp3 = PromptTemplate(
    input_variables=["question, player, star_player"],
    template="give idea of how {player} and {star_player} think of {question}.",
)
chain3 = LLMChain(llm=llm, prompt=promtp3, output_key='answer')

overall_chains = SequentialChain(
    chains=[chain1, chain2, chain3],
    input_variables=['topic', 'star_player',],
    output_variables=['question', 'player', 'answer'],
    verbose=False,
)
pprint(overall_chains.invoke({'topic': 'soccer', 'star_player': 'messi'}))

"""
{'answer': 'Son Heung-min and Messi likely both understand the importance of '
           'physical fitness and endurance in soccer. They would recognize '
           'that a high level of physical fitness and',
 'player': 'Son Heung-min',
 'question': 'How does the level of physical fitness and endurance of soccer '
             'players impact their performance on the field?',
 'star_player': 'messi',
 'topic': 'soccer'}
"""

# 최신 LCEL 방식

QUESTION_TEMPLATE = "Suggest a research question about {sports}."
PLAYER_TEMPLATE = "Name famous korean {sports} player."
ANSWER_TEMPLATE = "Give idea of how {player} and {star_player} think of {question}."
TRANSLATE_TEMPLATE = "Translate following article into {language}. ```{answer}```"

prompt_question = PromptTemplate.from_template(template=QUESTION_TEMPLATE)
prompt_player = PromptTemplate.from_template(template=PLAYER_TEMPLATE)
prompt_answer = PromptTemplate.from_template(template=ANSWER_TEMPLATE)
prompt_translate = PromptTemplate.from_template(template=TRANSLATE_TEMPLATE)

chain_question = prompt_question | llm | StrOutputParser()
chain_player = prompt_player | llm | StrOutputParser()
chain_answer = prompt_answer | llm | StrOutputParser()
chain_translate = prompt_translate | llm | StrOutputParser()

overall_chain = RunnableMap({
    'player': chain_player,
    'question': chain_question,
    'star_player': RunnableLambda(lambda x: x['star_player']),
    'language': RunnableLambda(lambda x: x['language']),
}) | RunnableMap({
    'answer': chain_answer,
    'language': RunnableLambda(lambda x: x['language']),
}) | chain_translate

user_input = {
    'sports': 'soccer',
    'star_player': 'ronaldo',
    'language': 'korean',
}

result = overall_chain.invoke(input=user_input)
print(result)
# 손흥민과 호날두는 둘 다 축구에서 신체적 건강과 인내심의 중요성을 이해하고 있습니다. 그들은 ...

result = overall_chain.stream(input=user_input)
for chunk in result:
    print(chunk, end='', flush=True)
# 손흥민과 호날두는 둘 다 축구에서 신체적 건강과 인내심의 중요성을 이해하고 있습니다. 그들은 ...

RunnableMap 동작 구조

1. invoke() 호출 시 입력 dict 가 RunnableMap 으로 들어감
2. 각 value 가 Runnable 객체면 invoke() 가 실행, 일반 함수/lambda 면 자동으로 RunnableLambda 로 wrapping 되어 실행
3. 최종적으로 입력 dict 를 key 별로 parallel 하게 실행 → 결과 dict 로 반환