One interface: the Runnable

LangChain looks sprawling, but almost everything — prompts, models, parsers, retrievers, whole chains, agents — implements one interface: the Runnable. If something is a Runnable, you can:

• .invoke(input) — run it once on one input.
• .stream(input) — run it and yield output incrementally (token streaming).
• .batch(inputs) — run it over a list, concurrently.
• compose it with | into a bigger Runnable that also has all three methods.

That uniformity is the payoff: learn the interface once and every component — and every composition of components — behaves the same way, with streaming, batching, and async handled for you.

LCEL: composing Runnables

LCEL (LangChain Expression Language) is just composing Runnables with |. A prompt-model-parser chain reads as a data flow:

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

prompt = ChatPromptTemplate.from_template("Explain {topic} in one sentence.")   # a Runnable: fills the template
model = ChatOpenAI(model="gpt-4o-mini")                                          # a Runnable: calls the LLM
parser = StrOutputParser()                                                      # a Runnable: message -> plain string

chain = prompt | model | parser        # compose into ONE Runnable (prompt's output feeds model, then parser)

chain.invoke({"topic": "embeddings"})  # run once -> "Embeddings are numeric vectors that capture meaning."
for chunk in chain.stream({"topic": "RAG"}):   # stream tokens as they arrive (the whole chain streams for free)
    print(chunk, end="")
chain.batch([{"topic": "agents"}, {"topic": "MCP"}])   # run over a list concurrently

Because the whole chain is itself a Runnable, .stream and .batch work on the composition without any extra code — that's the ergonomic win over hand-rolling.

Parallel and passthrough

Two Runnables you'll use constantly for shaping data flow:

• RunnableParallel (written as a plain dict) — run several Runnables on the same input and collect a dict of results. This is how you assemble a prompt's variables from multiple sources at once.
• RunnablePassthrough — forward the input unchanged (optionally adding fields). It's the "keep the original question while I also compute context" piece — exactly the RAG shape from the Frameworks topic.

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

# Run `retriever` and pass the raw question through, both from the same input, into one dict:
setup = RunnableParallel(
    context=retriever,                       # retriever(input) -> chunks  (runs on the input string)
    question=RunnablePassthrough(),          # forward the input string unchanged
)
rag_chain = setup | prompt | model | parser  # {context, question} fills the prompt, then generate, then parse
rag_chain.invoke("What does the Pro plan cost?")   # the full RAG chain as one Runnable

Memory: remembering the conversation

A bare chain is stateless — each invoke is independent. To make it conversational you wrap it so prior messages are stored per session and re-injected on each turn. LangChain's RunnableWithMessageHistory does this: you give it a function that returns a history store for a session id, and it prepends the history automatically.

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory

store = {}                                             # session_id -> history object (use a DB in production)
def get_history(session_id: str):
    if session_id not in store:                        # first time we see this session…
        store[session_id] = InMemoryChatMessageHistory()   # …create an empty history
    return store[session_id]                           # return the per-session message log

chat = RunnableWithMessageHistory(chain, get_history)  # wrap the chain so it auto-loads + appends history

# The `configurable` session_id selects WHICH conversation this call belongs to:
chat.invoke({"topic": "agents"}, config={"configurable": {"session_id": "user-42"}})
chat.invoke({"topic": "and how do they differ from chains?"},           # this turn 'remembers' the previous one
            config={"configurable": {"session_id": "user-42"}})

A tool-calling agent in LangChain

Everything you hand-rolled in Topics 1–2 — the reason→act→observe loop with function calling — LangChain packages as an agent. You bind tools to a model, wrap it in an executor that runs the loop (with the bound max_iterations, error handling, and tracing you'd otherwise write yourself), and invoke it:

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Get today's weather for a city."""       # the docstring BECOMES the tool description the model reads
    return f"Sunny, 21°C in {city}"             # (real impl would call a weather API)

# Bind tools + wrap in an executor that runs the tool-call loop for you:
agent = create_tool_calling_agent(model, [get_weather], prompt)      # builds the reason/act step
executor = AgentExecutor(agent=agent, tools=[get_weather],
                         max_iterations=5, handle_parsing_errors=True)   # the bounded, error-handled loop
executor.invoke({"input": "Should I bring an umbrella in Paris?"})   # runs the whole agent to a final answer

Notice what you didn't write: the loop, the tool_call_id plumbing, the step bound, streaming, and the trace. That's the trade from the Frameworks topic in miniature — less control, far less boilerplate — and the @tool docstring is the same "descriptions are prompts" rule from Topic 1.

Pitfalls

• Chasing the API, not the model. Imports churn; the Runnable/LCEL/agent concepts don't. Anchor on those and check current docs for signatures.
• In-memory history in production. It's per-process and volatile — use a backed store and trim/summarize long histories.
• Unbounded agents. Always set max_iterations (and handle parsing errors) — the same loop-bound discipline as the hand-rolled version.
• Over-abstracting simple flows. For a one-shot prompt, prompt | model | parser is plenty; you don't need an agent. (The chain-vs-agent judgment from Topic 1 still rules.)
• Weak @tool docstrings. The docstring is the tool description the model routes on — write it carefully.

Recap

• LangChain's unifying abstraction is the Runnable (invoke/stream/batch); composing Runnables with | (LCEL) yields a new Runnable with all three methods for free.
• RunnableParallel (a dict) fans out over one input; RunnablePassthrough forwards it — together they build the RAG-style {context, question} shape.
• RunnableWithMessageHistory adds per-session memory; use a backed store and trim long histories in production.
• A LangChain agent (create_tool_calling_agent + AgentExecutor) packages the reason→act→observe loop with bounds, error handling, and tracing — your hand-rolled loop, productionized.
• Worked examples map to the exercises: LCEL chain (Ex 1), parallel/passthrough (Ex 2), message-history memory (Ex 3), a tool-calling AgentExecutor (Ex 4).

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

chain = (ChatPromptTemplate.from_template("Explain {topic} in one sentence.")
         | ChatOpenAI(model="gpt-4o-mini")
         | StrOutputParser())

print(chain.invoke({"topic": "embeddings"}))
for tok in chain.stream({"topic": "RAG"}):
    print(tok, end="", flush=True)
print(chain.batch([{"topic": "agents"}, {"topic": "MCP"}, {"topic": "LCEL"}]))

from datetime import datetime
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

setup = RunnableParallel(
    context=retriever,                                  # retriever(question) -> chunks
    question=RunnablePassthrough(),                     # forward the question unchanged
    now=lambda _: datetime.now().isoformat(),          # any callable is a Runnable too
)
chain = setup | prompt | model | parser                # prompt uses {context}, {question}, {now}
chain.invoke("What does the Pro plan cost?")

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory

store = {}
def get_history(sid):
    store.setdefault(sid, InMemoryChatMessageHistory())
    return store[sid]

chat = RunnableWithMessageHistory(chain, get_history)
cfg = {"configurable": {"session_id": "u1"}}
print(chat.invoke({"input": "What are agents?"}, config=cfg))
print(chat.invoke({"input": "And how do those differ from chains?"}, config=cfg))  # 'those' = agents, remembered

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Get today's weather for a city."""
    return f"Rainy, 12°C in {city}"

@tool
def calculator(expression: str) -> str:
    """Evaluate an arithmetic expression."""
    return str(eval(expression))

agent = create_tool_calling_agent(model, [get_weather, calculator], prompt)
executor = AgentExecutor(agent=agent, tools=[get_weather, calculator],
                         max_iterations=5, handle_parsing_errors=True, verbose=True)
print(executor.invoke({"input": "If it's raining in Paris, what's 12 * 7?"}))