Recap — LCEL Patterns You'll Use Constantly

# LCEL Patterns You'll Use Constantly LCEL (LangChain Expression Language) is pipe-based composition. The `|` operator chains components: output of left becomes input of right. Five patterns handle most production chains. ## Pattern 1 — Linear chain The basics: ```python from langchain_core.prompts import ChatPromptTemplate from langchain_anthropic import ChatAnthropic from langchain_core.output_parsers import StrOutputParser prompt = ChatPromptTemplate.from_template( "Translate this to {language}: {text}" ) llm = ChatAnthropic(model="claude-sonnet-4-6") parser = StrOutputParser() chain = prompt | llm | parser result = chain.invoke({"text": "Hello world", "language": "Hindi"}) # "नमस्ते दुनिया" ``` `prompt` produces a `PromptValue`. `llm` consumes it, produces a `ChatMessage`. `parser` extracts the string. Each component is type-aware. LangChain validates types at construction. ## Pattern 2 — Structured output with Pydantic Get parsed JSON automatically: ```python from pydantic import BaseModel, Field from langchain_core.output_parsers import PydanticOutputParser class Customer(BaseModel): name: str = Field(description="Customer's full name") email: str = Field(description="Email address") issue: str = Field(description="Brief description of their issue") parser = PydanticOutputParser(pydantic_object=Customer) prompt = ChatPromptTemplate.from_template( "Extract customer info from this message:\n{message}\n\n{format_instructions}" ).partial(format_instructions=parser.get_format_instructions()) chain = prompt | llm | parser result = chain.invoke({"message": "Hi I'm Anuj (anuj@x.com), shoes don't fit"}) # Customer(name="Anuj", email="anuj@x.com", issue="shoes don't fit") ``` Even better in 2026 — most LLM providers support structured output natively. Use `llm.with_structured_output(Customer)`: ```python structured_llm = llm.with_structured_output(Customer) chain = prompt | structured_llm # No parser needed; schema enforced server-side ``` ## Pattern 3 — Parallel branches Run multiple chains simultaneously: ```python from langchain_core.runnables import RunnableParallel joke_chain = ( ChatPromptTemplate.from_template("Tell a joke about {topic}") | llm | StrOutputParser() ) poem_chain = ( ChatPromptTemplate.from_template("Write a poem about {topic}") | llm | StrOutputParser() ) parallel = RunnableParallel(joke=joke_chain, poem=poem_chain) result = parallel.invoke({"topic": "programming"}) # {"joke": "Why do programmers...", "poem": "Code flows..."} ``` Both chains run concurrently. Useful for: - Generating multiple outputs (summary + tags + sentiment). - A/B testing prompts (run both, compare). - Independent sub-tasks. ## Pattern 4 — Conditional routing (RunnableBranch) Route to different chains based on input: ```python from langchain_core.runnables import RunnableBranch classify_chain = ( ChatPromptTemplate.from_template( "Is this question about 'product' or 'support'? Output one word.\n{question}" ) | llm | StrOutputParser() ) product_chain = ChatPromptTemplate.from_template( "Answer this product question: {question}" ) | llm | StrOutputParser() support_chain = ChatPromptTemplate.from_template( "Answer this support question: {question}" ) | llm | StrOutputParser() router = RunnableBranch( (lambda x: "product" in x["classification"].lower(), product_chain), (lambda x: "support" in x["classification"].lower(), support_chain), product_chain, # default ) full_chain = ( {"question": lambda x: x["question"], "classification": classify_chain} | router ) result = full_chain.invoke({"question": "How do I return shoes?"}) ``` Classifier decides; router routes. For complex routing, prefer LangGraph (Module 2). ## Pattern 5 — RAG chain The standard retrieval-augmented chain: ```python from langchain_core.runnables import RunnablePassthrough retriever = vectorstore.as_retriever(search_kwargs={"k": 5}) rag_prompt = ChatPromptTemplate.from_template(''' Answer the question using ONLY the context. If the context doesn't have the answer, say "I don't know". Context: {context} Question: {question} ''') def format_docs(docs): return "\n\n".join(d.page_content for d in docs) rag_chain = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | rag_prompt | llm | StrOutputParser() ) answer = rag_chain.invoke("How do I return a defective product?") ``` Notice `RunnablePassthrough` — passes the input through unchanged (the user question stays as `question`). The retriever fetches docs based on the question; `format_docs` joins them into a string. ## Streaming, batching, async Every LCEL chain has these for free: ```python # Stream tokens as they generate for chunk in chain.stream({"text": "..."}): print(chunk, end="") # Batch many inputs (parallel under the hood) results = chain.batch([{"text": "a"}, {"text": "b"}, {"text": "c"}]) # Async result = await chain.a