1.5. RunnableLambda

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 애플리케이션에서 핵심적인 역할을 합니다.