LangChain Runnable Parallel 실행 예제

개요

LangChain은 대규모 언어 모델(LLM)을 활용한 애플리케이션 개발을 위한 강력한 프레임워크이다. 그중 RunnableParallel은 여러 작업을 동시에 실행하여 처리 시간을 단축하고 효율성을 높일 수 있는 핵심 기능이다. 본 글에서는 LangChain의 RunnableParallel의 개념, 활용 방법 및 다양한 예시를 통해 실제 구현 방법을 살펴본다.

설명

RunnableParallel은 LangChain의 스키마 중 하나로, 여러 실행 가능한 컴포넌트를 병렬로 처리할 수 있게 해주는 클래스이다. 이는 동일한 입력을 여러 프로세스에 동시에 전달하여 각각 다른 작업을 수행하고, 그 결과를 하나의 딕셔너리로 수집한다.

주요 사용 사례:

• 동일한 텍스트에 대해 여러 분석을 동시에 수행
• 다양한 LLM 모델의 응답을 비교
• 여러 문서에 대한 처리를 병렬화하여 시간 단축
• 복잡한 워크플로우에서의 맵-리듀스 패턴 구현

특징

RunnableParallel의 주요 특징은 다음과 같다:

1. 동시성 처리: 여러 작업을 병렬로 실행하여 전체 처리 시간을 단축한다.
2. 구조화된 출력: 각 병렬 작업의 결과를 키-값 형태의 딕셔너리로 반환하여 후속 처리가 용이하다.
3. 체인 결합 가능: 다른 LangChain 컴포넌트들과 쉽게 결합하여 복잡한 파이프라인 구성이 가능하다.
4. 비동기 지원: invoke() 뿐만 아니라 ainvoke() 메서드를 통한 비동기 실행을 지원한다.
5. 재시도 메커니즘: with_retry() 메서드를 통해 실패한 작업의 재시도 로직을 쉽게 구현할 수 있다.
6. 스케일링: 대량의 데이터를 처리할 때 효율적인 자원 활용이 가능하다.

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 애플리케이션 개발 시 고려해야 할 중요한 패턴 중 하나라고 할 수 있다.

참고문헌

• LangChain 공식 문서 - Runnables
• LangChain 공식 문서 - RunnableParallel
• LangChain GitHub 저장소
• LangChain 병렬 처리 예제