RunnablePassthrough가 무엇이고 어떻게 사용하는지, 그리고 RunnablePassthrough.assign에 대한 사용법을 기본적인 RAG 구성하는 방법을 통해 알아보자.
1. 기본적인 RAG 구성방법
RAG를 구성할 때 일반적으로 vector store를 검색하고 그 결과를 llm으로 던져서 결과를 가져온다. 그래서 일반적인 사용방법은 아래와 같다.
from dotenv import load_dotenv
from langchain_chroma import Chroma
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
load_dotenv()
embeddings = OpenAIEmbeddings()
vector_store = Chroma(
persist_directory="./chroma_db",
embedding_function=embeddings,
)
retriever = vector_store.as_retriever()
template = """Answer the question based only on the following context:
{context}
Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
model = ChatOpenAI()
chain = (
prompt
| model
| StrOutputParser()
)
# vector store에서 검색
retrieved_docs = retriever.invoke("한모 씨의 여름 휴가는 언제야?")
context = format_docs(retrieved_docs) # 검색된 Document를 format하여 문자열로 context에 저장한다.
result = chain.invoke({"context": context, "question": "한모 씨의 여름 휴가는 언제야?"})
print(result) # 한모 씨의 여름 휴가는 6월에 예정되어 있다.위와 같이 구성하면 retriever에서 get_relevant_documents 혹은 invoke를 통해 검색한 후에 chain을 통해 호출을 한다. 하지만 chain에서 이를 한번에 처리하도록 할 수도 있다. LCEL 방식을 사용하면 되는데 retriever를 인자로 넘길 때 자동으로 invoke가 호출이 된다.
2. LCEL 방식을 사용한 호출
chain = (
{"context": retriever}
| prompt
| model
| StrOutputParser()
)
result = chain.invoke({"input": "한모 씨의 여름 휴가는 언제야?"})위와 같이 LCEL을 사용하여 context를 넘기면 retriever에서 invoke가 호출된 결과와 함께 사용할 수 있다. 하지만 이를 호출하면 오류가 발생한다. invoke에서 dict 타입이 아니라 str 타입을 사용하도록 되어 있다.
그래서 input값을 str으로 바꿔줄려면 RunnablePassthrough를 통해 같이 넘기는 방식을 사용하면 된다.
chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt
| model
| StrOutputParser()
)
result = chain.invoke("한모 씨의 여름 휴가는 언제야?")
print(result) # 한모 씨의 여름 휴가는 6월에 예정되어 있다.위의 코드에서 {"context": retriever, "question": RunnablePassthrough()}의 실행방법을 설명하면 아래와 같다.
retriever는Runnable타입이라invoke가 자동으로 호출이 되고RunnablePassthrough는chain.invoke에 입력된 파라미터가 전달되는 것이다.
langchain.debug = True를 통해 호출해보면 아래와 같이 값이 전달되는 것을 확인할 수 있다.
{
"prompts": [
"Human: Answer the question based only on the following context:\n[Document(metadata={'source': './data/news.txt'}, page_content='40대 직장인 한모 씨는 올해 여름휴가를 예년보다 앞당겨 6월에 쓰기로 했다. ..., Document(metadata={'source': './data/news.txt'}, page_content=\"xxxxxxxxx, Document(metadata={'source': './data/news.txt'}, page_content=\"xxxxxxxxx, Document(metadata={'source': './data/news.txt'}, page_content=\"xxxxxxxxx\")]\n\nQuestion: 한모 씨의 여름 휴가는 언제야?"
]
}만일 prompt의 문서를 format을 하고 싶으면 아래와 같이 사용하면 된다. (format_docs 추가)
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
chain = (
{
"context": retriever | format_docs,
"question": RunnablePassthrough(),
}
| prompt
| model
| StrOutputParser()
)실행되는 내용을 확인을 하면 아래와 같이 문서의 내용만 가져다가 llm으로 전달을 한다.
{
"prompts": [
"Human: Answer the question based only on the following context:\n40대 직장인 한모 씨는 올해 여름휴가를 예년보다 앞당겨 6월에 쓰기로 했다. xxxxxxx..."
]
}3. prompt에 파라미터를 추가하는 경우
위의 코드에서 context는 retriever를 통해 문서의 내용이 추가되고 question은 invoke의 매개변수로 전달이 되었다. 여기서 prompt에 현재 시간을 추가적으로 넘기는 경우를 생각해보자.
template = """Answer the question based only on the following context:
{context}
current date: {current_date}
Question: {question}
"""기존 prompt에 current_date를 추가되었다. 이 경우는 어떻게 실행할 수 있을까? RunnablePassthrough.assign을 통해 사용할 수 있다.
def current_date():
return "2024-09-22"
chain = (
{
"context": retriever | format_docs,
"question": RunnablePassthrough(),
}
| RunnablePassthrough.assign(current_date=lambda x: current_date())
| prompt
| model
| StrOutputParser()
)이렇게 실행이 되면 기존 context, question외에 current_date가 파라미터로 넘어간다. 즉 아래와 같다.
{
"context": "문서의 내용",
"question": "질문 내용",
"current_date": "2024-09-22"
}current_date()를 함수로 정의하고 RunnablePassthrough.assign를 통해 함수의 결과를 리턴하도록 하면 된다. (문자열 자체를 넘길 수는 없고 Runnable로 정의해야 한다)
RunnablePassthrough.assign 사용법
RunnablePassthrough.assign의 기본 사용법에 대한 설명이다.
langchain RAG 코드를 보다 보면 아래의 예제가 많이 보인다. 여기서 RunnablePassthrough.assign은 무슨 역할을 하는 것일까?
rag_chain_from_docs = (
RunnablePassthrough.assign(context=(lambda x: format_docs(x["context"])))
| prompt
| model
| StrOutputParser()
)이 코드에서는 RunnablePassthrough.assign을 사용하여 입력 데이터에서 "context" 키에 해당하는 값을 가공하여 format_docs 함수로 변환한 후, 새롭게 "context" 필드에 할당한다. 이 변환된 값은 이후의 체인 (즉, prompt, model, StrOutputParser)으로 전달된다.
그러면 RunnablePassthrough.assign에 대해 좀 더 알아보자.
RunnablePassthrough.assign는 LangChain에서 제공하는 Runnable 인터페이스 중 하나로, 입력 데이터를 그대로 전달하거나, 입력 데이터에 새로운 값을 추가하여 출력을 생성하는 데 사용된다.
RunnablePassthrough는 주로 입력을 수정하지 않고 그대로 전달할 때 유용하며, 여기에 .assign 메소드를 사용하면 입력 데이터에 추가적인 키-값 쌍을 쉽게 할당할 수 있다.
RunnablePassthrough.assign의 작동 방식
- 입력 그대로 전달:
RunnablePassthrough는 입력 데이터를 가공하지 않고 그대로 전달하는 역할을 한다. - 키-값 쌍 추가:
.assign메소드를 사용하면 입력 데이터를 유지하면서 추가적으로 지정된 값들을 입력에 할당한다. 예를 들어, 딕셔너리 형태의 입력에 새로운 키-값을 추가하고 싶을 때 유용하다.
RunnablePassthrough.assign의 기본 사용법
기본적으로 assign 메서드는 기존 입력 데이터에 새로운 필드를 추가하거나 기존 필드를 수정한다. 아래는 그 기본 사용 예시이다.
from langchain.schema.runnable import RunnablePassthrough
# 예시 데이터
input_data = {"name": "Alice", "age": 30}
# assign을 사용해 동적으로 필드 추가 및 수정
runnable = RunnablePassthrough().assign(greeting=lambda x: f"Hello, {x['name']}!")
# 결과 출력
result = runnable.invoke(input_data)
print(result)실행결과
{'name': 'Alice', 'age': 30, 'greeting': 'Hello, Alice!'}위 예제에서는 name과 age 필드를 유지하면서 greeting 필드를 새로 추가하여 데이터를 동적으로 업데이트했다.
RunnablePassthrough.assign의 고급 사용법
RunnablePassthrough.assign의 강력한 점은 단순한 값 할당뿐만 아니라 함수 호출 결과를 필드로 동적으로 추가할 수 있다는 것이다. 이를 통해 데이터의 흐름을 유연하게 처리할 수 있다. 예를 들어, 외부 API로부터 데이터를 받아와 그 결과를 필드로 추가할 수 있다.
import time
from langchain.schema.runnable import RunnablePassthrough
# 시간 기반으로 데이터를 추가하는 함수
def get_timestamp(_):
return time.time()
# 기본 입력 데이터
input_data = {"name": "Bob", "age": 25}
# assign을 통해 필드 업데이트
runnable = RunnablePassthrough().assign(timestamp=get_timestamp)
# 결과 출력
result = runnable.invoke(input_data)
print(result)실행결과
{'name': 'Bob', 'age': 25, 'timestamp': 1695239803.4874}위 예제에서는 get_timestamp 함수가 현재 타임스탬프를 반환하며, 이 값을 timestamp 필드에 할당하고 있다. 이렇게 동적인 필드 생성 및 업데이트는 매우 다양한 응용 가능성을 제공한다.
실전 예제: 사용자 데이터 처리 파이프라인
이제 RunnablePassthrough.assign을 좀 더 복잡한 시나리오에 적용해 보자. 예를 들어, 사용자의 프로필 데이터를 처리하면서 몇 가지 추가적인 데이터를 동적으로 생성하는 경우이다.
from langchain.schema.runnable import RunnablePassthrough
# 사용자 데이터를 받아 동적으로 이메일 도메인 생성
def extract_email_domain(data):
email = data["email"]
return email.split("@")[-1]
# 사용자 정보
input_data = {
"name": "Charlie",
"email": "charlie@example.com",
"age": 28
}
# assign을 사용해 동적으로 도메인 필드를 추가
runnable = RunnablePassthrough().assign(domain=extract_email_domain)
# 결과 출력
result = runnable.invoke(input_data)
print(result)실행결과
{'name': 'Charlie', 'email': 'charlie@example.com', 'age': 28, 'domain': 'example.com'}이 예제에서는 이메일에서 도메인을 추출하여 domain 필드를 새로 생성했다. 이런 방식으로 동적 데이터 파이프라인을 구축하면, 각 단계에서 필요한 데이터를 추가하거나 수정하는 것이 매우 간편해진다.
