728x90
반응형
개요
LangChain은 대규모 언어 모델(LLM)을 활용한 애플리케이션 개발을 위한 강력한 프레임워크이다. 그중 RunnableParallel
은 여러 작업을 동시에 실행하여 처리 시간을 단축하고 효율성을 높일 수 있는 핵심 기능이다. 본 글에서는 LangChain의 RunnableParallel의 개념, 활용 방법 및 다양한 예시를 통해 실제 구현 방법을 살펴본다.
설명
RunnableParallel
은 LangChain의 스키마 중 하나로, 여러 실행 가능한 컴포넌트를 병렬로 처리할 수 있게 해주는 클래스이다. 이는 동일한 입력을 여러 프로세스에 동시에 전달하여 각각 다른 작업을 수행하고, 그 결과를 하나의 딕셔너리로 수집한다.
주요 사용 사례:
- 동일한 텍스트에 대해 여러 분석을 동시에 수행
- 다양한 LLM 모델의 응답을 비교
- 여러 문서에 대한 처리를 병렬화하여 시간 단축
- 복잡한 워크플로우에서의 맵-리듀스 패턴 구현
특징
RunnableParallel의 주요 특징은 다음과 같다:
- 동시성 처리: 여러 작업을 병렬로 실행하여 전체 처리 시간을 단축한다.
- 구조화된 출력: 각 병렬 작업의 결과를 키-값 형태의 딕셔너리로 반환하여 후속 처리가 용이하다.
- 체인 결합 가능: 다른 LangChain 컴포넌트들과 쉽게 결합하여 복잡한 파이프라인 구성이 가능하다.
- 비동기 지원:
invoke()
뿐만 아니라ainvoke()
메서드를 통한 비동기 실행을 지원한다. - 재시도 메커니즘:
with_retry()
메서드를 통해 실패한 작업의 재시도 로직을 쉽게 구현할 수 있다. - 스케일링: 대량의 데이터를 처리할 때 효율적인 자원 활용이 가능하다.
Example
LangChain의 RunnableParallel을 활용한 다양한 예제를 살펴보자:
1. 기본 병렬 실행 예제
from langchain.schema.runnable import RunnableParallel
from langchain.prompts import PromptTemplate
from langchain.chat_models import ChatOpenAI
from langchain.schema.output_parser import StrOutputParser
# LLM 정의
llm = ChatOpenAI(temperature=0)
# 두 개의 다른 요약 생성을 병렬로 실행
summarizer = RunnableParallel(
short_summary=PromptTemplate.from_template(
"Summarize the following text in 1-2 sentences: {text}"
) | llm | StrOutputParser(),
detailed_summary=PromptTemplate.from_template(
"Summarize the following text in 5-7 sentences with key points: {text}"
) | llm | StrOutputParser(),
)
# 병렬 실행
result = summarizer.invoke({"text": "Your long article text here..."})
print("Short summary:", result["short_summary"])
print("\nDetailed summary:", result["detailed_summary"])
2. 다양한 분석 작업 병렬 처리
text_analyzer = RunnableParallel(
sentiment=PromptTemplate.from_template(
"Analyze the sentiment of this text. Return only 'positive', 'negative', or 'neutral': {input}"
) | llm | StrOutputParser(),
key_topics=PromptTemplate.from_template(
"Extract exactly 3 main topics from this text as a comma-separated list: {input}"
) | llm | StrOutputParser(),
audience=PromptTemplate.from_template(
"Who is the target audience for this text? Be brief: {input}"
) | llm | StrOutputParser(),
language_complexity=PromptTemplate.from_template(
"On a scale of 1-10, how complex is the language in this text? Just provide the number: {input}"
) | llm | StrOutputParser(),
)
# 사용 예
analysis = text_analyzer.invoke({"input": "Climate change is a pressing issue that requires immediate action..."})
3. 여러 LLM 모델 병렬 비교
prompt = PromptTemplate.from_template("Answer this question: {question}")
# 다양한 모델을 사용하여 동일한 질문에 대한 응답 비교
model_comparison = RunnableParallel(
gpt4=prompt | ChatOpenAI(model="gpt-4") | StrOutputParser(),
gpt35=prompt | ChatOpenAI(model="gpt-3.5-turbo") | StrOutputParser(),
claude=prompt | ChatAnthropic(model="claude-2") | StrOutputParser(),
)
# 질문에 대한 다양한 모델의 응답 비교
responses = model_comparison.invoke({"question": "What are the ethical implications of AI?"})
4. 맵-리듀스 패턴과 병렬 처리 조합
# 여러 문서에 대한 병렬 처리 함수
async def process_documents(documents):
# 각 문서에 대한 처리 함수 정의
async def process_single_doc(doc):
# 단일 문서에 대한 병렬 처리
doc_processor = RunnableParallel(
summary=summarize_chain,
key_facts=extract_chain,
metadata=RunnableLambda(lambda x: {"id": doc.get("id"), "source": doc.get("source")})
)
return await doc_processor.ainvoke({"text": doc.get("content")})
# 모든 문서를 동시에 처리
tasks = [process_single_doc(doc) for doc in documents]
return await asyncio.gather(*tasks)
결론
LangChain의 RunnableParallel은 복잡한 LLM 애플리케이션 개발에 있어 처리 시간을 단축하고 효율성을 극대화할 수 있는 강력한 도구이다. 특히 여러 LLM 호출이 필요한 시나리오, 다양한 분석이 요구되는 작업, 대량의 문서 처리 등에서 큰 효과를 발휘한다.
병렬 처리의 장점을 활용하여 사용자 경험을 개선하고, 자원 활용을 최적화할 수 있으며, 특히 비동기 처리와 결합하여 더욱 효율적인 애플리케이션 구축이 가능하다. 이는 프로덕션 환경에서의 LLM 애플리케이션 개발 시 고려해야 할 중요한 패턴 중 하나라고 할 수 있다.
참고문헌
728x90
반응형
'AI > langchain' 카테고리의 다른 글
LangChain vs LangGraph (0) | 2025.06.22 |
---|