LangChain - Dynamic Chain

How to create a dynamic (self-constructing) chain

• chain의 일부를 입력값에 따라 runtime에 할당하고 싶을때 사용하는 기법이다.
• RunnableLambda의 속성을 활용해서 Dynamic chain을 구성 할 수 있다.

실습

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model = "gpt-3.5-turbo")

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable, RunnablePassthrough, chain

contextualize_instructions = """
Convert the latest user question into a standalone question given the chat history. 
Don't answer the question, return the question and nothing else (no descriptive text).
"""

contextualize_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", contextualize_instructions),
        ("placeholder", "{chat_history}"),
        ("human","{question}"),
    ]
)

contextualize_question = contextualize_prompt | llm | StrOutputParser()

위에서 contextualize_instructions를 통해서 채팅 기록을 기반으로 새로운 질문을 생성하는 chain을 구성한다.

qa_instructions = (
    """Answer the user question given the following context:\n\n{context}."""
)

qa_prompt = ChatPromptTemplate.from_messages(
    [('system',qa_instructions),('human','{question}')]    
)

Chat Prompt Template을 구성한다.

@chain
def contextualize_if_needed(input_ : dict) -> Runnable:
    if input_.get("chat_history"):
        #이 과정을 통해서 실제 출력값을 생성하는건 아니고 또 다른 Runnable을 반환한다.
        return contextualize_question
    else:
        return RunnablePassthrough()

@chain
def fake_retriever(input_ : dict) -> str:
    return "egypt's population in 2024 is about 111 million"

full_chain = (
    RunnablePassthrough.assign(qusetion = contextualize_if_needed).assign(
        context=fake_retriever
    )
    | qa_prompt
    | llm
    | StrOutputParser()
)

• 만약에 RunnablePassThrough를 통해서 입력된 dict() 객체에 "chat_history"라는 key가 있다면 contextualize_question이라는 Runnable을 반환한다.
  • 반환된 Runnable은 full chain이 실행되면 스스로 실행된다.
• 만약에 없다면 RunnablePassThrough를 반환한다.

여기서 Runnable이란 invoked, batched, streamed, transformed and composed 될 수 있는 일의 단위이다.

• invoke/ainvoke: 하나의 input을 output으로 변경시켜주는 메서드
• batch/abatch: 여러 개의 input을 output으로 변경시켜주는 메서드
• stream/astream: 결과물을 stream 해준다.
• astream_log: 출력 및 입력에서 선택된 중간 결과를 Stream 해준다.

Runnable을 구성하는 주요 primitive(구성 요소)는 RunnableSequence and RunnableParallel이다.

• RunnableSequence는 한 Runnable의 결과물을 다음 과정의 input으로 사용함으로써 Runnable들을 순차적으로 invoke한다.
  • | 연산자를 사용하거나 Runnable의 list를 통해서 구성한다.
• RunnableParallel은 동일한 input을 여러 Runnable들에 전달함으로써 Runnable들을 동시에 invoke한다.
  • dict 타입을 활용해서 구성한다.

출처 : 링크

full_chain.invoke(
    {
        "question": "what about egypt",
        "chat_history": [
            ("human", "what's the population of indonesia"),
            ("ai", "about 276 million"),
        ],
    }
)

결과

"Egypt's population in 2024 is about 111 million."

for chunk in contextualize_if_needed.stream(
    {
        "question" : "what about egypt",
        "chat_history": [
            ("human", "what's the population of indonesia"),
            ("ai", "about 276 million"),
        ],
    }
):
    print(chunk)

결과

What
 is
 the
 population
 of
 Egypt
?