[langchain] LangChain Expression Language(LCEL)
1. LangChain Expression Language(LCEL) 소개
LangChain 표현 언어, 혹은 LCEL은 LangChain 컴포넌트들을 연결하는 선언적 방식입니다. LCEL은 처음부터 프로토타입을 코드 수정 없이 바로 프로덕션에 배포할 수 있도록 설계되었습니다. 간단한 “프롬프트 + LLM” 체인부터 100단계 이상의 복잡한 체인까지, 많은 사용자들이 LCEL 체인을 성공적으로 프로덕션에서 운영하고 있습니다.
2. LCEL 사용 이유
2.1. streaming support
LCEL로 체인을 구축하면 가장 빠른 첫 토큰 출력 시간을 확보할 수 있습니다 (첫 번째 출력 조각이 나올 때까지 걸리는 시간). 예를 들어, 일부 체인의 경우 LLM에서 나오는 토큰을 바로 스트리밍 출력 파서로 전송하여, LLM 제공자가 토큰을 출력하는 속도에 맞춰 분석된 점진적인 출력 조각을 받을 수 있습니다.
2.2. 비동기 지원
LCEL로 구축된 체인은 동기 API (예: 프로토타입을 만들 때 주피터 노트북에서 사용)뿐만 아니라 비동기 API (예: LangServe 서버에서 사용)로도 호출할 수 있습니다. 이를 통해 프로토타입과 프로덕션에서 동일한 코드를 사용할 수 있으며, 뛰어난 성능과 동일 서버에서 많은 동시 요청을 처리할 수 있습니다.
2.3. 최적화된 병렬 실행
LCEL 체인에서 병렬로 실행할 수 있는 단계가 있는 경우 (예: 여러 검색기에서 문서를 가져오는 경우) 동기 및 비동기 인터페이스 모두에서 이를 자동으로 처리하여 가능한 최소 지연 시간을 제공합니다.
2.4. 재시도 및 대체 옵션
LCEL 체인의 어느 부분이든 재시도 및 대체 옵션을 설정할 수 있습니다. 이는 대규모에서 체인의 신뢰성을 높이는 좋은 방법입니다. 현재 우리는 재시도/대체에 대한 스트리밍 지원을 추가 중이며, 이를 통해 지연 시간 없이 신뢰성을 높일 수 있습니다.
2.5. 중간 결과에 접근
더 복잡한 체인의 경우, 최종 출력이 생성되기 전에 중간 단계의 결과에 접근하는 것이 매우 유용할 수 있습니다. 이는 사용자에게 작업 진행 상황을 알리거나 체인을 디버깅하는 데 사용할 수 있습니다. 중간 결과를 스트리밍할 수 있으며, LangServe 서버에서 항상 사용 가능합니다.
2.6. 입력 및 출력 스키마
입력 및 출력 스키마는 모든 LCEL 체인에 대해 Pydantic 및 JSONSchema 스키마를 제공하며, 이는 체인의 구조에서 추론됩니다. 이를 통해 입력 및 출력의 유효성을 검증할 수 있으며, LangServe의 핵심 기능입니다.
2.7. LangSmith 추적 통합
체인이 점점 더 복잡해질수록 각 단계에서 무슨 일이 일어나고 있는지 이해하는 것이 중요해집니다. LCEL에서는 모든 단계가 LangSmith에 자동으로 로그가 기록되어 최대한의 가시성과 디버깅 가능성을 제공합니다.
3. LCEL 사용 방법
3.0. Runnable
Runnable은 LangChain에서 작업을 실행할 수 있는 객체를 말합니다. 즉, 특정 기능을 수행하는 함수를 감싸서 동기적 또는 비동기적으로 호출할 수 있게 만들어주는 구조입니다. Runnable은 다양한 작업을 모듈화하고 재사용 가능하게 만들며, 이를 통해 작업 흐름을 체계적으로 구성할 수 있습니다.
- 다양한 작업을 실행할 수 있도록 해주는 객체입니다.
- 동기적(invoke) 또는 비동기적(ainvoke)으로 호출 가능합니다.
- 함수나 메서드를 감싸서 더 복잡한 체인 또는 워크플로우로 확장할 수 있습니다.
- 병렬 실행, 재시도, 대체 경로(fallbacks) 등의 기능을 손쉽게 추가할 수 있습니다.
3.1. Runnable.invoke() 및 Runnable.ainvoke()
- invoke()는 동기 방식으로 런너블(Runnable)을 호출합니다.
- ainvoke()는 비동기 방식으로 런너블을 호출하는 방법입니다.
from langchain_core.runnables import RunnableLambda
runnable = RunnableLambda(lambda x: str(x))
runnable.invoke(5)
# Async variant:
# await runnable.ainvoke(5)
출력
'5'3.2. Runnable.batch() 및 Runnable.abatch()
- 여러 입력값에 대해 런너블을 한꺼번에 실행할 수 있습니다.
- batch()는 동기 방식, abatch()는 비동기 방식으로 실행됩니다.
from langchain_core.runnables import RunnableLambda
runnable = RunnableLambda(lambda x: str(x))
runnable.batch([7, 8, 9])
# Async variant:
# await runnable.abatch([7, 8, 9])
출력
['7', '8', '9']
3.3. Runnable.stream() 및 Runnable.astream()
- 여러 입력값을 순차적으로 처리하여 스트리밍 방식으로 출력합니다.
- 비동기 방식으로 실행하려면 astream()을 사용합니다.
from langchain_core.runnables import RunnableLambda
def func(x):
for y in x:
yield str(y)
runnable = RunnableLambda(func)
for chunk in runnable.stream(range(5)):
print(chunk)
# Async variant:
# async for chunk in await runnable.astream(range(5)):
# print(chunk)
출력
0
1
2
3
4
3.4. 체인 연결 (Pipe operator |)
- 여러 런너블을 체인으로 연결하여 순차적으로 실행할 수 있습니다.
- 예를 들어, runnable1은 입력값을 딕셔너리로 만들고, runnable2는 해당 딕셔너리를 복제하여 두 개로 늘립니다. 두 런너블을 | 연산자로 연결한 후 chain.invoke(2)를 호출하면 [{'foo': 2}, {'foo': 2}]가 출력됩니다.
from langchain_core.runnables import RunnableLambda
runnable1 = RunnableLambda(lambda x: {"foo": x})
runnable2 = RunnableLambda(lambda x: [x] * 2)
chain = runnable1 | runnable2
chain.invoke(2)
출력
[{'foo': 2}, {'foo': 2}]
3.5. 병렬 실행 (RunnableParallel)
- 두 개 이상의 런너블을 병렬로 실행할 수 있습니다.
- RunnableParallel 클래스를 이용합니다.
from langchain_core.runnables import RunnableLambda, RunnableParallel
runnable1 = RunnableLambda(lambda x: {"foo": x})
runnable2 = RunnableLambda(lambda x: [x] * 2)
chain = RunnableParallel(first=runnable1, second=runnable2)
chain.invoke(2)
3.6. 함수 래핑 (RunnableLambda)
일반 함수를 RunnableLambda를 사용하여 런너블로 변환할 수 있습니다.
from langchain_core.runnables import RunnableLambda
def func(x):
return x + 5
runnable = RunnableLambda(func)
runnable.invoke(2)
3.7. 입력 및 출력 딕셔너리 병합 (RunnablePassthrough.assign)
- 입력 딕셔너리는 그대로 유지되며, 새로운 값이 출력 딕셔너리에 추가됩니다.
- 여러 개의 결과를 하나의 딕셔너리에 병합해 관리할 수 있습니다.
- 입력값을 수정하지 않고 새로운 값을 추가하여 다양한 결과를 관리할 때 유용합니다.
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
# 첫 번째 런너블: 입력값에 7을 더하는 함수
runnable1 = RunnableLambda(lambda x: x["foo"] + 7)
# RunnablePassthrough를 사용해 "bar"라는 키로 runnable1의 결과를 할당
chain = RunnablePassthrough.assign(bar=runnable1)
# 입력값으로 {"foo": 10}을 전달하여 체인을 실행
result = chain.invoke({"foo": 10})
print(result)
출력
{'foo': 10, 'bar': 17}3.8. 기본 인자 추가 (Runnable.bind)
- bind()를 사용해 런너블에 기본 인자를 설정할 수 있습니다.
- 나중에 invoke()나 ainvoke()를 호출할 때, 기본값으로 설정된 인자는 다시 지정할 필요가 없습니다.
from typing import Optional
from langchain_core.runnables import RunnableLambda
def func(main_arg: dict, other_arg: Optional[str] = None) -> dict:
if other_arg:
return {**main_arg, **{"foo": other_arg}}
return main_arg
runnable1 = RunnableLambda(func)
bound_runnable1 = runnable1.bind(other_arg="bye")
bound_runnable1.invoke({"bar": "hello"})
출력
{'bar': 'hello', 'foo': 'bye'}3.9. 작업 실패시 Fallbacks
- 만약 주요 작업이 실패했을 때(예: 네트워크 오류, 서버 장애, 함수 오류 등), 지정된 대체 작업이 수행됩니다.
from langchain_core.runnables import RunnableLambda
runnable1 = RunnableLambda(lambda x: x + "foo")
runnable2 = RunnableLambda(lambda x: str(x) + "foo")
chain = runnable1.with_fallbacks([runnable2])
chain.invoke(5)
출력
'5foo'
3.10. 작업 재시도
Runnable.with_retry() 메서드는 LangChain의 **런너블(Runnable)**에서 작업이 실패할 경우 재시도하는 기능을 추가하는 메서드입니다. 이 메서드는 실행 중 예기치 않은 오류가 발생했을 때, 일정한 횟수까지 작업을 자동으로 다시 시도하게 만들어 안정적인 실행을 보장합니다.
이 기능은 비동기 작업, 네트워크 요청, 또는 불안정한 작업에서 특히 유용합니다. 네트워크 오류나 일시적인 서버 문제로 인해 작업이 실패했을 때 재시도를 통해 성공할 가능성을 높일 수 있습니다.
최종 실패 시 예외를 발생시키며, 설정된 재시도 횟수 내에서 성공하지 못하면 오류를 반환합니다.
from langchain_core.runnables import RunnableLambda
# 카운터 변수를 사용하여 몇 번째 시도인지를 추적
counter = -1
def func(x):
global counter
counter += 1
print(f"attempt with {counter=}")
# 첫 번째 시도는 0으로 나누기 때문에 실패, 두 번째 시도부터 성공
return x / counter
# 함수 func을 RunnableLambda로 래핑하고, 재시도 기능 추가 (최대 2회 재시도)
chain = RunnableLambda(func).with_retry(stop_after_attempt=2)
# 체인 실행
result = chain.invoke(2)
print(result)
출력
attempt with counter=0
ZeroDivisionError: division by zero
attempt with counter=1
2.0
3.11. 실행 설정 구성 (RunnableConfig)
- invoke에서 config 파라미터를 설정 할 수 있어요.
- 딕셔너리 형태로 config 내용을 설정합니다.
- 설정 내용
- max_concurrency: 동시에 실행하는 작업에 수
- timeout: 최대 실행 시간
- 그외 사용자 정의 설정
from langchain_core.runnables import RunnableLambda, RunnableParallel
runnable1 = RunnableLambda(lambda x: {"foo": x})
runnable2 = RunnableLambda(lambda x: [x] * 2)
runnable3 = RunnableLambda(lambda x: str(x))
chain = RunnableParallel(first=runnable1, second=runnable2, third=runnable3)
chain.invoke(7, config={"max_concurrency": 2})
- with_config() 메서드를 사용해 설정을 구성 가능합니다.
- 'config'라는 파라미터에 값을 넣어줍니다.
from langchain_core.runnables import RunnableLambda, RunnableParallel
runnable1 = RunnableLambda(lambda x: {"foo": x})
runnable2 = RunnableLambda(lambda x: [x] * 2)
runnable3 = RunnableLambda(lambda x: str(x))
chain = RunnableParallel(first=runnable1, second=runnable2, third=runnable3)
configured_chain = chain.with_config(max_concurrency=2)
chain.invoke(7)
3.12. 실행 설정 동적 구성
configurable_fields()는 LangChain에서 런너블(Runnable) 객체의 특정 필드들을 동적으로 설정할 수 있게 해주는 함수입니다. 이를 통해 필드 값을 실행 시점에 동적으로 변경할 수 있어, 더 유연한 작업을 구성할 수 있습니다. 이 메서드는 런너블의 속성 중 일부를 사용자 정의 설정 값으로 바꾸고, 나중에 호출 시 설정된 값을 적용할 수 있게 만듭니다.
from langchain_core.runnables import ConfigurableField, RunnableLambda
# 단순한 lambda 함수로, 설정된 key로 값을 반환하는 런너블 정의
def my_func(input: dict, config: dict):
output_key = config["configurable"].get("output_key", "default_key") # 기본값 "default_key"
return {output_key: input["foo"]}
# 람다 함수를 RunnableLambda로 변환
runnable = RunnableLambda(my_func)
# "output_key"를 동적으로 설정할 수 있도록 함
configurable_runnable = runnable.configurable_fields(
output_key=ConfigurableField(id="output_key")
)
# 런너블 실행 시 output_key를 "custom_key"로 설정
result = configurable_runnable.invoke(
{"foo": 42}, config={"configurable": {"output_key": "custom_key"}}
)
print(result) # 출력: {'custom_key': 42}
# output_key를 설정하지 않으면 기본값 "default_key" 사용
result = configurable_runnable.invoke({"foo": 42})
print(result) # 출력: {'default_key': 42}
ConfigurableField 객체는 1개의 설정값을 담을 수 있습니다.
- id (str): 필드의 고유 식별자.
- name (str | None): 필드의 이름 (기본값: None)
- description (str | None): 필드에 대한 설명 (기본값: None).
- annotation (Any | None): 필드에 대한 주석 (기본값: None).
- s_shared (bool): 필드가 공유되는지 여부를 나타내며, 기본값은 False입니다.
runnable.configurable_fields 메서드를 이용해 나중에 config값을 설정할 수 있도록 필드를 세팅하는 것입니다.
위 예제 코드에서는 output_key라는 필드를 동적으로 설정할 수 있습니다. 나중에 config에서 output_key 값을 설정할 수 있게 만듭니다.
3.13. 대체 옵션
메서드를 사용해 특정 실행 가능 객체(Runnable)에 대해 다양한 구성 요소를 설정할 수 있으며, 이 구성을 통해 동작을 변경할 수 있습니다.
configurable_alternatives 함수는 LangChain의 Runnable 객체가 여러 대체 옵션을 가질 수 있도록 설정하는 기능을 제공합니다. 이 함수는 실행 시점에 사용할 대체 모델이나 메서드를 동적으로 선택할 수 있도록 도와줍니다.
예를 들어, 두 가지 다른 LLM(Large Language Model)을 설정하고, 사용자가 실행 시점에 어떤 모델을 사용할지 결정할 수 있게 하는 방식으로 활용됩니다. 기본값이 설정되며, 이를 통해 특정 키를 통해 대체 가능한 모델을 고를 수 있습니다.
from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI
# ChatAnthropic 모델을 생성하고 설정 가능한 대체 모델 옵션을 추가합니다.
model = ChatAnthropic(
model_name="claude-3-sonnet-20240229" # 기본으로 사용할 ChatAnthropic 모델을 설정합니다.
).configurable_alternatives(
ConfigurableField(id="llm"), # 대체 모델을 선택할 수 있는 필드 "llm"을 설정합니다.
default_key="anthropic", # 기본적으로 "anthropic" 모델(ChatAnthropic)을 사용합니다.
openai=ChatOpenAI() # 대체 옵션으로 OpenAI 모델(ChatOpenAI)을 추가합니다.
)
# 1. 사용자가 모델을 따로 선택하지 않은 경우, 기본적으로 설정된 ChatAnthropic 모델이 실행됩니다.
# "which organization created you?"라는 질문에 대해 ChatAnthropic 모델이 응답합니다.
print(model.invoke("which organization created you?").content)
# 2. 모델의 설정을 변경하여, OpenAI 모델(ChatOpenAI)을 사용하도록 설정합니다.
# 이때는 'llm' 필드의 값을 'openai'로 설정해주면 됩니다.
# 다시 같은 질문을 하면 이번에는 OpenAI 모델이 응답합니다.
print(
model.with_config(
configurable={"llm": "openai"} # 'llm' 필드를 'openai'로 설정하여 OpenAI 모델을 사용합니다.
).invoke("which organization created you?").content
)
configureable_alternatives 메서드에서 config의 키와 값을 정해준다.
- 'llm' 키값에 'openai' 값을 넣어주면 'ChatOpenAI'모델을 사용함.
configurable={"llm": "openai"} 형태로 openai로 변형해줄 수 있다.
3.14. 체인 동적구성
LCEL(LangChain Expression Language)로, LangChain의 체인을 동적으로 구성하는 방법에 대해 배우는 것입니다. 주어진 코드에서 RunnableLambda와 RunnableParallel을 사용하여 입력값에 따라 동적으로 다른 실행 흐름을 설정하는 방법을 보여줍니다.
from langchain_core.runnables import RunnableLambda, RunnableParallel
runnable1 = RunnableLambda(lambda x: {"foo": x})
runnable2 = RunnableLambda(lambda x: [x] * 2)
chain = RunnableLambda(lambda x: runnable1 if x > 6 else runnable2)
chain.invoke(7)
출력
{'foo': 7}
6 미만일 때
chain.invoke(5)
출력
[5, 5]
3.15. 이벤트 스트림
이벤트 스트림은 데이터나 이벤트가 발생하는 순서대로 비동기적으로 처리되는 데이터 흐름입니다.
# nest_asyncio를 이용해 Jupyter notebook 환경에서 비동기 코드가 문제없이 실행되도록 설정
import nest_asyncio
nest_asyncio.apply()
from langchain_core.runnables import RunnableLambda, RunnableParallel
# RunnableLambda를 정의, "first"라는 이름을 부여
runnable1 = RunnableLambda(lambda x: {"foo": x}, name="first")
# 비동기 함수 정의
async def func(x):
for _ in range(5):
yield x # 입력값을 5번 반복해서 이벤트처럼 비동기적으로 생성
# 두 번째 RunnableLambda로 비동기 함수 연결, "second"라는 이름을 부여
runnable2 = RunnableLambda(func, name="second")
# 두 runnable을 연결하여 체인 생성
chain = runnable1 | runnable2
# 체인의 이벤트를 비동기적으로 처리하여 출력
async for event in chain.astream_events("bar", version="v2"):
print(f"event={event['event']} | name={event['name']} | data={event['data']}")
3.16. 완료한 순서대로 배치된 출력을 생성
import time
from langchain_core.runnables import RunnableLambda, RunnableParallel
# RunnableLambda를 정의하여 주어진 시간만큼 대기한 후, "slept x"를 출력하는 작업을 정의
runnable1 = RunnableLambda(lambda x: time.sleep(x) or print(f"slept {x}"))
# [5, 1]이라는 리스트를 입력으로 받아 각 요소를 순차적으로 처리
for idx, result in runnable1.batch_as_completed([5, 1]):
print(idx, result)
batch_as_completed 이 메서드는 병렬로 작업을 실행하고, 먼저 완료된 작업부터 순서대로 그 결과를 반환합니다.
or 연산자는 앞의 표현식이 False(또는 None)일 때만 뒤의 표현식을 평가합니다. 여기서 time.sleep(x)는 대기 시간을 발생시키며, None을 반환합니다. 따라서 time.sleep(x)가 None이므로 or 뒤의 print(f"slept {x}")가 실행됩니다.
3.17. 딕셔너리의 일부 리턴(Return subset of output dict)
이 코드는 LangChain에서 특정 키에 대해 변형을 적용하고, 결과 딕셔너리의 일부만 선택하는 방법을 보여줍니다. 두 가지 중요한 개념이 사용되었는데, 하나는 RunnableLambda로 입력 값을 변형하고, 다른 하나는 RunnablePassthrough로 입력 값의 일부를 전달하며 선택된 키만 결과로 반환하는 기능입니다.
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
runnable1 = RunnableLambda(lambda x: x["baz"] + 5)
chain = RunnablePassthrough.assign(foo=runnable1).pick(["foo", "bar"])
chain.invoke({"bar": "hi", "baz": 2})
3.18. 선언적으로 runnable의 배치 버전 만들기
선언적으로 체인을 구성한다는 것은 어떻게 동작하는지 일일이 절차를 기술하기보다, 원하는 동작을 고수준에서 간단히 표현하는 방식으로 체인을 구성한다는 의미입니다. 즉, 프로세스를 하나하나 명령형으로 작성하는 대신, 여러 작업을 연결하고 각각의 역할을 선언적으로 정의하여 수행하도록 설정하는 것입니다.
runnable2.map()은 리스트의 각 요소에 대해 5를 더하는 작업을 자동으로 적용하도록 선언적으로 정의된 것입니다. 따라서 프로그래머가 리스트의 각 요소를 일일이 처리하는 코드(예: 반복문)를 작성하지 않아도, 고수준에서 처리 과정을 정의하고 LangChain이 이를 알아서 처리하게 됩니다. 이 방식은 코드의 가독성과 유지보수성을 높이는 데 큰 도움이 됩니다.
from langchain_core.runnables import RunnableLambda
# 첫 번째 RunnableLambda는 입력 값 x를 받아, 0부터 x-1까지의 숫자로 리스트를 생성
runnable1 = RunnableLambda(lambda x: list(range(x)))
# 두 번째 RunnableLambda는 입력 값을 받아 5를 더하는 작업을 수행
runnable2 = RunnableLambda(lambda x: x + 5)
# 첫 번째 runnable1의 결과 리스트를 runnable2.map()에 넘겨, 각 요소에 5를 더한 후 반환하는 체인을 정의
chain = runnable1 | runnable2.map()
# 입력 값 3을 chain에 전달
# 0부터 2까지의 리스트 [0, 1, 2]가 생성되고, 각 요소에 5가 더해져 [5, 6, 7]이 반환됨
chain.invoke(3)
3.19. runnable의 그래프 만들기
print_ascii() 함수를 이용해 그래프를 그려 체인을 확인할 수 있어요.
from langchain_core.runnables import RunnableLambda, RunnableParallel
runnable1 = RunnableLambda(lambda x: {"foo": x})
runnable2 = RunnableLambda(lambda x: [x] * 2)
runnable3 = RunnableLambda(lambda x: str(x))
chain = runnable1 | RunnableParallel(second=runnable2, third=runnable3)
chain.get_graph().print_ascii()
출력
+-------------+
| LambdaInput |
+-------------+
*
*
*
+------------------------------+
| Lambda(lambda x: {'foo': x}) |
+------------------------------+
*
*
*
+-----------------------------+
| Parallel<second,third>Input |
+-----------------------------+
**** ***
**** ****
** **
+---------------------------+ +--------------------------+
| Lambda(lambda x: [x] * 2) | | Lambda(lambda x: str(x)) |
+---------------------------+ +--------------------------+
**** ***
**** ****
** **
+------------------------------+
| Parallel<second,third>Output |
+------------------------------+
3.20 체인의 전체 프롬프트 확인
get_prompts() 메서드를 이용해 프롬프트를 확인할 수 있어요.
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
prompt1 = ChatPromptTemplate.from_messages(
[("system", "good ai"), ("human", "{input}")]
)
prompt2 = ChatPromptTemplate.from_messages(
[
("system", "really good ai"),
("human", "{input}"),
("ai", "{ai_output}"),
("human", "{input2}"),
]
)
fake_llm = RunnableLambda(lambda prompt: "i am good ai")
chain = prompt1.assign(ai_output=fake_llm) | prompt2 | fake_llm
for i, prompt in enumerate(chain.get_prompts()):
print(f"**prompt {i=}**\n")
print(prompt.pretty_repr())
print("\n" * 3)
출력
**prompt i=0**
================================ System Message ================================
good ai
================================ Human Message =================================
{input}
**prompt i=1**
================================ System Message ================================
really good ai
================================ Human Message =================================
{input}
================================== AI Message ==================================
{ai_output}
================================ Human Message =================================
{input2}
3.21. lifecycle listeners
`Add lifecycle listeners`는 실행의 시작과 끝과 같은 특정 시점에 이벤트를 처리할 수 있는 리스너(Listener)를 추가하는 기능을 의미합니다. with_listeners() 메소드를 이용해 실행할 수 있습니다.
리스너(listener)는 특정 이벤트가 발생했을 때 자동으로 호출되어 작업을 수행하는 함수나 객체입니다. 이벤트 드리븐(event-driven) 방식의 프로그래밍에서 자주 사용됩니다.
이 코드는 체인이 실행될 때, 실행 시작과 종료 시점에 특정 작업을 수행하도록 리스너를 추가하는 방법을 보여줍니다.
import time
from langchain_core.runnables import RunnableLambda
from langchain_core.tracers.schemas import Run
# 실행이 시작될 때 실행 시간을 출력하는 리스너 함수
def on_start(run_obj: Run):
print("start_time:", run_obj.start_time)
# 실행이 종료될 때 종료 시간을 출력하는 리스너 함수
def on_end(run_obj: Run):
print("end_time:", run_obj.end_time)
# 입력받은 시간만큼 대기하는 함수 (RunnableLambda로 정의)
runnable1 = RunnableLambda(lambda x: time.sleep(x))
# 체인에 리스너 추가: 실행 시작 시 on_start, 종료 시 on_end 호출
chain = runnable1.with_listeners(on_start=on_start, on_end=on_end)
# 2초 동안 대기하는 체인을 실행
chain.invoke(2)