LangChain Chains 가이드

LangChain Chains 완벽 가이드

LangChain의 체인(Chain)은 여러 컴포넌트를 연결하여 복잡한 LLM 애플리케이션을 구축하는 핵심 개념입니다.

LCEL (LangChain Expression Language)

LCEL은 체인을 선언적으로 구성하는 새로운 방법입니다.

기본 구문

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 컴포넌트 정의
prompt = ChatPromptTemplate.from_template("번역해줘: {text}")
model = ChatOpenAI(model="gpt-4o")
output_parser = StrOutputParser()

# 파이프 연산자로 체인 구성
chain = prompt | model | output_parser

# 실행
result = chain.invoke({"text": "Hello, world!"})

병렬 실행

from langchain_core.runnables import RunnableParallel

# 여러 체인 병렬 실행
analysis = RunnableParallel(
    summary=summary_chain,
    sentiment=sentiment_chain,
    keywords=keyword_chain,
)

result = analysis.invoke({"text": document})

주요 체인 패턴

1. Sequential Chain

# 순차 실행
chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | output_parser
)

2. Branching Chain

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: "코드" in x["query"], code_chain),
    (lambda x: "번역" in x["query"], translation_chain),
    default_chain
)

3. Retrieval Chain (RAG)

from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain

# 문서 체인 생성
document_chain = create_stuff_documents_chain(llm, prompt)

# 검색 체인 생성
retrieval_chain = create_retrieval_chain(retriever, document_chain)

response = retrieval_chain.invoke({"input": "질문"})

고급 기능

스트리밍

for chunk in chain.stream({"text": "긴 텍스트"}):
    print(chunk, end="", flush=True)

배치 처리

inputs = [
    {"text": "Hello"},
    {"text": "World"},
    {"text": "AI"}
]

results = chain.batch(inputs, config={"max_concurrency": 5})

폴백 처리

from langchain_core.runnables import RunnableWithFallbacks

chain_with_fallback = main_chain.with_fallbacks([
    fallback_chain_1,
    fallback_chain_2
])

메모리 통합

from langchain.memory import ConversationBufferMemory
from langchain_core.runnables import RunnableWithMessageHistory

memory = ConversationBufferMemory(return_messages=True)

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history=lambda session_id: memory,
    input_messages_key="input",
    history_messages_key="history",
)