R, Python 분석과 프로그래밍의 친구 (by R Friend)

애플리케이션을 개발하다보면 인풋의 조건, 또는 이전 단계의 결과에 따라 이후 단계에 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'실행

먼저, 실습 환경에 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

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

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