RAG & LANCHAIN (3)- Langchain 개념 & 문법 정리 CH13 LangChain Expression Language(LCEL)

이영락·2024년 9월 3일
0

인공지능 공부

목록 보기
21/33

목차

  1. RunnablePassthrough 개요
  2. API 키 관리 설정
  3. 데이터 전달과 활용
  4. RunnablePassthrough와 RunnableParallel의 사용 예시
  5. 검색기 예제: FAISS 벡터 저장소와의 통합

🏖️ RunnablePassthrough 개요

  • RunnablePassthrough 개념

    • RunnablePassthrough는 데이터를 전달하는 역할을 하는 클래스이다.
    • 이 클래스는 invoke() 메서드를 통해 입력된 데이터를 그대로 반환한다.
    • 이를 통해 데이터를 변경하지 않고 파이프라인의 다음 단계로 전달할 수 있다.

  • 사용 시나리오
    RunnablePassthrough는 다음과 같은 상황에서 유용하다:

    • 데이터를 변환하거나 수정할 필요가 없는 경우: 데이터를 그대로 전달해야 할 때 사용된다.
    • 파이프라인의 특정 단계를 건너뛰어야 하는 경우: 특정 단계에서 데이터 변환이 필요하지 않을 때 유용하다.
    • 디버깅 또는 테스트 목적으로 데이터 흐름을 모니터링해야 하는 경우: 데이터의 흐름을 추적하고 문제를 찾는 데 도움이 된다.

  • 파이프라인에서의 활용

    • RunnablePassthrough는 Runnable 인터페이스를 구현하므로 다른 Runnable 객체와 함께 파이프라인에서 사용할 수 있다.
    • 데이터를 그대로 전달하거나, 추가적인 데이터 키를 더하는 작업을 수행할 수 있다.

🏖️ API 키 관리 설정

  • API 키를 환경변수로 관리하기 위한 설정

    • API 키 관리를 위해 dotenv 라이브러리를 사용하여 환경변수로 설정할 수 있다.

    • 아래와 같이 설정 파일을 통해 API 키를 로드할 수 있다:

      from dotenv import load_dotenv

# API 키 정보 로드
load_dotenv()

  • LangSmith 추적 설정

    • LangSmith 추적을 설정하여 프로젝트를 관리하고 로그를 기록할 수 있다.

    • 이를 통해 API 호출과 데이터 흐름을 추적할 수 있으며, 디버깅과 분석에 유용하다:

      from langchain_teddynote import logging

# 프로젝트 이름을 입력합니다.
logging.langsmith("LCEL-Advanced")

🏖️ 데이터 전달과 활용

  • 데이터 전달 방식

    • RunnablePassthrough는 입력된 데이터를 변경하지 않고 그대로 전달하거나, 추가 키를 더하여 전달할 수 있다.
    • 이 클래스는 RunnableParallel과 함께 사용되어 데이터를 맵의 새로운 키에 할당하는 데 활용된다.

  • 단독 호출

    • RunnablePassthrough()를 단독으로 호출하면 입력을 그대로 전달한다.
    • 예를 들어, RunnablePassthrough()는 단순히 입력 데이터를 받아 그대로 반환한다.

  • assign 메서드와 함께 호출

    • RunnablePassthrough.assign(...)을 사용하면 입력을 받아 assign 함수에 전달된 추가 인자를 더할 수 있다.
    • 예를 들어, 특정 입력 값에 대해 추가적인 계산을 수행하여 새로운 키에 할당할 수 있다.

RunnablePassthrough와 RunnableParallel의 사용 예시

  • 병렬 작업 정의
    RunnableParallel 클래스를 사용하여 병렬로 실행 가능한 작업을 정의할 수 있다:

    from langchain_core.runnables import RunnableParallel, RunnablePassthrough

runnable = RunnableParallel(
    # 전달된 입력을 그대로 반환하는 Runnable을 설정합니다.
    passed=RunnablePassthrough(),
    # 입력의 "num" 값에 3을 곱한 결과를 반환하는 Runnable을 설정합니다.
    extra=RunnablePassthrough.assign(mult=lambda x: x["num"] * 3),
    # 입력의 "num" 값에 1을 더한 결과를 반환하는 Runnable을 설정합니다.
    modified=lambda x: x["num"] + 1,
)

  • RunnableParallel 동작

    • 이 코드에서 passed 속성은 RunnablePassthrough 인스턴스를 할당하여 입력을 그대로 반환하도록 설정된다.
    • extra 속성은 RunnablePassthrough.assign() 메서드를 사용하여 입력의 "num" 값에 3을 곱한 결과를 "mult" 키에 할당하는 작업을 정의한다.
    • modified 속성은 람다 함수를 사용하여 입력의 "num" 값에 1을 더하는 작업을 정의한다.

  • 병렬 작업 실행

    • runnable.invoke() 메서드를 호출하여 {"num": 1} 입력으로 병렬 작업을 실행한다.
    • 이 결과로 각 작업의 출력이 결합된 딕셔너리가 반환된다:
      {'passed': {'num': 1}, 'extra': {'num': 1, 'mult': 3}, 'modified': 2}

🏖️ 검색기 예제: FAISS 벡터 저장소와의 통합

  • FAISS 벡터 저장소 생성

    • 텍스트로부터 FAISS 벡터 저장소를 생성하여, 검색기로 사용할 수 있다:

      from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

vectorstore = FAISS.from_texts(
    [
        "테디는 랭체인 주식회사에서 근무를 하였습니다.",
        "셜리는 테디와 같은 회사에서 근무하였습니다.",
        "테디의 직업은 개발자입니다.",
        "셜리의 직업은 디자이너입니다.",
    ],
    embedding=OpenAIEmbeddings(),
)

    • 이 코드에서는 FAISS 벡터 저장소를 생성하여 텍스트 데이터를 벡터로 변환하고 저장소에 저장한다.

  • 검색 체인 구성

    • 검색 체인을 구성하여, 특정 질문에 대한 답변을 얻을 수 있다:

      from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

retriever = vectorstore.as_retriever()
template = """Answer the question based only on the following context:
{context}

Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
model = ChatOpenAI(model_name="gpt-4o-mini")

    • 이 코드에서 retrieval_chain은 컨텍스트와 질문을 입력받아 프롬프트를 생성하고, 모델을 통해 답변을 생성하는 전체 과정이다:

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

  • 검색 체인 실행

    • 검색 체인을 실행하여 질문에 대한 답변을 얻을 수 있다:

      retrieval_chain.invoke("테디의 직업은 무엇입니까?")
'테디의 직업은 개발자입니다.'

retrieval_chain.invoke("셜리의 직업은 무엇입니까?")
'셜리의 직업은 디자이너입니다.'

    • 이 예제에서 검색 체인은 질문에 대한 답변을 생성하며, RunnablePassthrough는 질문 데이터를 그대로 전달하는 역할을 한다.

목차

  1. Runnable 구조 검토 개요
  2. API 키 관리 설정
  3. 그래프 구성 확인
  4. 그래프 출력
  5. 프롬프트 가져오기

🏖️ Runnable 구조 검토 개요

  • Runnable 구조 검토의 필요성
    • LCEL로 runnable을 생성한 후에는 이를 검사하여 어떤 일이 일어나고 있는지 더 잘 파악할 필요가 있다.
    • 이 노트북에서는 다양한 방법을 통해 runnable 구조를 검토하고 이해하는 과정을 다룬다.

🏖️ API 키 관리 설정

  • API 키를 환경변수로 관리

    • dotenv 라이브러리를 사용하여 API 키를 환경변수로 관리한다.

    • 이는 보안상 API 키를 코드에 직접 노출하지 않고 안전하게 관리할 수 있는 방법이다:

      from dotenv import load_dotenv

# API 키 정보 로드
load_dotenv()

  • LangSmith 추적 설정

    • LangSmith 추적을 설정하여 프로젝트의 로그를 관리하고 분석할 수 있다.

    • 이를 통해 프로젝트의 실행 과정과 데이터 흐름을 추적할 수 있다:

      from langchain_teddynote import logging

# 프로젝트 이름을 입력합니다.
logging.langsmith("LCEL-Advanced")

🏖️ 그래프 구성 확인

  • FAISS 벡터 저장소 생성 및 초기화

    • 텍스트 데이터로부터 FAISS 벡터 저장소를 생성한다:

      from langchain.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

vectorstore = FAISS.from_texts(
    ["Teddy is an AI engineer who loves programming!"],
    embedding=OpenAIEmbeddings(),
)

  • Chain 구성

    • Runnable을 활용하여 검색과 질문 처리를 포함한 체인을 구성한다:

      from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

retriever = vectorstore.as_retriever()

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

Question: {question}"""

prompt = ChatPromptTemplate.from_template(template)
model = ChatOpenAI(model="gpt-4o-mini")

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

  • 그래프 구성 확인

    • chain.get_graph() 메서드를 사용하여 체인의 실행 그래프를 얻는다.
    • 이 그래프는 체인의 각 노드와 노드 간의 연결을 나타내며, 각 단계 간의 데이터 흐름을 시각화한다:
      graph = chain.get_graph()
nodes = graph.nodes
edges = graph.edges

🏖️ 그래프 출력

  • 그래프를 ASCII 형식으로 출력

    • 체인의 그래프를 ASCII 형식으로 출력하여 구조를 시각적으로 확인할 수 있다:
      graph.print_ascii()

  • 그래프 구조 예시

    • 출력된 그래프는 체인의 각 단계를 노드로 표현하고, 데이터 흐름을 연결선으로 나타낸다:
      ```
       +---------------------------------+         
       | ParallelInput |         
       +---------------------------------+         
                **               **                
             ***                   ***             
           **                         **
      +----------------------+ +-------------+
      | VectorStoreRetriever | | Passthrough |
      +----------------------+ +-------------+ 
                      **               **                
                  ***         ***                  
                     **     **                     
       +----------------------------------+        
       | ParallelOutput |        
       +----------------------------------+        
                         *                         
                         *                         
                         *                         
              +--------------------+               
              | ChatPromptTemplate |               
              +--------------------+               
                         *                         
                         *                         
                         *                         
                  +------------+                   
                  | ChatOpenAI |                   
                  +------------+                   
                         *                         
                         *                         
                         *                         
               +-----------------+                 
               | StrOutputParser |                 
               +-----------------+                 
                         *                         
                         *                         
                         *                         
            +-----------------------+              
            | StrOutputParserOutput |              
            +-----------------------+              
```

🏖️ 프롬프트 가져오기

  • 체인에서 사용된 프롬프트 가져오기

    • chain.get_prompts() 메서드를 사용하여 체인에서 사용된 프롬프트 객체의 리스트를 반환받을 수 있다:
      prompts = chain.get_prompts()

  • 프롬프트 내용

    • 반환된 프롬프트는 체인의 각 단계에서 사용된 템플릿과 입력 변수를 포함한다:
      [ChatPromptTemplate(input_variables=['context', 'question'], messages=[HumanMessagePromptTemplate(prompt=PromptTemplate(input_variables=['context', 'question'], template='Answer the question based only on the following context:\n{context}  \n\nQuestion: {question}'))])]

목차

  1. RunnableLambda 개요
  2. 사용자 정의 함수 실행 방법
  3. RunnableLambda 사용 예시
  4. RunnableConfig와 함께 사용하는 방법
  5. 결과 확인 및 출력

🏖️ RunnableLambda 개요

  • RunnableLambda 정의
    • RunnableLambda는 사용자 정의 함수를 실행할 수 있는 기능을 제공합니다.
    • 이를 통해 개발자는 자신만의 함수를 정의하고, 해당 함수를 RunnableLambda를 사용하여 실행할 수 있습니다.
    • 예를 들어, 데이터 전처리, 계산, 외부 API와의 상호작용 등의 작업을 수행하는 함수를 정의하고 실행할 수 있습니다.

🏖️ 사용자 정의 함수 실행 방법

  • 함수 실행 시 주의사항
    • 사용자 정의 함수를 RunnableLambda로 래핑하여 사용할 수 있습니다.
    • 여기서 주의할 점은 사용자 정의 함수가 받을 수 있는 인자가 1개뿐이라는 점입니다.
    • 여러 인수를 받는 함수를 구현하려면, 단일 입력을 받아들이고 이를 여러 인수로 풀어내는 래퍼 함수를 작성해야 합니다.

🏖️ RunnableLambda 사용 예시

  • 간단한 함수 정의

    • 예를 들어, 텍스트의 길이를 반환하는 함수 length_function을 정의할 수 있습니다:
      def length_function(text):
    return len(text)

  • 여러 인수를 받는 함수와 래퍼

    • 두 텍스트의 길이를 곱하는 함수 _multiple_length_function과 이를 래핑하는 함수 multiple_length_function을 정의할 수 있습니다:

      def _multiple_length_function(text1, text2):
    return len(text1) * len(text2)

def multiple_length_function(_dict):
    return _multiple_length_function(_dict["text1"], _dict["text2"])

  • 체인 구성

    • RunnableLambda를 사용하여 앞서 정의한 함수를 체인에 연결할 수 있습니다:

      from operator import itemgetter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template("what is {a} + {b}?")
model = ChatOpenAI()

chain = (
    {
        "a": itemgetter("input_1") | RunnableLambda(length_function),
        "b": {"text1": itemgetter("input_1"), "text2": itemgetter("input_2")}
        | RunnableLambda(multiple_length_function),
    }
    | prompt
    | model
    | StrOutputParser()
)

  • 체인 실행

    • 체인을 실행하여 결과를 확인합니다:
      chain.invoke({"input_1": "bar", "input_2": "gah"})
# '3 + 9 = 12' 라는 결과를 얻을 수 있습니다.

🏖️ RunnableConfig와 함께 사용하는 방법

  • RunnableConfig 사용

    • RunnableLambda는 선택적으로 RunnableConfig를 수용할 수 있으며, 이를 통해 콜백, 태그 및 기타 구성 정보를 중첩된 실행에 전달할 수 있습니다.

  • 예제 코드

    • 텍스트를 파싱하고 오류를 수정하는 함수 parse_or_fix을 RunnableLambda로 래핑하여 사용합니다:

      from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableConfig
import json

def parse_or_fix(text: str, config: RunnableConfig):
    fixing_chain = (
        ChatPromptTemplate.from_template(
            "Fix the following text:\n\ntext\n{input}\n\nError: {error}"
            " Don't narrate, just respond with the fixed data."
        )
        | ChatOpenAI()
        | StrOutputParser()
    )

    for _ in range(3):
        try:
            return json.loads(text)
        except Exception as e:
            text = fixing_chain.invoke({"input": text, "error": e}, config)

    return "Failed to parse"

  • 실행 및 결과 확인

    • RunnableLambda를 사용하여 parse_or_fix 함수를 호출하고 결과를 확인할 수 있습니다:

      from langchain.callbacks import get_openai_callback

with get_openai_callback() as cb:
    output = RunnableLambda(parse_or_fix).invoke(
        input="{foo:: bar}",
        config={"tags": ["my-tag"], "callbacks": [cb]},
    )
    print(f"\n\n수정한결과:\n{output}")

  • 결과 출력

    • 수정된 결과는 다음과 같이 출력됩니다:
      config: {'tags': ['my-tag'], 'metadata': {}, 'callbacks': , 'recursion_limit': 25, 'configurable': {}}
수정한결과:
{'foo': 'bar'}

목차

  1. RunnableBranch 개요
  2. 입력에 따른 동적 로직 라우팅
  3. RunnableLambda를 활용한 사용자 정의 함수 라우팅
  4. RunnableBranch 사용 예시
  5. 전체 체인 구성 및 실행

🏖️ 04 LLM 체인 라우팅(RunnableLambda, RunnableBranch)

🏖️ RunnableBranch 개요

  • RunnableBranch 정의
    • RunnableBranch는 입력 데이터에 따라 동적으로 로직을 라우팅할 수 있는 강력한 도구입니다.
    • 입력 데이터의 특성에 따라 다양한 처리 경로를 유연하게 정의할 수 있으며, 복잡한 의사 결정 트리를 간단하고 직관적으로 구현할 수 있습니다.
    • 이러한 특성은 코드의 가독성과 유지보수성을 높여주고, 로직의 모듈화와 재사용성을 촉진합니다.
    • 또한, 런타임에 동적으로 분기 조건을 평가하고 적절한 처리 루틴을 선택하여 시스템의 적응력과 확장성을 높이는 데 기여합니다.

🏖️ 입력에 따른 동적 로직 라우팅

  • 동적 로직 라우팅의 중요성

    • 입력 데이터의 다양성과 변동성이 큰 애플리케이션에서는 동적 로직 라우팅이 매우 유용합니다.
    • RunnableBranch는 입력에 따라 적절한 처리 경로를 선택할 수 있도록 돕습니다.

  • 라우팅을 수행하는 두 가지 방법

    1. RunnableLambda를 활용한 조건부 실행 객체 반환 (권장)
      • RunnableLambda를 사용하여 조건부로 실행 가능한 객체를 반환하는 방법입니다.
    2. RunnableBranch 사용
      • 입력 데이터에 따라 여러 가지 조건을 정의하고, 해당 조건에 맞는 처리를 실행하는 방식입니다.

🏖️ RunnableLambda를 활용한 사용자 정의 함수 라우팅

  • 질문 분류 체인 구성

    • 예를 들어, 들어오는 질문을 '수학', '과학', '기타'로 분류하는 체인을 구성할 수 있습니다:

      from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

prompt = PromptTemplate.from_template(
    """주어진 사용자 질문을 `수학`, `과학`, 또는 `기타` 중 하나로 분류하세요. 한 단어 이상으로 응답하지 마세요.
    <question>{question}</question>
    Classification:"""
)

chain = (
    prompt
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

  • 사용자 정의 함수로 라우팅하기

    • 입력 데이터를 기준으로 특정 체인을 선택하는 사용자 정의 함수를 정의합니다:
      def route(info):
    if "수학" in info["topic"].lower():
        return math_chain
    elif "과학" in info["topic"].lower():
        return science_chain
    else:
        return general_chain

  • RunnableLambda를 사용한 라우팅

    • RunnableLambda로 사용자 정의 함수를 래핑하여 라우팅을 수행합니다:

      from operator import itemgetter
from langchain_core.runnables import RunnableLambda

full_chain = (
    {"topic": chain, "question": itemgetter("question")}
    | RunnableLambda(route)
    | StrOutputParser()
)

🏖️ RunnableBranch 사용 예시

  • RunnableBranch를 활용한 동적 분기

    • RunnableBranch를 사용하여 입력 값에 따라 실행할 조건과 Runnable을 정의할 수 있습니다:

      from operator import itemgetter
from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: "수학" in x["topic"].lower(), math_chain),
    (lambda x: "과학" in x["topic"].lower(), science_chain),
    general_chain,
)

  • RunnableBranch 문법

    • RunnableBranch는 (조건, Runnable) 쌍의 리스트와 기본 Runnable로 초기화됩니다.
    • 호출 시 전달된 입력 값을 각 조건에 전달하여 해당 분기를 선택합니다.
    • True로 평가되는 첫 번째 조건에 해당하는 Runnable을 실행하며, 조건에 맞지 않는 경우 기본 Runnable을 실행합니다.

🏖️ 전체 체인 구성 및 실행

  • 전체 체인 구성

    • RunnableBranch를 사용하여 전체 체인을 구성하고, 입력된 질문에 따라 적절한 체인을 실행합니다:
      full_chain = (
    {"topic": chain, "question": itemgetter("question")} 
    | branch 
    | StrOutputParser()
)

  • 체인 실행 예시

    • 질문을 입력하여 체인을 실행하면 다음과 같은 결과를 얻을 수 있습니다:

      full_chain.invoke({"question": "미적분의 개념에 대해 말씀해 주세요."})
'깨봉선생님께서 말씀하시기를, 미적분은 수학의 한 분야로, 함수의 변화와 그에 따른 양을 연구하는 학문입니다...'

full_chain.invoke({"question": "중력 가속도는 어떻게 계산하나요?"})
'아이작 뉴턴 선생님께서 말씀하시기를, 중력 가속도는 물체가 지구와 같은 천체의 중력에 의해 받는 가속도를 의미합니다...'

full_chain.invoke({"question": "RAG(Retrieval Augmented Generation)은 무엇인가요?"})
'RAG(리트리벌 증강 생성)는 정보 검색과 생성 모델을 결합한 접근 방식으로...'

목차

  1. RunnableParallel 개요
  2. 입력 및 출력 조작
  3. itemgetter 사용 예시
  4. 병렬 처리 이해
  5. 병렬 처리 성능 비교

🏖️ 05. RunnableParallel

🏖️ RunnableParallel 개요

  • RunnableParallel 정의
    • RunnableParallel은 여러 Runnable 객체를 병렬로 실행하고, 각 Runnable의 출력을 맵(map) 형태로 반환하는 데 유용하게 사용됩니다.
    • 이를 통해 시퀀스 내에서 하나의 Runnable 출력이 다음 Runnable의 입력 형식에 맞게 조작될 수 있습니다.
    • 이 클래스는 입력과 출력을 조작하고, 병렬 처리 성능을 높이기 위해 사용됩니다.

🏖️ 입력 및 출력 조작

  • 입력 조작 방식

    • RunnableParallel은 입력 데이터를 조작하여 prompt에 대한 "context"와 "question"이라는 키를 가진 맵 형태로 전달합니다.
    • 예를 들어, 사용자 입력이 단순히 질문일 경우, retriever를 사용하여 컨텍스트를 가져오고, 사용자 입력을 "question" 키 아래에 전달해야 합니다:
      retrieval_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)
    • RunnableParallel을 구성할 때, 유형 변환이 자동으로 처리되므로 입력을 별도로 래핑할 필요가 없습니다.

  • 다양한 방식의 입력 처리

    • 다음 세 가지 방식 모두 동일하게 처리됩니다:
      1. {"context": retriever, "question": RunnablePassthrough()}
      2. RunnableParallel({"context": retriever, "question": RunnablePassthrough()})
      3. RunnableParallel(context=retriever, question=RunnablePassthrough())

🏖️ itemgetter 사용 예시

  • itemgetter 사용

    • RunnableParallel과 결합할 때 Python의 itemgetter를 사용하여 맵에서 데이터를 추출할 수 있습니다:

      from operator import itemgetter

chain = (
    {
        "context": itemgetter("question") | retriever,
        "question": itemgetter("question"),
        "language": itemgetter("language"),
    }
    | prompt
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

  • 질문에 대한 응답 예시

    • 이 체인을 사용하여 주어진 질문에 대해 다양한 언어로 응답할 수 있습니다:
      chain.invoke({"question": "What is Teddy's occupation?", "language": "Korean"})
'Teddy의 직업은 AI 엔지니어입니다.'

🏖️ 병렬 처리 이해

  • 병렬 실행

    • RunnableParallel을 사용하면 여러 Runnable을 병렬로 실행하고, 각 Runnable의 출력을 맵으로 반환하는 것이 가능합니다:

      from langchain_core.runnables import RunnableParallel

map_chain = RunnableParallel(capital=capital_chain, area=area_chain)
map_chain.invoke({"country": "대한민국"})
# {'capital': '대한민국의 수도는 서울입니다.', 'area': '대한민국의 면적은 약 100,363.4km² 입니다.'}

  • 병렬 처리에서의 변수 차이

    • 각 체인의 입력 템플릿의 변수가 다르더라도 문제없이 병렬 실행이 가능합니다:
      map_chain2.invoke({"country1": "대한민국", "country2": "미국"})
# {'capital': '대한민국의 수도는 서울입니다.', 'area': '미국의 면적은 약 9,834만 제곱 킬로미터입니다.'}

🏖️ 병렬 처리 성능 비교

  • 병렬 처리 성능

    • RunnableParallel을 사용하여 병렬 처리 성능을 비교할 수 있습니다.

    • area_chain, capital_chain, map_chain의 실행 시간을 비교한 결과, map_chain이 두 체인을 병렬로 실행함에도 불구하고 거의 동일한 실행 시간을 갖습니다:

      %%timeit
area_chain.invoke({"country": "대한민국"})
# 907 ms ± 132 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

%%timeit
capital_chain.invoke({"country": "대한민국"})
# 790 ms ± 116 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

%%timeit
map_chain.invoke({"country": "대한민국"})
# 853 ms ± 159 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

목차

  1. 동적 속성 지정 개요
  2. configurable_fields 사용 예시
  3. configurable_alternatives 사용 예시
  4. LLM 객체의 대안 설정
  5. 프롬프트의 대안 설정
  6. 프롬프트 & LLM 설정 변경
  7. 설정 저장 및 재사용

🏖️ 06. 동적 속성 지정(configurable_fields, configurable_alternatives)

🏖️ 동적 속성 지정 개요

  • 동적 속성 지정이란?
    • Chain 호출 시 다양한 옵션을 동적으로 설정할 수 있는 방법입니다.
    • configurable_fields: 실행 가능한 객체의 특정 필드를 동적으로 구성할 수 있습니다.
    • configurable_alternatives: 런타임 중에 설정할 수 있는 특정 실행 가능한 객체에 대한 대안을 나열할 수 있습니다.

🏖️ configurable_fields 사용 예시

  • ChatOpenAI 모델의 동적 구성

    • model_name 속성을 동적으로 구성하여 다양한 GPT 모델을 사용할 수 있습니다:

      from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatOpenAI(temperature=0).configurable_fields(
    model_name=ConfigurableField(
        id="gpt_version",
        name="Version of GPT",
        description="Official model name of GPTs. ex) gpt-4o, gpt-4o-mini",
    )
)

  • 동적 속성 지정

    • model.invoke() 메서드를 호출할 때 config 파라미터를 사용하여 동적으로 model_name을 설정할 수 있습니다:
      model.invoke("대한민국의 수도는 어디야?", config={"configurable": {"gpt_version": "gpt-3.5-turbo"}})

  • with_config 메서드 사용

    • with_config 메서드를 사용하여 구성된 설정을 체인에 적용할 수 있습니다:
      model.with_config(configurable={"gpt_version": "gpt-4o-mini"}).invoke("대한민국의 수도는 어디야?")

🏖️ configurable_alternatives 사용 예시

  • LLM 객체의 대안 설정

    • LLM 객체의 대안(alternatives)을 설정하여 다양한 모델을 사용할 수 있습니다:

      from langchain_anthropic import ChatAnthropic
from langchain_openai import ChatOpenAI

llm = ChatAnthropic(
    temperature=0, model="claude-3-5-sonnet-20240620"
).configurable_alternatives(
    ConfigurableField(id="llm"),
    default_key="anthropic",
    openai=ChatOpenAI(model="gpt-4o-mini"),
    gpt4o=ChatOpenAI(model="gpt-4o"),
)

  • 체인 실행 예시

    • chain.invoke() 메서드를 호출할 때 with_config 메서드를 사용하여 설정을 변경할 수 있습니다:
      chain.with_config(configurable={"llm": "openai"}).invoke({"topic": "뉴진스"})

🏖️ 프롬프트의 대안 설정

  • 프롬프트 대안 설정

    • 프롬프트도 LLM과 유사하게 대안 설정이 가능합니다:
      prompt = PromptTemplate.from_template("{country} 의 수도는 어디야?").configurable_alternatives(
    ConfigurableField(id="prompt"),
    default_key="capital",
    area=PromptTemplate.from_template("{country} 의 면적은 얼마야?"),
    population=PromptTemplate.from_template("{country} 의 인구는 얼마야?"),
    eng=PromptTemplate.from_template("{input} 을 영어로 번역해주세요."),
)

  • 설정 변경을 통한 실행

    • 설정을 변경하여 프롬프트를 호출할 수 있습니다:
      chain.with_config(configurable={"prompt": "area"}).invoke({"country": "대한민국"})

🏖️ 프롬프트 & LLM 설정 변경

  • 프롬프트와 LLM을 조합하여 설정 변경
    • 프롬프트와 LLM 모두 변경하여 다양한 구성을 실현할 수 있습니다:
      chain.with_config(configurable={"prompt": "founder", "llm": "openai"}).invoke({"company": "애플"})

🏖️ 설정 저장 및 재사용

  • 설정 저장
    • 설정된 체인을 별도의 객체로 저장하여 재사용할 수 있습니다:
      gpt4_competitor_chain = chain.with_config(configurable={"llm": "gpt4", "prompt": "competitor"})
gpt4_competitor_chain.invoke({"company": "애플"})

목차

  1. @chain 데코레이터 개요
  2. @chain 데코레이터를 사용한 Runnable 구성
  3. 프롬프트 템플릿 정의
  4. 사용자 정의 체인 생성
  5. custom_chain 실행 및 결과 예시

🏖️ 07. @chain 데코레이터로 Runnable 구성

🏖️ @chain 데코레이터 개요

  • @chain 데코레이터란?
    • @chain 데코레이터를 사용하여 임의의 함수를 체인으로 변환할 수 있습니다.
    • 이는 함수를 RunnableLambda로 래핑하는 것과 기능적으로 동일하며, 이를 통해 함수를 간편하게 Runnable한 객체로 만들 수 있습니다.

🏖️ @chain 데코레이터를 사용한 Runnable 구성

  • 사용자 정의 체인 구성

    • @chain 데코레이터를 사용하여 함수를 데코레이팅함으로써, 해당 함수를 Runnable로 변환할 수 있습니다:

      from langchain_core.runnables import chain

@chain
def custom_chain(text):
    chain1 = prompt1 | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()
    output1 = chain1.invoke({"topic": text})

    chain2 = prompt2 | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()
    return chain2.invoke({"sentence": output1})

🏖️ 프롬프트 템플릿 정의

  • 프롬프트 템플릿 정의

    • 두 개의 프롬프트 템플릿을 정의하여 사용자 정의 체인에서 사용합니다:

      from langchain_core.prompts import ChatPromptTemplate

prompt1 = ChatPromptTemplate.from_template("{topic} 에 대해 짧게 한글로 설명해주세요.")
prompt2 = ChatPromptTemplate.from_template("{sentence} 를 emoji를 활용한 인스타그램 게시글로 만들어주세요.")

🏖️ 사용자 정의 체인 생성

  • custom_chain 함수

    • custom_chain 함수는 입력 텍스트를 기반으로 사용자 정의 체인을 실행합니다.

    • 첫 번째 체인은 주어진 주제에 대한 설명을 생성하고, 두 번째 체인은 이를 이모지를 활용한 인스타그램 게시글로 변환합니다:

      @chain
def custom_chain(text):
    chain1 = prompt1 | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()
    output1 = chain1.invoke({"topic": text})

    chain2 = prompt2 | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()
    return chain2.invoke({"sentence": output1})

🏖️ custom_chain 실행 및 결과 예시

  • custom_chain 실행

    • custom_chain은 이제 Runnable 객체로 변환되었으므로, invoke() 메서드를 사용하여 실행할 수 있습니다:
      print(custom_chain.invoke("양자역학"))

  • 실행 결과 예시

    • 실행 결과는 다음과 같이 반환됩니다:

      🌌🔬✨ 양자역학: 미시 세계의 신비를 탐구하다! 

양자역학은 원자와 아원자 입자들의 놀라운 현상을 설명하는 이론이에요. 🧪💫 고전역학과는 다르게, 입자의 위치와 운동량을 동시에 정확히 알 수 없다는 점이 매력적이죠! 🤔🔍

여기 몇 가지 핵심 개념을 소개할게요: 
- 파동-입자 이중성 🌊➡️⚛️
- 불확정성 원리 ❓🔄
- 양자 얽힘 🔗💖

이 모든 것이 현대 물리학과 기술의 기초가 된답니다! 🌍💡 양자 세계의 신비를 함께 탐험해봐요! 🚀🔭 #양자역학 #물리학 #과학의미래 #신비로운세계

목차

  1. RunnableWithMessageHistory 개요
  2. 메시지 기록 관리의 중요성
  3. RunnableWithMessageHistory 활용 예시
  4. 인메모리 방식으로 메시지 기록 관리
  5. Redis를 활용한 영구 메시지 기록 관리
  6. 다양한 메시지 관리 패턴
  7. 결론

🏖️ 08. RunnableWithMessageHistory 개요

  • RunnableWithMessageHistory란?
    • RunnableWithMessageHistory는 특정 유형의 작업(체인)에 메시지 기록을 추가하는 기능을 제공합니다.
    • 이 기능은 대화형 애플리케이션이나 복잡한 데이터 처리 작업에서 특히 유용하며, 메시지 기록을 관리하여 이전 메시지의 맥락을 유지하면서 응답을 생성할 수 있습니다.

🏖️ 메시지 기록 관리의 중요성

  • 메시지 기록 관리의 필요성
    • 대화형 챗봇 개발: 사용자와의 대화 내역을 기반으로 챗봇의 응답을 조정할 수 있습니다.
    • 복잡한 데이터 처리: 데이터 처리 과정에서 이전 단계의 결과를 참조하여 다음 단계의 로직을 결정할 수 있습니다.
    • 상태 관리가 필요한 애플리케이션: 사용자의 이전 선택을 기억하고 그에 따라 다음 화면이나 정보를 제공할 수 있습니다.

🏖️ RunnableWithMessageHistory 활용 예시

  • 실제 활용 예시

    • 아래 예시는 RunnableWithMessageHistory를 사용하여 사용자 입력과 메시지 기록을 관리하는 방법을 보여줍니다:

      from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

model = ChatOpenAI()
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "당신은 {ability} 에 능숙한 어시스턴트입니다. 20자 이내로 응답하세요"),
        MessagesPlaceholder(variable_name="history"),  # 대화 기록 변수
        ("human", "{input}"),
    ]
)
runnable = prompt | model

🏖️ 인메모리 방식으로 메시지 기록 관리

  • 인메모리 방식

    • 인메모리 방식은 메시지 기록을 메모리에 저장하며, 개발 단계나 간단한 애플리케이션에 적합합니다:

      from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

store = {}  # 세션 기록을 저장할 딕셔너리

def get_session_history(session_ids: str) -> ChatMessageHistory:
    if session_ids not in store:
        store[session_ids] = ChatMessageHistory()
    return store[session_ids]

with_message_history = RunnableWithMessageHistory(
    runnable,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

🏖️ Redis를 활용한 영구 메시지 기록 관리

  • Redis를 사용한 영구 저장

    • Redis를 사용하여 메시지 기록을 영구적으로 저장하는 방법을 보여줍니다:

      from langchain_community.chat_message_histories import RedisChatMessageHistory

REDIS_URL = "redis://localhost:6379/0"

def get_message_history(session_id: str) -> RedisChatMessageHistory:
    return RedisChatMessageHistory(session_id, url=REDIS_URL)

with_message_history = RunnableWithMessageHistory(
    runnable,
    get_message_history,
    input_messages_key="input",
    history_messages_key="history",
)

🏖️ 다양한 메시지 관리 패턴

  • 다양한 메시지 관리 패턴

    • 메시지 입력과 출력을 다양한 방식으로 관리할 수 있습니다:

      • Messages 객체를 입력, dict 형태의 출력:

        from langchain_core.messages import HumanMessage
from langchain_core.runnables import RunnableParallel

chain = RunnableParallel({"output_message": ChatOpenAI()})

with_message_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    output_messages_key="output_message",
)

      • Messages 객체를 입력, Messages 객체의 출력:

        with_message_history = RunnableWithMessageHistory(
    ChatOpenAI(),
    get_session_history,
)

      • 모든 메시지 입력과 출력을 위한 단일 키를 가진 Dict:

        from operator import itemgetter

with_message_history = RunnableWithMessageHistory(
    itemgetter("input_messages") | ChatOpenAI(),
    get_session_history,
    input_messages_key="input_messages",
)

🏖️ 결론

  • 결론
    • RunnableWithMessageHistory는 메시지 기록을 관리함으로써 대화형 애플리케이션의 흐름을 더 잘 제어하고, 사용자의 이전 요청에 따라 적절하게 응답할 수 있게 해주는 강력한 도구입니다.
    • 메시지 기록을 관리하는 방법을 선택할 때는 애플리케이션의 요구사항과 데이터의 중요성을 고려해야 합니다.

목차

  1. 사용자 정의 제네레이터(generator) 개요
  2. 제너레이터 활용 방법
  3. 기본적인 제너레이터 사용 예시
  4. 제너레이터로 스트리밍 처리
  5. 사용자 정의 출력 파서(split_into_list)
  6. 비동기 제너레이터(asplit_into_list) 사용
  7. 결론

🏖️ 09. 사용자 정의 제네레이터(generator) 개요

  • 제너레이터란?
    • 제너레이터 함수는 yield 키워드를 사용하여 이터레이터처럼 동작하는 함수입니다.
    • LCEL 파이프라인에서 제너레이터를 활용하면, 입력을 처리하고 수정된 출력을 스트리밍 형태로 제공할 수 있습니다.

🏖️ 제너레이터 활용 방법

  • 제너레이터 활용 시나리오
    • 사용자 정의 출력 파서 구현: 이전 단계의 출력을 수정하면서 스트리밍 기능을 유지할 수 있습니다.
    • 실시간 처리: 입력 데이터를 실시간으로 처리하고 즉각적인 출력을 제공하는데 유용합니다.

🏖️ 기본적인 제너레이터 사용 예시

  • 예시 코드

    • 아래 코드는 주어진 회사와 유사한 5개의 회사를 쉼표로 구분된 목록으로 작성하는 제너레이터를 보여줍니다:

      from langchain.prompts.chat import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template(
    "Write a comma-separated list of 5 companies similar to: {company}"
)
model = ChatOpenAI(temperature=0.0, model="gpt-4-turbo-preview")

str_chain = prompt | model | StrOutputParser()

🏖️ 제너레이터로 스트리밍 처리

  • 스트리밍 처리
    • 스트리밍을 사용하면 실시간으로 데이터를 처리하고 결과를 출력할 수 있습니다:
      for chunk in str_chain.stream({"company": "Google"}):
    print(chunk, end="", flush=True)

🏖️ 사용자 정의 출력 파서(split_into_list)

  • 사용자 정의 출력 파서

    • split_into_list 함수는 쉼표로 구분된 문자열을 리스트로 분할합니다:
      def split_into_list(input: Iterator[str]) -> Iterator[List[str]]:
    buffer = ""
    for chunk in input:
        buffer += chunk
        while "," in buffer:
            comma_index = buffer.index(",")
            yield [buffer[:comma_index].strip()]
            buffer = buffer[comma_index + 1 :]
    yield [buffer.strip()]

  • 함수 적용

    • str_chain의 출력을 split_into_list 함수로 분할하여 처리합니다:
      list_chain = str_chain | split_into_list

🏖️ 비동기 제너레이터(asplit_into_list) 사용

  • 비동기 제너레이터

    • 비동기 제너레이터를 사용하여 입력을 처리하고 비동기 출력을 제공합니다:
      async def asplit_into_list(input: AsyncIterator[str]) -> AsyncIterator[List[str]]:
    buffer = ""
    async for chunk in input:
        buffer += chunk
        while "," in buffer:
            comma_index = buffer.index(",")
            yield [buffer[:comma_index].strip()]
            buffer = buffer[comma_index + 1:]
    yield [buffer.strip()]

  • 비동기 스트리밍

    • 비동기 스트리밍을 통해 실시간 출력을 관리합니다:
      async for chunk in alist_chain.astream({"company": "Google"}):
    print(chunk, flush=True)

🏖️ 결론

  • 결론
    • 사용자 정의 제너레이터를 사용하면 복잡한 데이터 처리와 실시간 응답을 구현하는 데 유용하며, 비동기 제너레이터를 활용하면 효율적인 비동기 작업을 수행할 수 있습니다.
    • 이러한 기술을 통해 스트리밍 데이터와 대규모 데이터를 효과적으로 처리할 수 있습니다.

🏖️ Runtime Arguments 바인딩

때로는 Runnable 시퀀스 내에서 Runnable을 호출할 때, 이전 Runnable의 출력이나 사용자 입력에 포함되지 않은 상수 인자를 전달해야 할 경우가 있습니다. 이럴 때는 Runnable.bind()를 사용하면 이러한 인자를 쉽게 전달할 수 있습니다.

기본 예제

RunnablePassthrough를 사용하여 {equation_statement} 변수를 프롬프트에 전달하고, StrOutputParser를 사용하여 모델의 출력을 문자열로 파싱하는 runnable 객체를 생성합니다. 이후 runnable.invoke() 메서드를 호출하여 "x raised to the third plus seven equals 12"라는 방정식 문장을 전달하고 결과를 출력합니다.

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "Write out the following equation using algebraic symbols then solve it. Use the format\n\nEQUATION:...\nSOLUTION:...\n\n",
        ),
        (
            "human",
            "{equation_statement}",
        ),
    ]
)

model = ChatOpenAI(model="gpt-4", temperature=0)

runnable = (
    {"equation_statement": RunnablePassthrough()} | prompt | model | StrOutputParser()
)

print(runnable.invoke("x raised to the third plus seven equals 12"))

결과:

EQUATION: x^3 + 7 = 12
SOLUTION: x^3 = 12 - 7
x^3 = 5
x = ∛5

Stop 단어를 사용하여 모델 호출

특정 stop 단어를 사용하여 모델을 호출하고자 합니다. model.bind()를 사용하여 언어 모델을 호출하고, 생성된 텍스트에서 "SOLUTION" 토큰까지만 출력합니다.

runnable = (
    {"equation_statement": RunnablePassthrough()}
    | prompt
    | model.bind(stop="SOLUTION")
    | StrOutputParser()
)

print(runnable.invoke("x raised to the third plus seven equals 12"))

결과:

EQUATION: x^3 + 7 = 12

OpenAI Functions 기능 연결

binding의 유용한 활용 방법 중 하나는 호환되는 OpenAI 모델에 OpenAI Functions를 연결하는 것입니다. 아래는 OpenAI Functions를 스키마에 맞게 정의한 코드입니다.

openai_function = {
    "name": "solver",
    "description": "Formulates and solves an equation",
    "parameters": {
        "type": "object",
        "properties": {
            "equation": {
                "type": "string",
                "description": "The algebraic expression of the equation",
            },
            "solution": {
                "type": "string",
                "description": "The solution to the equation",
            },
        },
        "required": ["equation", "solution"],
    },
}

bind() 메서드를 사용하여 solver라는 이름의 함수 호출을 모델에 바인딩합니다.

prompt = ChatPromptTemplate.from_messages(
    [
        (
            <"system",
            "Write out the following equation using algebraic symbols then solve it.",
        ),
        ("human", "{equation_statement}"),
    ]
)

model = ChatOpenAI(model="gpt-4", temperature=0).bind(
    function_call={"name": "solver"},
    functions=[openai_function],
)

runnable = {"equation_statement": RunnablePassthrough()} | prompt | model
runnable.invoke("x raised to the third plus seven equals 12")

결과:

AIMessage(content='', additional_kwargs={'function_call': {'arguments': '{\n"equation": "x^3 + 7 = 12",\n"solution": "x = ∛5"\n}', 'name': 'solver'}})

OpenAI tools 연결하기

OpenAI에서 제공하는 tools를 연결하여 활용하는 방법에 대해 설명하겠습니다.

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "주어진 위치의 현재 날씨를 가져옵니다",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "도시와 주, 예: San Francisco, CA",
                    },
                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
                },
                "required": ["location"],
            },
        },
    }
]

model = ChatOpenAI(model="gpt-3.5-turbo").bind(tools=tools)
model.invoke("샌프란시스코, 뉴욕, 로스앤젤레스의 현재 날씨에 대해 알려줘?")

결과:

AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_iHdvMMJ8xA0hGNRi3Xu5lJLR', 'function': {'arguments': '{"location": "San Francisco, CA"}', 'name': 'get_current_weather'}, 'type': 'function'}, {'id': 'call_ShDVUUvz9XcVOcMwvFl3eazS', 'function': {'arguments': '{"location": "New York, NY"}', 'name': 'get_current_weather'}, 'type': 'function'}, {'id': 'call_gdW3CfrYWa3Ap8kIPE7FFanF', 'function': {'arguments': '{"location": "Los Angeles, CA"}', 'name': 'get_current_weather'}, 'type': 'function'}]})

🏖️ 폴백(fallback) 모델 지정

LLM 애플리케이션에서는 LLM API 문제, 모델 출력 품질 저하, 다른 통합 관련 이슈 등 다양한 오류와 실패 상황이 발생할 수 있습니다. 이러한 문제를 우아하게 처리하고 격리하는 데 fallback 기능을 활용할 수 있습니다.

LLM API Error 대처 방법

LLM API 오류 처리는 fallback을 사용하는 가장 일반적인 사례 중 하나입니다. LLM API에 대한 요청은 다양한 이유로 실패할 수 있습니다. API가 다운되었거나, 속도 제한에 도달했거나, 그 외 여러 가지 문제가 발생할 수 있습니다. 따라서 fallback을 사용하면 이러한 유형의 문제로부터 보호하는 데 도움이 됩니다.

중요: 기본적으로 많은 LLM 래퍼(wrapper)는 오류를 포착하고 재시도합니다. fallback을 사용할 때는 이러한 기본 동작을 해제하는 것이 좋습니다. 그렇지 않으면 첫 번째 래퍼가 계속 재시도하고 실패하지 않을 것입니다.

%pip install -qU langchain langchain-openai

from langchain_anthropic import ChatAnthropic
from langchain_openai import ChatOpenAI
from unittest.mock import patch
import httpx
from openai import RateLimitError

request = httpx.Request("GET", "/")
response = httpx.Response(200, request=request)
error = RateLimitError("rate limit", response=response, body="")

openai_llm = ChatOpenAI(max_retries=0)
anthropic_llm = ChatAnthropic(model="claude-3-opus-20240229")
llm = openai_llm.with_fallbacks([anthropic_llm])

with patch("openai.resources.chat.completions.Completions.create", side_effect=error):
    try:
        print(openai_llm.invoke("Why did the chicken cross the road?"))
    except RateLimitError:
        print("에러 발생")

결과:

에러 발생

OpenAI API의 비용 제한(rate limit)을 시뮬레이션하고, API 호출비용 제한 오류가 발생했을 때의 동작을 테스트하는 예제입니다. OpenAI의 GPT 모델을 시도하는데 에러가 발생했고, fallback 모델인 Anthropic의 모델이 대신 추론을 수행했다는 점을 확인할 수 있습니다.

with patch("openai.resources.chat.completions.Completions.create", side_effect=error):
    try:
        print(llm.invoke("대한민국의 수도는 어디야?"))
    except RateLimitError:
        print("에러 발생")

결과:

content='대한민국의 수도는 서울특별시입니다.'

llm.with_fallbacks() 설정한 모델도 일반 runnable 모델과 동일하게 동작합니다. 아래의 코드 역시 "오류 발생"은 출력되지 않습니다. fallback 모델이 잘 수행했기 때문입니다.

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "질문에 짧고 간결하게 답변해 주세요."),
        ("human", "{country} 의 수도는 어디입니까?"),
    ]
)
chain = prompt | llm

with patch("openai.resources.chat.completions.Completions.create", side_effect=error):
    try:
        print(chain.invoke({"country": "대한민국"}))
    except RateLimitError:
        print("오류 발생")

결과:

content='대한민국의 수도는 서울특별시입니다.'

특정 오류를 구체적으로 명시한 경우

특정 오류를 처리하기 위해 fallback이 호출되는 시점을 더 명확하게 지정할 수도 있습니다. 예를 들어, 특정 예외 클래스나 오류 코드를 지정함으로써 해당 오류가 발생했을 때만 fallback 로직이 실행되도록 설정할 수 있습니다.

llm = openai_llm.with_fallbacks(
    [anthropic_llm],
    exceptions_to_handle=(KeyboardInterrupt,),
)

chain = prompt | llm

with patch("openai.resources.chat.completions.Completions.create", side_effect=error):
    try:
        print(chain.invoke({"country": "대한민국"}))
    except RateLimitError:
        print("오류 발생")

결과:

content='대한민국의 수도는 서울특별시입니다.'

여러 개의 폴백 모델을 순차적으로 지정

fallback 모델에 여러 개의 모델을 지정하면 순차적으로 시도하게 됩니다.

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

prompt_template = "질문에 짧고 간결하게 답변해 주세요.\n\nQuestion:\n{question}\n\nAnswer:"
prompt = PromptTemplate.from_template(prompt_template)

chat_model = ChatOpenAI(model_name="gpt-fake")
bad_chain = prompt | chat_model
fallback_chain1 = prompt | ChatOpenAI(model="gpt-3.6-turbo")
fallback_chain2 = prompt | ChatOpenAI(model="gpt-3.5-turbo")
fallback_chain3 = prompt | ChatOpenAI(model="gpt-4-turbo-preview")

chain = bad_chain.with_fallbacks([fallback_chain1, fallback_chain2, fallback_chain3])

chain.invoke({"question": "대한민국의 수도는 어디야?"})

결과:

AIMessage(content='서울입니다.')
profile
이영락
AI Engineer / 의료인공지능
이전 포스트

RAG & LANCHAIN (6)- Langchain 기반 RAG 기본 구조 update 버전

다음 포스트

seleinum 내용 정리

0개의 댓글

관련 채용 정보