2장: LangChain 아키텍처와 LCEL 심층 분석

from langchain_core.runnables import Runnable
 
# 모든 Runnable이 공통으로 제공하는 메서드
# .invoke(input)      - 단일 입력 처리
# .stream(input)      - 스트리밍 출력
# .batch(inputs)      - 배치 처리
# .ainvoke(input)     - 비동기 호출
# .astream(input)     - 비동기 스트리밍
# .astream_events()   - 이벤트 스트리밍

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
 
# 프롬프트 -> 모델 -> 파서 체인
prompt = ChatPromptTemplate.from_template(
    "다음 주제에 대해 3문장으로 설명해주세요: {topic}"
)
model = ChatOpenAI(model="gpt-4o", temperature=0.7)
parser = StrOutputParser()
 
chain = prompt | model | parser
 
# 실행
result = chain.invoke({"topic": "마이크로서비스 아키텍처"})

from langchain_core.runnables import RunnableParallel
 
# 동일한 입력으로 여러 체인을 병렬 실행
analysis_chain = RunnableParallel(
    summary=prompt_summary | model | parser,
    keywords=prompt_keywords | model | parser,
    sentiment=prompt_sentiment | model | parser,
)
 
# 세 체인이 동시에 실행됨
results = analysis_chain.invoke({"text": "분석할 텍스트..."})
# results["summary"], results["keywords"], results["sentiment"]

from langchain_core.runnables import RunnableBranch
 
# 조건에 따라 다른 체인 실행
route = RunnableBranch(
    (lambda x: "코드" in x["question"], code_chain),
    (lambda x: "데이터" in x["question"], data_chain),
    general_chain,  # 기본 경로
)
 
result = route.invoke({"question": "Python 코드를 리뷰해주세요"})

from langchain_core.runnables import RunnableLambda
 
def preprocess(input_dict: dict) -> dict:
    """입력 전처리"""
    return {
        "question": input_dict["question"].strip().lower(),
        "context": input_dict.get("context", ""),
    }
 
chain = RunnableLambda(preprocess) | prompt | model | parser

from langchain_core.runnables import RunnableConfig
 
# 기본 체인 정의
chain = prompt | model | parser
 
# 재시도 미들웨어 적용
chain_with_retry = chain.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
)
 
# 폴백 미들웨어 적용
fallback_model = ChatOpenAI(model="gpt-4o-mini")
fallback_chain = prompt | fallback_model | parser
 
chain_with_fallback = chain_with_retry.with_fallbacks(
    [fallback_chain]
)
 
# 속도 제한
from langchain_core.rate_limiters import InMemoryRateLimiter
 
rate_limiter = InMemoryRateLimiter(
    requests_per_second=10,
    check_every_n_seconds=0.1,
)
 
model_with_rate_limit = ChatOpenAI(
    model="gpt-4o",
    rate_limiter=rate_limiter,
)

from langchain_core.runnables import RunnableConfig
from langchain_core.callbacks import CallbackManagerForChainRun
import time
 
def timing_middleware(chain):
    """실행 시간을 측정하는 미들웨어"""
    original_invoke = chain.invoke
 
    def timed_invoke(input, config=None, **kwargs):
        start = time.perf_counter()
        result = original_invoke(input, config=config, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"Chain execution: {elapsed:.3f}s")
        return result
 
    chain.invoke = timed_invoke
    return chain

from langchain_core.messages import HumanMessage
 
# 텍스트 + 이미지 멀티모달 메시지
message = HumanMessage(
    content=[
        {"type": "text", "text": "이 이미지에서 무엇이 보이나요?"},
        {
            "type": "image_url",
            "image_url": {"url": "https://example.com/image.png"},
        },
    ]
)
 
response = model.invoke([message])

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanExporter
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
    OTLPSpanExporter,
)
 
# OpenTelemetry 설정
provider = TracerProvider()
exporter = OTLPSpanExporter(endpoint="http://localhost:4317")
provider.add_span_processor(BatchSpanExporter(exporter))
trace.set_tracer_provider(provider)
 
# LangChain은 자동으로 OpenTelemetry 트레이스를 생성
chain = prompt | model | parser
result = chain.invoke({"topic": "AI"})
# -> 자동으로 span 생성: prompt -> model -> parser

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import (
    RunnablePassthrough,
    RunnableParallel,
)
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
 
# 1. 벡터 스토어 설정
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(
    collection_name="docs",
    embedding_function=embeddings,
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
 
# 2. 프롬프트 정의
rag_prompt = ChatPromptTemplate.from_template("""
다음 컨텍스트를 바탕으로 질문에 답변해주세요.
컨텍스트에 없는 내용은 "해당 정보를 찾을 수 없습니다"라고 답변해주세요.
 
컨텍스트:
{context}
 
질문: {question}
""")
 
# 3. 모델 설정 (속도 제한 + 폴백)
from langchain_core.rate_limiters import InMemoryRateLimiter
 
rate_limiter = InMemoryRateLimiter(requests_per_second=10)
 
primary_model = ChatOpenAI(
    model="gpt-4o",
    rate_limiter=rate_limiter,
)
fallback_model = ChatOpenAI(model="gpt-4o-mini")
 
# 4. 문서 포맷팅 함수
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)
 
# 5. RAG 체인 조합
rag_chain = (
    RunnableParallel(
        context=retriever | format_docs,
        question=RunnablePassthrough(),
    )
    | rag_prompt
    | primary_model.with_fallbacks([fallback_model])
    | StrOutputParser()
).with_retry(stop_after_attempt=3)
 
# 6. 실행
result = rag_chain.invoke("LangChain의 LCEL이란 무엇인가요?")