Intro
지난 포스트에서 LCEL 인터페이스를 살펴보면서 Runnable한 객체를 실행하는 여러 방법에 대해 살펴보았습니다. LCEL로 생성한 Chain도 결국 Runnable한 객체라는 것도 확인했죠.
이번 포스트에서는 Chain을 구성하면서 사용할 수 있는 다양한 Runnable 클래스에 대해 알아보려고 합니다. 크게 다음 3가지에 대해 정리할 예정입니다.
1. RunnableParallel
2. RunnablePassthrough
3. RunnableLambda
1. RunnableParallel
RunnableParallel는 LCEL을 활용해 생성한 복수의 chain을 병렬로 동시에 수행할 수 있도록 해주는 클래스입니다.
각 Chain에 대한 key를 설정해 RunnableParallel의 파라미터로 입력해줌으로써 동시에 실행할 수 있는 객체를 생성하게 됩니다. 이 때 각 체인에 대한 key값은 각 chain에 입력되는 변수값이 아니라, 각 chain을 실행한 결과를 해당 key값에 넣어주기 위해 입력합니다.
1-1. RunnableParallel 기본 사용법
먼저 Chain 2개를 생성해보겠습니다.
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
chain_1 = (
PromptTemplate.from_template("{country}의 수도는?")
| ChatOpenAI(model="gpt-4o-mini", temperature=0.1)
| StrOutputParser()
)
chain_2 = (
PromptTemplate.from_template("{country}의 면적은?")
| ChatOpenAI(model="gpt-4o-mini", temperature=0.1)
| StrOutputParser()
)LCEL로 연결한 chain로 Runnable한 객체이기 때문에 RunnableParallel을 활용해 병렬처리 객체를 생성할 수 있습니다.
from langchain_core.runnables import RunnableParallel
combined = RunnableParallel(capital=chain_1, area=chain_2)이는 다음과 같이 실행할 수 있습니다.
combined.invoke({"country" : "미국"})1-2. 변수명이 다른 Chain 병렬 처리
그럼 두 chain에서 요구하는 PromptTemplate의 변수값이 다르면 어떻게 입력해야할까요?
먼저 변수 값이 다른 2개의 Chain을 생성해보겠습니다.
chain_1 = (
PromptTemplate.from_template("{country1}의 수도는?")
| ChatOpenAI(model="gpt-4o-mini", temperature=0.1)
| StrOutputParser()
)
chain_2 = (
PromptTemplate.from_template("{country2}의 면적은?")
| ChatOpenAI(model="gpt-4o-mini", temperature=0.1)
| StrOutputParser()
)
combined = RunnableParallel(capital=chain_1, area=chain_2)이런 경우, 단순히 각 변수명에 대한 값을 key-value 쌍으로 만들어 하나의 python dictionary로 만들어 입력하면 됩니다.
combined.invoke({"country1" : "미국", "country2" : "대한민국"})1-3. RunnableParrallel 배치 처리
RunnableParrallel로 생성한 chain 병렬 실행 객체도 결국은 Runnable한 객체입니다. 표준화된 인터페이스에 따라 호출할 수 있으므로, batch 메서드로 배치처리도 가능하겠죠? 다음과 같이 사용합니다.
combined.batch([
{"country1" : "미국", "country2" : "대한민국"},
{"country1" : "대한민국", "country2" : "일본"},
{"country1" : "일본", "country2" : "중국"}
])각 Chain에 대한 실행 결과가 여러 개 나오게 됩니다. 따라서 다음과 같은 출력 형태를 가집니다.
[{'capital': '미국의 수도는 워싱턴 D.C.입니다.',
'area': '대한민국의 면적은 약 100,210 평방킬로미터입니다. 이는 한반도의 남쪽 부분에 해당하며, 북한과의 경계를 포함한 면적입니다.'},
{'capital': '대한민국의 수도는 서울입니다.',
'area': '일본의 면적은 약 377,975 평방킬로미터입니다. 이는 일본이 세계에서 62번째로 큰 나라임을 의미합니다.'},
{'capital': '일본의 수도는 도쿄(東京)입니다.',
'area': '중국의 면적은 약 9,596,961 평방킬로미터입니다. 이는 세계에서 세 번째로 큰 국가로, 러시아와 캐나다에 이어 위치하고 있습니다.'}]2. RunnablePassthrough
RunnablePassthrough는 입력된 값을 그대로 전달하는 역할을 수행합니다. 어떤 값이 입력되면 RunnablePassthrough가 위치한 그 자리에 해당 값을 그대로 남겨줍니다.
RunnablePassthrough는 보통 chain에서 입력할 Prompt key 값이 너무 많거나 기억하기 힘들 때 많이 사용됩니다. key 값 지정없이 특정 위치에 전달된 값을 그대로 전달하고 싶을 때 사용하는 것이죠.
간단하게 RunnablePassthrough의 동작을 확인해보겠습니다. RunnablePassthrough도 Runnable한 객체이니까 invoke로 실행할 수 있습니다.
from langchain_core.runnables import RunnablePassthrough
RunnablePassthrough().invoke({"num" : 10}){'num': 10}이렇게 넣은 전달한 값을 그대로 전달하는 것을 확인할 수 있습니다.
이제 통상적으로 사용하는 Chain에서의 RunnablePassthrough의 디자인 패턴을 살펴보겠습니다.
chain = (
{"num" : RunnablePassthrough()}
| PromptTemplate.from_template("{num}의 약수를 모두 알려줘.")
| ChatOpenAI(model="gpt-4o-mini", temperature=0)
| StrOutputParser()
)
chain.invoke(10)이렇게 작성하면 invoke() 메서드에 전달한 10이 RunnablePassthrough()에 전달되어 해당 위치에 10을 전달하게 되고, {"num" : 10}으로 완성된 입력 dictionary가 PromptTemplate에 전달되어 전체 Chain이 실행되는 방식으로 동작합니다.
답변 결과도 살펴보죠.
'10의 약수는 1, 2, 5, 10입니다.'마지막으로 한가지만 더 살펴보겠습니다. RunnablePassthrough.assign()에 대해 알아보겠습니다. 이는 입력 값으로 들어온 값의 key/value 쌍과 특정 처리를 마친 key/value 쌍을 합쳐서 전달하는 역할을 수행합니다.
즉, 원래 입력과 어떤 처리를 수행한 이후의 값을 모두 입력값으로 전달하는 것이죠.
다음의 예시에서 동작을 확인해보겠습니다.
RunnablePassthrough.assign(new_num=lambda x : x["num"] * 3).invoke({"num" : 1}){'num': 1, 'new_num': 3}이렇게 RunnablePassthrough를 사용하면 Chain의 디자인 패턴을 더욱 다양하게 구성할 수 있습니다.
3. RunnableLambda
RunnableLambda는 들어온 입력값에 어떤 함수처리를 한 결과를 전달하는 역할을 수행합니다. 이를 사용하면 프롬프트 입력으로 들어가기 전에 사용자 정의 함수를 통해 특정 처리를 마친 값을 넣어줄 수 있습니다.
구체적인 사용 방법을 알아보겠습니다.
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from datetime import datetime
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
def get_today(tmp=None):
return datetime.today().strftime("%b-%d")
prompt = PromptTemplate.from_template(
"{today}가 생일인 유명인 {n}명을 나열해주세요. 생년월일도 함께 표기해주세요."
)
model = ChatOpenAI(
model="gpt-4o-mini",
temperature=0.1
)
chain = (
{"today" : RunnableLambda(get_today), "n" : RunnablePassthrough()}
| prompt
| model
| StrOutputParser()
)여기서 사용자 정의 함수에 한번 주목해보겠습니다. 왜 내부에서 사용하지도 않는 입력 값을 두었을까요? 이는 RunnableLambda의 동작 방식 때문입니다.
RunnableLambda는 입력값을 사용자 정의 함수를 통해 처리한 값을 전달하는 것을 염두에 두고 구현되었습니다. 따라서, 사용자 정의 함수를 RunnableLambda로 감싸서 객체로 만들면 사용자가 입력한 값이 무조건 함수의 입력 인자로 입력되도록 구현되었습니다.
따라서 내부에서 입력값을 사용하지 않아도 이 입력값을 받아줄 변수가 1개는 필요하게 됩니다.
이제 세부 동작 방식을 살펴보겠습니다.
{"today" : RunnableLambda(get_today), "n" : RunnablePassthrough()}이렇게 구현된 chain의 첫부분에chian.invoke(3)과 같이 호출하면RunnableLambda(get_today)의 실행결과로 오늘 날짜가 반환되고,RunnablePassthrough()의 동작으로 3이 그대로 해당 자리에 남겨져서- 최종적으로 PromptTemplate에 입력되는 dictionary
{"today" : "<날짜>", "n" : 3}가 완성됩니다. - 이후로 전체 Chain 이 실행되어 응답을 반환하게 됩니다.
이렇게 입력값을 PromptTemplate에 넣어주기전 특정 처리를 거치도록 구현할 수 있습니다.
Outro
오늘 알아본 RunnableParallel, RunnablePassthrough, RunnableLambda 는 많이 사용하는 것들이기 때문에 공부하고 익숙해지는 것이 중요하겠습니다.
이를 통해 단순히 컴포넌트를 연결하는 것을 넘어 더욱 복잡한 형태로 구성할 수 있을 것입니다.
출처
