러너블을 병렬로 실행하기

러너블을 병렬로 실행하기 (How to invoke runnables in parallel)

체인에 여러 독립적인 작업이 섞여 있을 때, 하나씩 순서대로 돌리는 건 느려요. RunnableParallel을 쓰면 서로 의존하지 않는 러너블들을 동시에 실행해 지연 시간을 줄일 수 있어요. 이번 가이드에서는 병렬 실행과 함께, 다음 단계가 요구하는 입력 형태로 출력을 맞춰주는 방법까지 살펴볼게요.

출처: 공식문서

사전 준비

이 가이드를 따라가려면 다음 개념에 익숙해져 있으면 좋아요.

LangChain Expression Language (LCEL)
러너블 이어 붙이기 (Chaining runnables)

RunnableParallel은 본질적으로 값이 러너블인 dict예요. (함수처럼 러너블로 변환될 수 있는 것도 값으로 넣을 수 있어요.) 이 dict의 모든 값을 병렬로 실행하는데, 각 값은 RunnableParallel 전체의 입력을 그대로 받아요. 최종 반환값은 각 값의 결과를 해당 키 아래에 담은 dict입니다.

`RunnableParallel`로 입력 포맷 맞추기

RunnableParallel은 작업을 병렬화하는 데 유용하지만, 한 러너블의 출력을 시퀀스의 다음 러너블이 기대하는 입력 형태로 바꾸는 데도 쓸 수 있어요. 체인을 갈라서(fork) 여러 구성 요소가 입력을 병렬로 처리하게 하고, 나중에 다른 구성 요소가 결과를 합쳐(join) 최종 응답을 만들어내는 방식이죠. 이런 체인은 다음과 같은 계산 그래프를 만들게 됩니다.

     Input
      / \
     /   \
 Branch1 Branch2
     \   /
      \ /
      Combine

아래 예시에서 prompt가 기대하는 입력은 "context"와 "question"이라는 키를 가진 map이에요. 그런데 사용자 입력은 질문 하나뿐이죠. 그래서 retriever로 컨텍스트를 얻고, 사용자 입력은 "question" 키 아래로 그대로 통과시켜야 해요.

from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

vectorstore = FAISS.from_texts(
    ["harrison worked at kensho"], embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()
template = """Answer the question based only on the following context:
{context}

Question: {question}
"""

# The prompt expects input with keys for "context" and "question"
prompt = ChatPromptTemplate.from_template(template)

model = ChatOpenAI()

retrieval_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

retrieval_chain.invoke("where did harrison work?")

API Reference: StrOutputParser | ChatPromptTemplate | RunnablePassthrough

'Harrison worked at Kensho.'

팁: RunnableParallel을 다른 러너블과 조합할 때는 dict를 굳이 RunnableParallel 클래스로 감싸지 않아도 돼요. 타입 변환이 자동으로 처리되거든요. 체인 안에서는 아래 세 가지가 모두 동일합니다.

{"context": retriever, "question": RunnablePassthrough()}

RunnableParallel({"context": retriever, "question": RunnablePassthrough()})

RunnableParallel(context=retriever, question=RunnablePassthrough())

타입 강제 변환에 대한 자세한 내용은 러너블 이어 붙이기 가이드의 타입 강제 변환 섹션을 참고하세요.

`itemgetter`를 축약으로 쓰기

RunnableParallel과 함께 쓸 때 Python의 itemgetter를 축약 표현으로 사용해 map에서 데이터를 꺼낼 수 있어요. itemgetter에 대한 자세한 내용은 Python 공식 문서를 참고하세요.

아래 예시는 itemgetter로 map에서 특정 키들을 꺼내 사용해요.

from operator import itemgetter

from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

vectorstore = FAISS.from_texts(
    ["harrison worked at kensho"], embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()

template = """Answer the question based only on the following context:
{context}

Question: {question}

Answer in the following language: {language}
"""
prompt = ChatPromptTemplate.from_template(template)

chain = (
    {
        "context": itemgetter("question") | retriever,
        "question": itemgetter("question"),
        "language": itemgetter("language"),
    }
    | prompt
    | model
    | StrOutputParser()
)

chain.invoke({"question": "where did harrison work", "language": "italian"})

API Reference: StrOutputParser | ChatPromptTemplate | RunnablePassthrough

'Harrison ha lavorato a Kensho.'

위 예시에서 itemgetter("question") | retriever처럼, itemgetter로 꺼낸 값을 다른 러너블과 다시 |로 이어 붙일 수도 있어요.

여러 단계 병렬화하기

RunnableParallel을 쓰면 여러 러너블을 병렬로 실행하고, 그 출력을 map으로 돌려받기 쉽습니다.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel
from langchain_openai import ChatOpenAI

model = ChatOpenAI()
joke_chain = ChatPromptTemplate.from_template("tell me a joke about {topic}") | model
poem_chain = (
    ChatPromptTemplate.from_template("write a 2-line poem about {topic}") | model
)

map_chain = RunnableParallel(joke=joke_chain, poem=poem_chain)

map_chain.invoke({"topic": "bear"})

API Reference: ChatPromptTemplate | RunnableParallel

{'joke': AIMessage(content="Why don't bears like fast food? Because they can't catch it!", response_metadata={'token_usage': {'completion_tokens': 15, 'prompt_tokens': 13, 'total_tokens': 28}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_d9767fc5b9', 'finish_reason': 'stop', 'logprobs': None}, id='run-fe024170-c251-4b7a-bfd4-64a3737c67f2-0'),
 'poem': AIMessage(content='In the quiet of the forest, the bear roams free\nMajestic and wild, a sight to see.', response_metadata={'token_usage': {'completion_tokens': 24, 'prompt_tokens': 15, 'total_tokens': 39}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None}, id='run-2707913e-a743-4101-b6ec-840df4568a76-0')}

병렬성의 이점

RunnableParallel은 독립적인 프로세스를 병렬로 실행할 때도 유용해요. map 안의 각 러너블이 병렬로 실행되기 때문이죠. 아래 예시에서 joke_chain, poem_chain, map_chain의 실행 시간이 거의 같은 걸 확인할 수 있어요. map_chain은 앞선 두 체인을 둘 다 실행하는데도 말이죠.

%%timeit

joke_chain.invoke({"topic": "bear"})

610 ms ± 64 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

%%timeit

poem_chain.invoke({"topic": "bear"})

599 ms ± 73.3 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

%%timeit

map_chain.invoke({"topic": "bear"})

643 ms ± 77.8 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

다음 단계

이제 RunnableParallel로 체인 단계를 포맷하고 병렬화하는 몇 가지 방법을 알게 됐어요. 더 배우고 싶다면 이 섹션의 다른 러너블 how-to 가이드를 확인해 보세요.

러너블을 병렬로 실행하기