[LangChain] RunnableBranch를 사용해 로직에 따라 동적으로 Routing 하기

애플리케이션을 개발하다보면 인풋의 조건, 또는 이전 단계의 결과에 따라 이후 단계에 LLM 모델이 실행해야 하는 과업(즉, 프롬프트)이 동적으로 달라져야 하는 경우가 있습니다. LangChain의 RunnableBranch 를 사용하면 로직에 따라서 인풋의 조건을 평가해서 이후 수행할 runnable을 동적으로 Routing 할 수 있습니다. 

 

아래에는 간단한 예를 들어서 RunnableBranch를 사용해서 Routing 하는 방법을 소개하겠습니다. 

 

(1) 인풋으로 받은 질문의 주제(Topic)를 "LangChain", "Anthropic", "Other" 의 3개 범주로 분류하기

(2) 각 Topic별로 프롬프트를 달리해서 'langchain_chain', 'anthropic_chain', 'general_chain' 만들기

(3) RunnableBranch() 를 사용해서 Topic 분류 결과에 따라 runnable Routing 하기

      - 만약 질문의 Topic이 "LangChain"이면 --> 'langchain_chain' 실행

      - 만약 질문의 Topic이 "Anthropic"이면   --> 'anthropic_chain' 실행

      - 만약 질문의 Topic이 "Other"이면         --> 'general_chain'실행

 

LangChain - RunnableBranch()로 동적으로 Routing 하기

 

 

 

먼저, 실습 환경에 openai, langchain 모듈이 설치가 안되어 있다면, 터미널에서 pip install 을 사용해서 설치를 하기 바랍니다. ('!'는 Jupyter Notebook에서 실행 시 추가)

 

! pip install -q openai langchain

 

 

 

(1) 인풋으로 받은 질문의 주제(Topic)를 "LangChain", "Anthropic", "Other" 의 3개 범주로 분류하기

 

1단계로서, 아래의 예처럼 template 에 인풋으로 받은 질문을 "LangChain", "Anthropic", "Other"의 3개 범주로 분류하고 단답형으로만 답하라고 지시를 했습니다.(classifier 모델이 따로 필요없다는게 놀랍습니다!) 그리고 '|' 를 사용해서 topic_chain = Prompt | Model | OutputParser를 chaining 하였습니다. 

 

from langchain.chat_models import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

OPENAI_API_KEY = "sk-xxxxx...." # set with yours

## Create a chain that will identify incoming questions as being about 'LangChain', 'Anthropic', 'Other'.
template = """Given the user question below, classify it \
as either being about 'LangChain', 'Anthropic', or 'Other'.

Do not respond with more than one word.

<question>
{question}
</question>

Classification:"""

prompt = PromptTemplate.from_template(template)
model = ChatOpenAI(temperature=0.0, model="gpt-3.5-turbo", openai_api_key=OPENAI_API_KEY)
parser = StrOutputParser()

topic_chain =  prompt | model | parser

 

 

invoke() 를 사용해서 topic_chain을 실행시켜보니 "LangChain", "Anthropic", "Other"의 3개 범주로 Topic을 제대로 분류하고 있네요. 

 

## Run the chains

topic_chain.invoke({"question": "How do I call LangChain?"})
# LangChain

topic_chain.invoke({"question": "How do I call Anthropic?"})
# Anthropic

topic_chain.invoke({"question": "What is 2 + 5?"})
# Other

 

 

 

 

(2) 각 Topic별로 프롬프트를 달리해서 'langchain_chain', 'anthropic_chain', 'general_chain' 만들기

 

이제 질문을 받았을 때 각 Topic 별로 프롬프트에 지시문을 다르게 해서 만들고, '|'로 서로 다른 Prompt + Model + OutputParser 를 chaining 해서 3개의 sub chains runnable을 만들었습니다. 

 

## Create three sub chains.
## (1) Topic: LangChain
langchain_chain = (
    PromptTemplate.from_template(
        """You are an expert in langchain.\
        Always answer questions starting with "Sure. I'm an expert in LangChain". \
        Respond to the following question:

        Question: {question}
        Answer:"""
    )
    | ChatOpenAI(openai_api_key=OPENAI_API_KEY)
    | StrOutputParser()
)

# (2) Topic: Anthropic
anthropic_chain = (
    PromptTemplate.from_template(
        """You are an expert in Anthropic. \
        Always answer questions staring with "Sure. I'm an expert in Anthropic". \
        Respond to the following question:

        Question: {question}
        Answer:"""
    )
    | ChatOpenAI(openai_api_key=OPENAI_API_KEY)
    | StrOutputParser()
)

# (3) Topic: Other
general_chain = (
    PromptTemplate.from_template(
        """Respond to the following question:

        Qeustion: {question}
        Answer:"""
    )
    | ChatOpenAI(openai_api_key=OPENAI_API_KEY)
    | StrOutputParser()
)

 

 

 

(3) RunnableBranch() 를 사용해서 Topic 분류 결과에 따라 runnable Routing 하기

 

마지막 단계로서, 앞의 단계에서 만든 구슬을 꿰어서 목걸이를 만들어보겠습니다. 

RunnableBranch 는 (condition, runnable) 의 쌍으로 로직을 구성하고 Routing을 합니다. 

 

                      조건 (condition)                    Routing     실행 (runnable)

      - 만약 질문의 Topic이 "LangChain"이면 ----------> 'langchain_chain' 실행

      - 만약 질문의 Topic이 "Anthropic"이면   ----------> 'anthropic_chain' 실행

      - 만약 질문의 Topic이 "Other"이면         ----------> 'general_chain'실행

 

## Dynamically route logic based on input using RunnableBranch()
from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: "langchain" in x["topic"].lower(), langchain_chain), # a list of (condition, runnable) pair
    (lambda x: "anthropic" in x["topic"].lower(), anthropic_chain), 
    general_chain
)

# Full Chain: topic_chain + branch
full_chain = {"topic": topic_chain, "question": lambda x: x["question"]} | branch

 

 

 

위에서 RunnableBranch를 사용해서 정의한 full_chain 을 invoke() 를 사용해서 실행시켜 보았습니다. 3개 Topic 별로 질문을 던져보니 잘 작동함을 확인할 수 있습니다. 

 

## Run full_chain runnable

## (1) Topic: 'LangChain' --> route to 'langchain_chain'
full_chain.invoke({"question": "How do I call LangChain?"})

# Sure. I'm an expert in LangChain. To call LangChain, you can use the LangChain API 
# or SDK provided by its developers. These tools allow you to interact with LangChain's 
# language processing capabilities programmatically. You would need to integrate the API 
# or SDK into your application or codebase and make the necessary API calls or 
# function calls to utilize LangChain's features.


## (2) Topic: 'Anthropic' --> route to 'anthropic_chain' 
full_chain.invoke({"question": "How do I call Anthropic?" })

# Sure. I'm an expert in Anthropic. Anthropic refers to the study of the relationship 
# between humans and their environment. It can also refer to the idea that the universe 
# is fine-tuned for the existence of human life. Therefore, you can call Anthropic 
# as the scientific study of human-environment interaction or the philosophical concept 
# of the fine-tuning of the universe.


## (3) Topic: 'Other' --> route to 'general_chain'
full_chain.invoke({"question": "What is 2 + 5?"})

# The sum of 2 and 5 is 7.

 

 

[ Reference ]

* LangChain Tutorial - RunnableBranch()

: https://python.langchain.com/docs/expression_language/how_to/routing#using-a-runnablebranch

 

 

이번 포스팅이 많은 도움이 되었기를 바랍니다.

행복한 데이터 과학자 되세요!  :-)