LangChain Core — Chains, Memory & RAG

The one idea: Runnables + LCEL essential

People bounce off LangChain because they think it's a hundred classes to memorize. It isn't. It's essentially one idea wearing a lot of hats, and the day that idea clicks, the entire framework stops feeling like a maze and starts feeling like a single, repeatable move. Miss the idea and you'll copy-paste tutorials forever, half-understanding each one; get it and you can read any LangChain snippet at a glance. So slow down on this section — everything after it is just this same trick applied again.

Here's the analogy that unlocks it: think of the Unix command line. You don't have one giant program that fetches, filters, sorts, and counts — you have small tools (cat, grep, sort, wc) and a | that pipes the output of one straight into the next. Each tool does one job and speaks a common language (text in, text out), so you can snap them together in any order. LangChain does exactly this for LLM apps: every building block speaks a common language, and | snaps them together. A prompt, a model, a parser, a retriever — each is a small tool; the pipe is the plumbing.

The common language has a name: the Runnable. A Runnable is simply anything with an .invoke(input) method — hand it an input, get an output. That's the whole contract. (It also offers .stream() for chunk-by-chunk output, .batch() for many inputs at once, and async variants — but those all come free once a thing is a Runnable.) Because prompts, models, parsers, and retrievers all implement that same tiny interface, they're interchangeable Lego bricks: they fit together because they share one shape.

The common mistake is reading chain = prompt | model | parser and assuming | is some LangChain magic keyword. It isn't magic and it isn't special syntax — it's Python's ordinary or operator, which every Runnable overloads (via __or__) to mean "compose." a | b builds a new Runnable that runs a, feeds its output into b, and returns the result. That's why you can keep chaining: the result of a pipe is itself a Runnable, so it too can be piped, invoked, streamed, or batched. Once you see that | is just function composition dressed in a friendly operator, LCEL holds no more surprises. Now the precise version: LCEL (LangChain Expression Language) lets you connect Runnables with the | operator, so the output of the left becomes the input of the right — exactly like a Unix pipe. A Runnable is anything with an .invoke(input) method (also .stream(), .batch(), and async variants). Prompts, models, parsers, retrievers — all Runnables.

Lab L2.1 · Your first chain essential

The fastest way to believe the "one idea" is to watch three lines do what took a dozen in Chapter 2. Back then, calling a model was a little ritual: assemble the messages, call the API, then reach into the response object to dig out the text. Three chores, three chances to get the plumbing wrong. This lab collapses all three into a single object you build once and run in one line — and the payoff isn't just brevity, it's that the object is now composable.

Picture an assembly line with three stations. The first station (the prompt) takes your raw inputs — a topic, an audience — and stamps them into a properly formatted message. The second (the model) takes that message and produces a reply. The third (the parser) trims the reply down to just the plain text you actually wanted. You don't carry the part between stations by hand; the conveyor belt — the | pipe — does it. Build the line once, then feed it inputs all day.

The mistake to avoid here is treating each piece as glued permanently to the others. Because prompt, model, and parser are independent Runnables joined only by pipes, any one is a drop-in swap: point at a different model, or replace the string parser with a typed one (that's the very next lab), and the other two stations don't notice. The values flowing through change; the wiring doesn't. That separation — stable structure, swappable parts — is exactly the maintainability win the whole chapter is selling, and you'll feel it first right here in the smallest possible chain.

shellpip install langchain langchain-core langchain-anthropic

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

model = ChatAnthropic(model="claude-opus-4-8", max_tokens=512)
prompt = ChatPromptTemplate.from_template(
    "Explain {topic} to a {audience} in two sentences."
)

chain = prompt | model | StrOutputParser()   # LCEL: three Runnables piped

print(chain.invoke({"topic": "vector databases", "audience": "new engineer"}))

shellpip install langchain langchain-core langchain-openai

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

model = ChatOpenAI(model="gpt-5.5", max_tokens=512)
prompt = ChatPromptTemplate.from_template(
    "Explain {topic} to a {audience} in two sentences."
)

chain = prompt | model | StrOutputParser()   # LCEL: three Runnables piped

print(chain.invoke({"topic": "vector databases", "audience": "new engineer"}))

Lab L2.2 · Structured output essential

The moment code — not a human — reads the model's answer, prose becomes a landmine. A person shrugs off "Sure! This looks like a billing issue, probably high priority." Your if ticket.priority == "high": line does not: it needs the word high sitting in a known field, spelled exactly, every single time. The instant you try to parse a free-text answer with string-slicing or a regex, you've signed up for a bug that shows up only when the model phrases things a little differently — which it will, tomorrow, in production, on the input you didn't test.

Think of it as the difference between a handwritten note and a form with checkboxes. The note might contain the answer, but you have to read and interpret it, and two notes never look quite alike. A form constrains the writer: category must be one of these four boxes, priority one of these three, summary goes on this line. Structured output turns the model from a note-writer into a form-filler. You hand it a schema — a Pydantic model listing the exact fields and their allowed values — and you get back a real Python object with typed attributes, not a paragraph you have to decode.

The common mistake is thinking this is a LangChain invention you're now locked into. It isn't — it's the same structured-output guarantee you met with the raw SDK in Chapter 2 (with_structured_output is just a thin wrapper over the provider's native feature), and the Literal[...] types become an enum the model is constrained to satisfy, so a surprise value like "urgent" simply can't come back. What LangChain adds is that the wrapped model is still a Runnable, so this typed step pipes into a chain exactly like everything else. Now the precise version: free text is a liability once code consumes it (Chapter 2), so LangChain gives you a typed path — bind a schema to the model and it returns a validated object.

structured.pyfrom pydantic import BaseModel, Field
from typing import Literal
from langchain_anthropic import ChatAnthropic

class Ticket(BaseModel):
    category: Literal["bug","feature","billing","other"]
    priority: Literal["low","medium","high"]
    summary: str = Field(description="one-line summary")

model = ChatAnthropic(model="claude-opus-4-8", max_tokens=512)
structured = model.with_structured_output(Ticket)   # returns a Ticket instance

t = structured.invoke("I was charged twice and support hasn't replied in 3 days!")
print(t.category, t.priority)   # billing high

structured.pyfrom pydantic import BaseModel, Field
from typing import Literal
from langchain_openai import ChatOpenAI

class Ticket(BaseModel):
    category: Literal["bug","feature","billing","other"]
    priority: Literal["low","medium","high"]
    summary: str = Field(description="one-line summary")

model = ChatOpenAI(model="gpt-5.5", max_tokens=512)
structured = model.with_structured_output(Ticket)   # returns a Ticket instance

t = structured.invoke("I was charged twice and support hasn't replied in 3 days!")
print(t.category, t.priority)   # billing high

Lab L2.3 · Adding memory intermediate

Here's the uncomfortable truth that surprises everyone building their first chatbot: the model has no memory at all. None. Each API call is a clean slate — the model that answered your last question has already forgotten it existed. If a chatbot seems to "remember" your name, that's an illusion the application maintains, not something the model does. Get this wrong and you'll ship an assistant that cheerfully asks "What's your name?" three turns after you told it, and you'll waste an afternoon hunting for a bug that isn't in your code — it's in your mental model.

So how does any chatbot remember anything? By cheating, honestly: it keeps a transcript on the side and resends the whole conversation with every new message. The model looks omniscient because on turn five it's handed turns one through four all over again, as if reading a script from the top each time. Picture an actor with total amnesia who nonetheless nails every scene — because a stagehand hands them the entire play-so-far before each line. The model is the amnesiac actor; the memory object is the stagehand.

Doing that stagehand work by hand is tedious and error-prone: you'd maintain a list, append each user turn and each reply, splice it into the prompt in the right spot, and keep separate lists for separate users. LangChain's memory Runnables automate exactly this — you tag each conversation with a session_id, and the wrapper loads that session's history into a placeholder before the call and appends the new exchange after it. The common mistake is to then believe memory is something clever or persistent: it is neither. It is the resend-the-list pattern with a tidy interface, which is why (as the note after the lab warns) it's unbounded, grows your token bill every turn, and vanishes on restart unless you back it with a real store. Now the precise version: the API is stateless (C2) — every call resends the history — and LangChain wraps that pattern so a chain "remembers" a conversation by session id, without you managing the list by hand.

memory_chain.pyfrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_anthropic import ChatAnthropic

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a concise assistant."),
    MessagesPlaceholder("history"),        # prior turns slot in here
    ("human", "{input}"),
])
chain = prompt | ChatAnthropic(model="claude-opus-4-8", max_tokens=512)

store = {}
def get_history(session_id):
    return store.setdefault(session_id, InMemoryChatMessageHistory())

chat = RunnableWithMessageHistory(chain, get_history,
    input_messages_key="input", history_messages_key="history")

cfg = {"configurable": {"session_id": "user-1"}}
chat.invoke({"input": "My name is Alice."}, config=cfg)
print(chat.invoke({"input": "What's my name?"}, config=cfg))  # "Alice"

memory_chain.pyfrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a concise assistant."),
    MessagesPlaceholder("history"),        # prior turns slot in here
    ("human", "{input}"),
])
chain = prompt | ChatOpenAI(model="gpt-5.5", max_tokens=512)

store = {}
def get_history(session_id):
    return store.setdefault(session_id, InMemoryChatMessageHistory())

chat = RunnableWithMessageHistory(chain, get_history,
    input_messages_key="input", history_messages_key="history")

cfg = {"configurable": {"session_id": "user-1"}}
chat.invoke({"input": "My name is Alice."}, config=cfg)
print(chat.invoke({"input": "What's my name?"}, config=cfg))  # "Alice"

Lab L2.4 · RAG, composed intermediate

This is the lab that pays off the whole chapter. Chapter 3 had you build Retrieval-Augmented Generation the long way — chunk the documents, embed them, store the vectors, retrieve the closest ones, stuff them into a prompt, generate. That was a page of careful code. Watch what happens when each of those stages is already a Runnable: the entire pipeline collapses into a handful of piped lines. If the "one idea" ever felt abstract, this is where it turns into leverage you can feel.

Why does RAG exist at all? Because a model only knows what it was trained on — it has never seen your refund policy, your runbooks, last week's incident notes. Ask it directly and it will either admit ignorance or, worse, confidently invent a plausible-sounding answer. RAG fixes this by doing what a good research assistant does: look it up first, then answer from what you found. Fetch the relevant passages from your own documents, hand them to the model as context, and instruct it to answer only from that. The model stops being an oracle and becomes a skilled paraphraser of facts you supplied.

The one new wrinkle here is that the chain doesn't start with a single value flowing down a line — it starts by doing two things at once. It needs the retrieved context and the original question, side by side, to fill the prompt. LCEL expresses that with a dictionary of Runnables ({"context": retriever | format_docs, "question": RunnablePassthrough()}): a parallel step where each key runs its own branch — one fetches and formats documents, the other forwards the question untouched — and the results merge into a dict the prompt consumes. So pipes aren't only for straight lines; a dict runs branches in parallel just as naturally.

The mistake this section wants to inoculate you against is blaming the framework when answers are bad. Wrapping RAG in LCEL changes none of the retrieval physics: if your chunks are too big, or your top-k grabs the wrong passages, the model is grounded in garbage and answers accordingly — same failure modes as Chapter 3, same fixes. The framework assembles the pipeline elegantly; it does not make the retrieval smart. Now the precise version: Chapter 3 built RAG from scratch (chunk, embed, store, retrieve, generate), and LangChain gives you each stage as a component that LCEL wires into one chain.

rag_chain.pyfrom langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_anthropic import ChatAnthropic

# retriever = vectorstore.as_retriever()  # built from your embedded docs (Ch 3 / A7)

prompt = ChatPromptTemplate.from_template(
    "Answer using ONLY this context. If it's not there, say you don't know.\n\n"
    "Context:\n{context}\n\nQuestion: {question}"
)
model = ChatAnthropic(model="claude-opus-4-8", max_tokens=512)

def format_docs(docs): return "\n\n".join(d.page_content for d in docs)

rag = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt | model | StrOutputParser()
)
print(rag.invoke("What is our refund window?"))

rag_chain.pyfrom langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# retriever = vectorstore.as_retriever()  # built from your embedded docs (Ch 3 / A7)

prompt = ChatPromptTemplate.from_template(
    "Answer using ONLY this context. If it's not there, say you don't know.\n\n"
    "Context:\n{context}\n\nQuestion: {question}"
)
model = ChatOpenAI(model="gpt-5.5", max_tokens=512)

def format_docs(docs): return "\n\n".join(d.page_content for d in docs)

rag = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt | model | StrOutputParser()
)
print(rag.invoke("What is our refund window?"))

The LangChain ecosystem, briefly intermediate

"LangChain" is one word for what is really a small family of packages, and confusing them is a common early stumble. People install langchain, follow a tutorial that imports from langchain-core, hit a class that lives in langchain-anthropic, and conclude the whole thing is broken. It isn't — it's just that the project split into focused pieces so you only pull in what you use, and knowing which piece owns what saves you a lot of confused import errors.

The mental model is a kitchen. langchain-core is the countertop and the standard fittings — the Runnable/LCEL primitives every recipe shares. The provider packages (langchain-anthropic and its siblings) are appliances you plug into those fittings: swap the appliance, keep the counter, and your recipes barely change — that's how you switch model vendors by changing one import. Retrievers and vector stores are the pantry (your RAG components from Chapter 3 and A7). And two bigger tools sit alongside for when a straight chain isn't enough: LangGraph for stateful, looping agent graphs (L4/L5), and LangSmith for tracing and evaluation across everything you build. The table below is your map of which piece is which.

Common pitfalls advanced

Almost every LangChain headache a beginner hits is one of a handful of predictable mistakes — and you've already met most of them in this chapter. The framework's smooth surface hides the fact that it's still just prompts, tokens, and retrieval underneath, so the traps aren't exotic; they're the same old LLM realities wearing a new coat. Reading this list before you hit them turns a lost afternoon into an "oh, right, I was warned about that."

Notice the theme running through the table: nearly every fix is "the framework didn't change the underlying reality, so treat it as you would raw." Old tutorials break because packages get reorganized (verify your imports). Memory balloons because it's still resending the transcript (trim it). Bad RAG answers are still a chunking-and-top-k problem, not a LangChain problem. And the two judgment calls at the bottom — reaching for a chain when you actually need runtime decisions, or wrapping a one-off script in a framework it doesn't need — are really the same lesson stated twice: pick the simplest tool that fits the job. A chain is a fixed pipeline; when the path must be decided on the fly you want an agent (L3) or a LangGraph graph (L4), and for a single call the raw SDK (C2) is often the clearer choice.

Exercises advanced

class Runnable:
    def __init__(self, fn):
        self.fn = fn
    def invoke(self, x):
        return self.fn(x)
    def __or__(self, other):        # self | other
        return Runnable(lambda x: other.invoke(self.invoke(x)))

prompt = Runnable(lambda topic: f"Write about {topic}.")
model  = Runnable(lambda p: f"[answer to] {p}")
parser = Runnable(lambda a: a.upper())

chain = prompt | model | parser
print(chain.invoke("cats"))   # [ANSWER TO] WRITE ABOUT CATS.

import re
from dataclasses import dataclass

@dataclass
class Person:
    name: str
    age: int

def parse_person(text):
    m = re.search(r"name=(\w+)\s+age=(\d+)", text)
    if not m:
        raise ValueError(f"cannot parse: {text!r}")
    return Person(name=m.group(1), age=int(m.group(2)))

print(parse_person("name=Ada age=36"))   # Person(name='Ada', age=36)

class Memory:
    def __init__(self):
        self.turns = []
    def load(self):
        return "\n".join(f"{r}: {t}" for r, t in self.turns)
    def save(self, user, ai):
        self.turns += [("user", user), ("ai", ai)]

mem = Memory()
def chat(user_msg):
    prompt = f"History:\n{mem.load()}\nUser: {user_msg}"
    reply = f"(reply to '{user_msg}')"      # stand-in for the model
    mem.save(user_msg, reply)
    return reply

chat("my name is Ada")
print(chat("what's my name?"))
print("---history---"); print(mem.load())

DOCS = [
    "Refunds are processed within 30 days.",
    "Support hours are 9am to 5pm.",
    "Passwords reset under Settings > Security.",
]
def retrieve(query, k=1):
    # toy relevance: overlap of query words with the doc
    q = set(query.lower().split())
    scored = sorted(DOCS, key=lambda d: len(q & set(d.lower().split())), reverse=True)
    return scored[:k]

def rag(query):
    ctx = "\n".join(retrieve(query))
    prompt = f"Context:\n{ctx}\nQ: {query}\nAnswer from context only."
    return f"[grounded answer] {retrieve(query)[0]}"   # stand-in model

print(rag("how long for a refund?"))   # cites the 30-day doc

class Runnable:
    def __init__(self, fn):
        self.fn = fn
    def invoke(self, x):
        return self.fn(x)
    def batch(self, xs):
        return [self.fn(x) for x in xs]
    def stream(self, x):
        for word in self.fn(x).split():
            yield word            # emit chunk by chunk

r = Runnable(lambda t: f"answer for {t} here")
print(r.invoke("a"))
print(r.batch(["a", "b"]))
print(list(r.stream("a")))       # ['answer', 'for', 'a', 'here']

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_anthropic import ChatAnthropic      # Claude
# from langchain_openai import ChatOpenAI          # OpenAI

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a concise assistant."),
    ("human", "Explain {topic} in one sentence."),
])
model  = ChatAnthropic(model="claude-opus-4-8", max_tokens=256)   # Claude
# model = ChatOpenAI(model="gpt-5.5", max_tokens=256)             # OpenAI
parser = StrOutputParser()

chain = prompt | model | parser          # the LCEL pipe, for real
print(chain.invoke({"topic": "prompt caching"}))
# batch and stream come for free:
# chain.batch([{"topic": "RAG"}, {"topic": "agents"}])
# for chunk in chain.stream({"topic": "LCEL"}): print(chunk, end="")

Knowledge check check yourself