더 나은 세상을 위하여…

프롬프트 단계는 검색기에서 검색된 문서들을 바탕으로 언어 모델이 사용할 질문이나 명령을 생성하는 과정입니다. 이 단계는 검색된 정보를 바탕으로 최종 사용자의 질문에 가장 잘 대응할 수 있는 응답을 생성하기 위해 필수적인 단계입니다.

프롬프트의 필요성

1. 문맥(Context) 설정: 프롬프트는 언어 모델이 특정 문맥에서 작동하도록 설정하는 역할을 합니다. 이를 통해 모델은 제공된 정보를 바탕으로 보다 정확하고 관련성 높은 답변을 생성할 수 있습니다.
2. 정보 통합: 여러 문서에서 검색된 정보는 서로 다른 관점이나 내용을 포함할 수 있습니다. 프롬프트 단계에서 이러한 정보를 통합하고, 모델이 이를 효율적으로 활용할 수 있는 형식으로 조정합니다.
3. 응답 품질 향상: 질문에 대한 모델의 응답 품질은 프롬프트의 구성에 크게 의존합니다. 잘 구성된 프롬프트는 모델이 보다 정확하고 유용한 정보를 제공하게 돕습니다.

RAG 프롬프트 구조

• 지시사항(Instruction)
• 질문(사용자 입력 질문)
• 문맥(검색된 정보)

프롬프트의 중요성

프롬프트 단계는 RAG 시스템에서 중요한 역할을 합니다. 이 단계를 통해 언어 모델은 사용자의 질문에 대해 최적화된 방식으로 응답을 생성할 수 있으며, 시스템 전체의 성능과 사용자 만족도에 직접적인 영향을 미칩니다. 프롬프트가 잘 구성되어 있지 않으면, 모델이 비효율적으로 작동할 수 있으며, 결과적으로 사용자의 요구에 부응하지 못하는 응답을 생성할 가능성이 높아집니다.

제로샷, 원샷, 퓨샷의 개념

LLM으로부터 더 좋은 응답을 하기 위해 사용자가 예시를 주는 것을 원샷, 퓨샷 프롬프트라고 합니다. 원샷 (One shot)은 예시를 하나만 주는 것이고 퓨샷 (Few shot)은 2개 이상 예시를 제공해 주는 것이죠. ‘샷’은 프롬프트 엔지니어링에서 ‘예시’라는 뜻으로 사용됩니다. 제로샷은 예시없이 응답을 구하는 것입니다. 예시가 없으면 자신이 원하는 대답이 나오지 않을 확률이 올라가긴 하지만 AI의 자율성과 창의성이 가미된 대답이 나오곤 합니다.

RunnableLambda는 LangChain에서 사용자 정의 함수를 실행하기 위해 제공되는 기능으로, Python의 callable 객체를 Runnable로 변환하여 체인에 통합할 수 있게 합니다. 이를 통해 복잡한 로직을 처리하거나 기존 기능을 확장할 수 있으며, 데이터 변환, 계산, 외부 API 호출 등 다양한 작업을 수행할 수 있습니다.

주요 특징

• 사용자 정의 함수 래핑:
  • RunnableLambda는 임의의 Python 함수 또는 람다를 Runnable 객체로 변환합니다. 이를 통해 체인 내에서 사용자 정의 처리를 쉽게 구현할 수 있습니다.
  • 예를 들어, 문자열 길이를 계산하거나 두 텍스트의 길이를 곱하는 작업을 수행할 수 있습니다.
• 단일 입력 처리:
  • 사용자 정의 함수는 반드시 단일 입력만 받을 수 있습니다. 여러 인자를 처리하려면 단일 입력을 받아 여러 인자로 풀어내는 래퍼(wrapper)를 작성해야 합니다.
• 체인 연결:
  • RunnableLambda는 다른 Runnable과 쉽게 연결될 수 있으며, | 연산자를 사용하여 체인을 구성할 수 있습니다. 이때 하나가 함수라면 자동으로 RunnableLambda로 변환됩니다.
• RunnableConfig 지원:
  • RunnableLambda는 선택적으로 RunnableConfig를 지원하며, 이를 통해 콜백, 태그 및 기타 구성 정보를 설정하여 중첩된 실행에 전달할 수 있습니다.

사용 예제

기본 사용

이 코드는 숫자 5를 문자열로 변환하는 간단한 예제입니다.

from langchain_core.runnables import RunnableLambda

# 람다 함수를 Runnable 객체로 변환
runnable = RunnableLambda(lambda x: str(x))
response = runnable.invoke(5)
print(response)

5

1) 체인 구성

체인을 구성하여 입력 데이터를 처리하고 결과를 반환합니다.

from langchain_core.runnables import RunnableLambda

# 두 함수 체인 연결
runnable1 = RunnableLambda(lambda x: {"foo": x})
runnable2 = RunnableLambda(lambda x: [x] * 2)
chain = runnable1 | runnable2

result = chain.invoke(2)
print(result)

[{‘foo’: 2}, {‘foo’: 2}]

2) 복잡한 로직 처리

여러 입력값을 처리하기 위해 래퍼 함수를 사용하여 데이터를 가공합니다.

def length_function(text):
   return len(text)

def multiple_length_function(_dict):
   return len(_dict["text1"]) * len(_dict["text2"])

runnable1 = RunnableLambda(length_function)
runnable2 = RunnableLambda(multiple_length_function)

result = runnable2.invoke({"text1": "hello", "text2": "world"})
print(result)

25

3) RunnableConfig 활용

RunnableConfig를 사용하여 태그와 콜백을 설정하고 오류를 처리하는 예제입니다

import json
from langchain_core.runnables import RunnableLambda, RunnableConfig

def parse_or_fix(text: str, config: RunnableConfig):
   try:
       return json.loads(text)
   except Exception as e:
       return f"Error: {e}"

config = {"tags": ["example-tag"]}
output = RunnableLambda(parse_or_fix).invoke('{"foo": "bar"}', config=config)
print(output)

{‘foo’: ‘bar’}

활용 사례

• 데이터 전처리: 입력 데이터를 정규화하거나 필터링하는 작업.
• 복잡한 계산: 여러 입력값을 기반으로 수학적 연산 수행.
• API 호출 및 응답 처리: 외부 API와 상호작용 후 데이터를 가공.
• 라우팅 및 분기 처리: 조건에 따라 실행할 함수를 선택하는 로직 구현.

장점

• 유연성: Python 함수와 LangChain의 체인을 쉽게 통합 가능.
• 확장성: 복잡한 워크플로우를 간단하게 캡슐화.
• 가독성: 체인의 각 단계가 명확하게 정의되어 유지보수가 용이.

RunnableLambda는 LangChain에서 사용자 정의 처리를 구현하는 데 매우 유용하며, 다양한 LLM 애플리케이션에서 핵심적인 역할을 합니다.

RunnableParallel은 LangChain에서 제공하는 주요 컴포지션 프리미티브 중 하나로, 여러 개의 Runnable을 병렬적으로 실행할 수 있도록 설계된 클래스입니다. 이 클래스는 동일한 입력 데이터를 여러 Runnable에 동시에 전달하여 독립적으로 처리한 후, 각 결과를 매핑하여 반환합니다.

특징 및 주요 기능

• 병렬 실행: 여러 Runnable을 동시에 실행하여 처리 속도를 높일 수 있습니다.
• 입력 공유: 동일한 입력 데이터를 모든 Runnable에 전달합니다.
• 결과 매핑: 각 Runnable의 출력값을 키-값 쌍으로 반환합니다.
• 유연한 설정: 딕셔너리 형태로 Runnable을 정의하거나, 파이프 연산자(|)를 사용해 손쉽게 구성할 수 있습니다.

사용 방법

1. 기본 사용 예제

아래는 숫자를 처리하는 세 가지 함수(add_one, mul_two, mul_three)를 병렬로 실행하는 예제입니다:

from langchain_core.runnables import RunnableLambda, RunnableParallel

# 함수 정의
def add_one(x: int) -> int:
   return x + 1

def mul_two(x: int) -> int:
   return x * 2

def mul_three(x: int) -> int:
   return x * 3

# Runnable 생성
runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
runnable_3 = RunnableLambda(mul_three)

# RunnableParallel로 병렬 실행 설정
parallel_runnable = RunnableParallel(
   mul_two=runnable_2,
   mul_three=runnable_3
)

# 실행
result = parallel_runnable.invoke(1)
print(result)

{‘mul_two’: 2, ‘mul_three’: 3}

2. LLM과 함께 사용

아래는 LangChain의 LLM을 활용해 병렬로 joke, poem, question을 생성하는 예제입니다:

from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import ChatOllama
from langchain_core.runnables import RunnableParallel
from langchain_core.output_parsers import StrOutputParser

# Prompt 정의
joke_template = ChatPromptTemplate.from_template("tell me a joke about {topic}")
poem_template = ChatPromptTemplate.from_template("write a 2-line poem about {topic}")
question_template = ChatPromptTemplate.from_template("make a question about {topic}")

# 모델 및 파서 정의
model = ChatOllama(model='exaone3.5')
parser = StrOutputParser()

# Chain 정의
joke_chain = joke_template | model | parser
poem_chain = poem_template | model | parser
question_chain = question_template | model | parser

# RunnableParallel 설정
parallel_chains = RunnableParallel(
   joke=joke_chain,
   poem=poem_chain,
   question=question_chain
)

# 실행
output = parallel_chains.invoke({"topic": "snowman"})
print(output)

{‘joke’: ‘Sure, here\’s a snowy joke for you:\n\nWhy don\’t snowmen ever get lost?\n\nBecause they always have a carrot-towards home! (get it? Carrot as in the vegetable, play on words with “course to” towards)\n\nHope that brings a smile to your face! ❄️😄’, ‘poem’: “Carved in winter’s chill, smiles bright beneath falling snow,  \nCozy coal eyes watch as icy drifts softly grow.”, ‘question’: “Here’s a fun question about snowmen:\n\n**What unique feature can you add to a snowman to make it look extra festive during the holiday season?**\n\n(Possible answers could include carrot noses, coal eyes and mouths, scarf made of yarn, or even a small hat adorned with bells!)”}

실행 방식 비교

병렬 처리를 통해 개별적으로 실행했을 때보다 더 빠르게 작업을 완료할 수 있습니다. 아래는 동일한 작업을 병렬 처리와 개별 처리로 실행했을 때의 시간 비교 예제입니다:

• 개별 처리 시간:
  • Joke Chain: 약 11초
  • Poem Chain: 약 0.9초
  • Question Chain: 약 0.75초
• 병렬 처리 시간: 약 0.92초

응용 사례

1. 다중 작업 병렬화: 동일한 입력 데이터로 여러 작업(예: 텍스트 생성, 분석 등)을 동시에 수행.
2. 독립적인 체인 실행: 서로 다른 입력 데이터를 사용하는 체인을 병렬로 실행하여 결과를 통합.

RunnableParallel은 LangChain에서 작업 효율성을 극대화하기 위한 강력한 도구입니다. 특히, 다중 작업을 동시에 처리해야 하는 경우 유용하며, 간단한 인터페이스와 유연한 설정으로 다양한 시나리오에 적용 가능합니다