LangChain Expression Language (LCEL) 심화

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_anthropic import ChatAnthropic

prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}")
model = ChatAnthropic(model="claude-sonnet-4-6")
parser = StrOutputParser()

# 체인 = prompt → model → parser
chain = prompt | model | parser

result = chain.invoke({"topic": "Python"})
print(result)

from langchain_core.runnables import RunnablePassthrough

# 입력에서 일부만 추출
chain = (
    {"context": RunnablePassthrough(), "question": RunnablePassthrough()}
    | prompt
    | model
)

# 사용
chain.invoke({"context": "...", "question": "..."})

from langchain_core.runnables import RunnableLambda

def to_uppercase(text: str) -> str:
    return text.upper()

# 체인에 함수 삽입
chain = prompt | model | parser | RunnableLambda(to_uppercase)


# 또는 직접
chain = prompt | model | parser | (lambda x: x.upper())

from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_anthropic import ChatAnthropic
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

# 1. 벡터 DB
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(embedding_function=embeddings, persist_directory="./db")
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# 2. 프롬프트
prompt = ChatPromptTemplate.from_template("""
다음 컨텍스트만 참고해서 답변하세요. 모르면 "모릅니다"라고 답변.

컨텍스트:
{context}

질문: {question}
""")

model = ChatAnthropic(model="claude-sonnet-4-6")

# 3. RAG 체인
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

# 사용
result = rag_chain.invoke("LangChain 시작 방법은?")

from langchain_core.runnables import RunnableBranch

# 입력에 따라 다른 체인 실행
def is_question_about_python(input):
    return "python" in input["question"].lower()


python_chain = python_prompt | model | parser
general_chain = general_prompt | model | parser

branched = RunnableBranch(
    (is_question_about_python, python_chain),
    general_chain,  # default
)

result = branched.invoke({"question": "Python list comprehension?"})

from langchain_core.runnables import RunnableParallel

# 여러 체인을 동시에 실행
parallel = RunnableParallel(
    summary=summary_chain,
    keywords=keywords_chain,
    sentiment=sentiment_chain,
)

result = parallel.invoke({"text": "..."})
# {
#   "summary": "...",
#   "keywords": [...],
#   "sentiment": "positive",
# }
# 직렬보다 3배 빠름