[LangChain] LLM 워크플로우 : Routing 구현

Routing은 이전 단계의 결과에따라 다음 단계를 비결정적으로 정의할 수 있도록 한다.  RunnableLambda 혹은 RunnableBranch를 이용하는 방법이있지만, 현재는 RunnableLambda를 사용하는 방법이 권장된다. 

 

RunnableLambda는 Python callable을 Runnable로 바꿔주기만하는데, 별도로 RunnableBranch를 사용하기보다는 Python 코드 분기문을 통해 처리를 하라는 것이다.

 

실습

유저의 질문이 파이썬 혹은 코틀린 프로그래밍과 관련 있을 경우, 이에 최적화된 프롬프트로 답변할 수 있도록 분기처리를 추가해볼 것이다.

 

먼저, 질문이 파이썬이나 코틀린과 관련되어있는지만 확인하는 체인을 준비한다.

from langchain_core.prompts import PromptTemplate

prompt_template = PromptTemplate.from_template(
    """\
    Given the user question below, \
    classify it as either being about `Python` or `Kotlin`. \
    Do not respond with more than one word.
    
    <question>
    {question}
    </question>
    
    Classification:
    """
)

topic_chain = (
    prompt_template
    | llm
)

 

그리고 파이썬, 코틀린, 일반적인 질문에 대해 답변할 수 있는 체인 3가지를 준비한다.

language_prompt_template = """\
    You are an experienced {language} developer.
    Always answer questions starting with as "[{language}Dev] ".
    Respond to the following question:
    
    Question: {question}
    Answer:\
    """

kotlin_chain = (
    PromptTemplate.from_template(language_prompt_template.format(
        language="Kotlin", question="{question}"))
    | llm
)
python_chain = (
    PromptTemplate.from_template(language_prompt_template.format(
        language="Python", question="{question}"))
    | llm
)
general_chain = (
    PromptTemplate.from_template("""\
    Respond to the following question:

    Question: {question}
    Answer:\
    """)
    | llm
)

 

주제가 정해지면 어떤 체인을 실행시킬지 결정하는 route 함수를 작성한다.

def route(info):
    if "python" in info["topic"].strip().lower():
        return python_chain
    elif "kotlin" in info["topic"].strip().lower():
        return kotlin_chain
    else:
        return general_chain

 

 

최종적으로 앞에서 만든 주제를 분류하는 체인과 결합해서, 주제 분류 후 route 함수로 내용이 넘어오도록 RunnableLambda를 사용하여 풀체인을 만들어 실행시킨다.

from langchain_core.runnables import RunnableLambda
from langchain_core.output_parsers import StrOutputParser

full_chain = (
    {"topic": topic_chain, "question": lambda x: x["question"]} | RunnableLambda(route)
)

question = """\
    다음 코드에서 발생하는 오류는 무엇이고, 어떻게 수정해야 할까요?
    
    ```
    def trim_str():
        some = "Thing".trim()
    ```\
    """
    
response = full_chain.invoke(
    {"question": question},
    config={
        "callbacks": [langfuse_handler]
    }
)

print(response)

 

테스트로 python 코드를 첨부했다. python에 존재하지 않지만, kotlin에 존재하는 String.trim() 함수를 작성하고, 이게 왜 틀린건지 질문해보았다.

 

실행 결과

2번의 생성이 있었다. 하나는 토픽을 가져오는것, 하나는 유저에 질문에 답하는 것이다.

 

topic_chain에서 주제에 대해 Python이라고 잘 분류되었고,

 

이어서 python_chain이 실행되어서 프롬프트대로 답변의 앞머리에 "[PythonDev]"가 붙어서 답변이 출력되고 있다. 답변도 정확하다!