#pip install langchain langchain-core langgraph numpyLangChain and Retrieval-Augmented Generation
Orchestration, memory, and retrieval, built against the current LangChain API
1 Introduction
A language model on its own answers from what it learned during training: it has no access to a specific organization’s documents, no memory of earlier turns in a conversation, and no way to call an external tool. LangChain is an open-source framework for closing that gap — it provides standard interfaces for connecting a model to external data, giving it persistent conversational state, and composing several model calls into a single workflow.
This notebook covers LangChain’s core abstractions and Retrieval-Augmented Generation (RAG) computationally, with every code example actually executed rather than described. Two constraints shape how that’s done. First, LangChain moved fast between 2023 and 2025: the framework reached its 1.0 release in October 2025, and several APIs commonly shown in older tutorials — LLMChain, SimpleSequentialChain, SequentialChain, and the classic ConversationBufferMemory family — have since been removed outright, not merely deprecated. This is verified directly below rather than asserted. Second, running real chains against a hosted model requires an API key and network access to that provider, neither of which is assumed here; instead, LangChain’s own deterministic fake models (FakeListLLM, FakeListChatModel) are used where a model call is needed, and a small self-contained retrieval pipeline is built by hand for the RAG section. This keeps every output in this notebook reproducible without external credentials, while the mechanics demonstrated — prompt composition, chaining, memory, and retrieval — are identical to what a real provider integration would do.
2 Checking the Current LangChain API Before Using It
Before building anything, it’s worth confirming what’s actually available in the installed version, since this determines which patterns below are current and which older material (including common tutorials) would fail outright.
import langchain
import langchain_core
print("langchain version: ", langchain.__version__)
print("langchain_core version:", langchain_core.__version__)
# The pre-2024 chain and memory classes commonly shown in tutorials
for module_name, names in [
("langchain.chains", ["LLMChain", "SimpleSequentialChain", "SequentialChain"]),
("langchain.memory", ["ConversationBufferMemory", "ConversationSummaryMemory"]),
]:
try:
mod = __import__(module_name, fromlist=names)
print(f"{module_name}: importable")
except ModuleNotFoundError as e:
print(f"{module_name}: NOT importable -> {e}")langchain version: 1.3.14
langchain_core version: 1.5.3
langchain.chains: NOT importable -> No module named 'langchain.chains'
langchain.memory: NOT importable -> No module named 'langchain.memory'Both langchain.chains and langchain.memory fail to import in the installed version — these modules were removed, not just deprecated. LLMChain was marked deprecated in LangChain 0.1.17 (April 2024) with removal scheduled for 1.0; ConversationBufferMemory and the related memory classes followed the same path starting in 0.3.1. The 1.0 release (October 2025) carried out that removal. Any code written against these classes — including the pattern shown in older training material on this topic — needs to be rewritten against the current API, which is what the rest of this notebook does.
3 Current Architecture
LangChain’s package layout reflects a deliberate split introduced as the framework matured:
langchain-core— the foundational abstractions: theRunnableinterface, prompt templates, output parsers, and the LangChain Expression Language (LCEL), which is the|(pipe) syntax used to compose components.langchain— higher-level, pre-built chains and utilities built on top oflangchain-core.langchain-communityand provider packages (langchain-openai,langchain-google-genai, and so on) — integrations with specific model providers, vector stores, and document loaders, kept as optional dependencies so a project only installs what it needs.- LangGraph — a separate, closely integrated framework for stateful, multi-step agent workflows, now the recommended way to build anything involving persistent memory or multi-turn tool use.
Every composable piece in this architecture — a prompt, a model, a retriever, an output parser — implements the same Runnable interface, which is why they can all be connected with the same | operator regardless of type. This single unifying abstraction has effectively replaced the older “six components” framing (Model I/O, Data Connection, Chains, Agents, Memory, Callbacks) used in pre-2024 documentation of this framework; the underlying concerns — models, prompts, retrieval, memory, and tool-using agents — are all still present, just organized around Runnables and LCEL rather than as six separate categories.
4 Chains
A chain connects a sequence of steps — prompt formatting, a model call, output parsing, possibly a retrieval step — into a single callable pipeline, so an application doesn’t have to manually manage passing state from one step to the next. In the current API, “building a chain” means composing Runnable objects with the | operator; there is no separate Chain class to instantiate for the common cases.
4.1 The simplest chain: prompt to model to parser
The most basic pattern takes a prompt template, fills it with a variable, sends it to a model, and extracts the text from the response. Below, FakeListLLM stands in for a real provider – it returns a fixed, pre-specified response regardless of input, which is exactly what makes this reproducible without an API key.
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.language_models.fake import FakeListLLM
llm = FakeListLLM(responses=["GenomeCraft Analytics"])
prompt = PromptTemplate.from_template(
"What is a good name for a company that makes {product}?"
)
chain = prompt | llm | StrOutputParser()
result = chain.invoke({"product": "genomics data pipelines"})
print(result)GenomeCraft Analytics4.2 Sequential composition
Where an older API would use SimpleSequentialChain to feed the output of one chain into the next, the current pattern uses RunnablePassthrough.assign() to build a dictionary that carries both the original input and each intermediate result forward.
from langchain_core.runnables import RunnablePassthrough
llm_name = FakeListLLM(responses=["GenomeCraft Analytics"])
llm_description = FakeListLLM(responses=[
"GenomeCraft Analytics builds reproducible pipelines for large-scale genomic data processing and interpretation."
])
name_prompt = PromptTemplate.from_template(
"What is a good name for a company that makes {product}?"
)
description_prompt = PromptTemplate.from_template(
"Write a 20-word description for the company: {company_name}"
)
name_chain = name_prompt | llm_name | StrOutputParser()
description_chain = description_prompt | llm_description | StrOutputParser()
overall_chain = (
{"company_name": name_chain}
| RunnablePassthrough.assign(description=description_chain)
)
result = overall_chain.invoke({"product": "genomics data pipelines"})
result{'company_name': 'GenomeCraft Analytics',
'description': 'GenomeCraft Analytics builds reproducible pipelines for large-scale genomic data processing and interpretation.'}4.3 Multiple inputs and outputs
The older SequentialChain class was used when a workflow needed more than one input or output variable, and access to every intermediate result rather than just the final one. The equivalent LCEL pattern nests RunnablePassthrough.assign() calls, so each new field is added to a running dictionary without discarding the earlier ones – this reproduces a four-step translate to summarize to detect-language to follow-up workflow, with all four outputs available at the end.
translate_llm = FakeListLLM(responses=[
"The GamersTech laptops impress with their exceptional performance and elegant design."
])
summarize_llm = FakeListLLM(responses=[
"GamersTech laptops balance strong gaming performance with a sleek, portable design."
])
language_llm = FakeListLLM(responses=["French"])
followup_llm = FakeListLLM(responses=[
"Merci pour ce retour tres positif sur nos ordinateurs portables GamersTech."
])
review = (
"Les ordinateurs portables GamersTech impressionnent par ses performances "
"exceptionnelles et son design elegant."
)
translate_chain = (
PromptTemplate.from_template("Translate the following review to English:\n\n{review}")
| translate_llm | StrOutputParser()
)
summarize_chain = (
PromptTemplate.from_template("Summarize the following review in one sentence:\n\n{english_review}")
| summarize_llm | StrOutputParser()
)
language_chain = (
PromptTemplate.from_template("What language is the following review written in?\n\n{review}")
| language_llm | StrOutputParser()
)
followup_chain = (
PromptTemplate.from_template(
"Write a follow-up response to this summary, in the specified language.\n\n"
"Summary: {summary}\nLanguage: {language}"
)
| followup_llm | StrOutputParser()
)
pipeline = (
RunnablePassthrough.assign(english_review=translate_chain)
| RunnablePassthrough.assign(summary=summarize_chain, language=language_chain)
| RunnablePassthrough.assign(followup_message=followup_chain)
)
output = pipeline.invoke({"review": review})
for key, value in output.items():
print(f"{key}:\n {value}\n")review:
Les ordinateurs portables GamersTech impressionnent par ses performances exceptionnelles et son design elegant.
english_review:
The GamersTech laptops impress with their exceptional performance and elegant design.
summary:
GamersTech laptops balance strong gaming performance with a sleek, portable design.
language:
French
followup_message:
Merci pour ce retour tres positif sur nos ordinateurs portables GamersTech.
4.4 Routing between sub-chains
A router chain sends an input to one of several specialized sub-chains depending on what kind of input it is – for instance, routing a support query to a billing-specific prompt or a technical-specific prompt. The current equivalent is RunnableBranch, which evaluates a sequence of (condition, chain) pairs and runs the first one that matches.
from langchain_core.runnables import RunnableBranch
billing_llm = FakeListLLM(responses=["This looks like a billing question -- routing to the billing team."])
technical_llm = FakeListLLM(responses=["This looks like a technical question -- routing to the engineering team."])
general_llm = FakeListLLM(responses=["Routing to general support."])
billing_chain = PromptTemplate.from_template("{query}") | billing_llm | StrOutputParser()
technical_chain = PromptTemplate.from_template("{query}") | technical_llm | StrOutputParser()
general_chain = PromptTemplate.from_template("{query}") | general_llm | StrOutputParser()
router = RunnableBranch(
(lambda x: "invoice" in x["query"].lower() or "charge" in x["query"].lower(), billing_chain),
(lambda x: "error" in x["query"].lower() or "crash" in x["query"].lower(), technical_chain),
general_chain, # default
)
for query in ["Why was I charged twice on my invoice?",
"The pipeline crashes with a segmentation fault.",
"What are your support hours?"]:
print(query, "->", router.invoke({"query": query}))Why was I charged twice on my invoice? -> This looks like a billing question -- routing to the billing team.
The pipeline crashes with a segmentation fault. -> This looks like a technical question -- routing to the engineering team.
What are your support hours? -> Routing to general support.5 Memory
A language model call is stateless by default: nothing from one call is automatically available to the next unless it’s explicitly included in the prompt. “Memory” in this context means a mechanism for storing prior turns and re-injecting them, so a conversation feels continuous rather than resetting at every message.
Four memory strategies recur across older LangChain material, distinguished by what they store:
- Buffer memory – stores the full conversation verbatim.
- Buffer window memory – stores only the most recent k exchanges, discarding older ones.
- Token buffer memory – keeps as much recent conversation as fits within a token budget, verbatim.
- Summary memory – replaces older turns with a running summary, generated by the model itself.
Each represents a different trade-off between fidelity (how much detail is preserved) and cost (how many tokens are spent re-sending history on every call). All four are implemented as classes in langchain.memory in older versions – a module that, as confirmed above, no longer exists in the installed version. The current recommended approach uses LangGraph’s checkpointing system, which persists conversation state outside the chain itself and supports the buffer/window/summary trade-off through how the state is read back rather than through separate classes for each strategy.
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.language_models.fake_chat_models import FakeListChatModel
from langchain_core.messages import HumanMessage
chat_model = FakeListChatModel(responses=[
"Hello! It's nice to meet you.",
"Your name is Sarah -- you told me a moment ago.",
])
def call_model(state: MessagesState):
response = chat_model.invoke(state["messages"])
return {"messages": [response]}
builder = StateGraph(MessagesState)
builder.add_node("model", call_model)
builder.add_edge(START, "model")
builder.add_edge("model", END)
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "session-1"}}
turn_1 = graph.invoke({"messages": [HumanMessage("Hi, my name is Sarah.")]}, config=config)
print("Turn 1:", turn_1["messages"][-1].content)
turn_2 = graph.invoke({"messages": [HumanMessage("What is my name?")]}, config=config)
print("Turn 2:", turn_2["messages"][-1].content)
persisted_state = graph.get_state(config)
print("\nMessages held in the checkpoint:", len(persisted_state.values["messages"]))Turn 1: Hello! It's nice to meet you.
Turn 2: Your name is Sarah -- you told me a moment ago.
Messages held in the checkpoint: 4The second call correctly answers “Sarah” because the full message history was retrieved from the checkpointer and passed back into the model, not because the fake model has any memory of its own – FakeListChatModel simply returns its next scripted response regardless of input. The thread_id in the config is what scopes the stored history to this particular conversation; a different thread_id would start from empty state.
6 Retrieval-Augmented Generation (RAG)
A model’s knowledge is fixed at training time. RAG addresses this by inserting a retrieval step before generation: given a query, first find the most relevant passages from an external document collection, then pass both the query and those passages to the model so its answer is grounded in retrieved material rather than in training data alone.
The pipeline has two stages:
- Retriever – converts the query into a vector and finds the most similar vectors in a pre-built index of document chunks.
- Generator – a language model that conditions its answer on both the original query and the retrieved chunks.
This addresses three specific weaknesses of using a language model alone: the model’s knowledge can be extended to include material it was never trained on (private documents, content newer than its training cutoff), answers can be traced back to a specific retrieved source, and the retrieval step reduces (though does not eliminate) the tendency to generate a plausible-sounding but unsupported answer.
6.1 Common applications
RAG is the standard architecture behind several application categories: chatbots that answer from internal company documents, legal or financial Q&A tools that summarize relevant clauses, research assistants that retrieve from academic or clinical literature, and support agents that answer from product documentation.
7 Vector Databases
A vector database is built to store and search high-dimensional numeric vectors – the embeddings produced by an embedding model – rather than the exact-match or range queries a conventional database is optimized for. Retrieval works by approximate nearest-neighbor search: given a query vector, find the stored vectors closest to it by some distance measure, typically cosine similarity.
Several tools recur in this space, each with a different operating model rather than a strict quality ranking: Chroma (lightweight, easy to run locally, common for prototyping), FAISS (a similarity-search library rather than a managed database, built for large-scale offline or self-hosted use), Pinecone and Qdrant (managed or self-hostable services built for production-scale, low-latency search), Weaviate (a vector-native database with built-in classification and hybrid search), and Redis (a general-purpose store that added vector search as a capability rather than being purpose-built for it). Which one fits a given project depends on scale, latency requirements, and whether self-hosting or a managed service is preferred – not on one being categorically “better” than the others.
8 Essential RAG Components, Built by Hand
The remaining pieces of a RAG pipeline – chunking, embedding, and similarity search – can be demonstrated directly without a hosted embedding model, using term-frequency vectors instead of dense neural embeddings. The mechanics (vectorize each chunk, vectorize the query the same way, rank by cosine similarity) are identical to what a real embedding model does; only the quality of the vectors themselves differs.
8.1 Chunking
Long documents are split into smaller pieces before embedding, because a single embedding vector for an entire document would blur together many different topics, making retrieval far less precise. Here, each chunk is already a short single-topic sentence, so this step is trivial – in practice a chunker would split a long document into passages of roughly a few hundred words each, often with a small overlap between consecutive chunks so a sentence spanning a chunk boundary isn’t lost entirely from either side.
document_chunks = [
"GWAS identifies associations between genetic variants and traits across the genome.",
"Fine-mapping narrows a GWAS locus down to the most likely causal variant.",
"Polygenic risk scores aggregate many small-effect variants into a single predictive score.",
"Retrieval-augmented generation grounds a language model's answer in retrieved documents.",
"Vector databases store embeddings and support fast approximate nearest neighbor search.",
]
document_chunks['GWAS identifies associations between genetic variants and traits across the genome.',
'Fine-mapping narrows a GWAS locus down to the most likely causal variant.',
'Polygenic risk scores aggregate many small-effect variants into a single predictive score.',
"Retrieval-augmented generation grounds a language model's answer in retrieved documents.",
'Vector databases store embeddings and support fast approximate nearest neighbor search.']8.2 A minimal embedding model
Each chunk is converted to a vector using term-frequency counts weighted by inverse document frequency (TF-IDF) – common words shared across every chunk contribute little to the vector, while words distinctive to a particular chunk dominate it. This is a much cruder representation than a trained neural embedding model (it has no notion of synonyms or word order), but the retrieval mechanism built on top of it – cosine similarity ranking – is exactly the same mechanism a production vector database uses on dense embeddings.
import numpy as np
import re
from collections import Counter
STOPWORDS = {"a", "an", "the", "is", "are", "in", "on", "of", "to", "and", "or",
"with", "how", "does", "do", "use", "s", "its", "this", "that",
"for", "by", "into", "across"}
def tokenize(text):
tokens = re.findall(r"[a-z0-9]+", text.lower())
return [t for t in tokens if t not in STOPWORDS]
def build_vocabulary(chunks):
vocab = sorted(set(tok for chunk in chunks for tok in tokenize(chunk)))
return {tok: i for i, tok in enumerate(vocab)}
def inverse_document_frequency(chunks, vocab):
n_chunks = len(chunks)
doc_freq = np.zeros(len(vocab))
for chunk in chunks:
for tok in set(tokenize(chunk)):
doc_freq[vocab[tok]] += 1
return np.log((n_chunks + 1) / (doc_freq + 1)) + 1
def embed(text, vocab, idf):
vec = np.zeros(len(vocab))
for tok, count in Counter(tokenize(text)).items():
if tok in vocab:
vec[vocab[tok]] = count * idf[vocab[tok]]
norm = np.linalg.norm(vec)
return vec / norm if norm > 0 else vec
vocabulary = build_vocabulary(document_chunks)
idf = inverse_document_frequency(document_chunks, vocabulary)
chunk_embeddings = [embed(chunk, vocabulary, idf) for chunk in document_chunks]
print(f"Vocabulary size: {len(vocabulary)}")
print(f"Embedding dimension per chunk: {chunk_embeddings[0].shape[0]}")Vocabulary size: 46
Embedding dimension per chunk: 468.3 Similarity search
The query is embedded with the same vocabulary and IDF weights, then ranked against every stored chunk by cosine similarity – since both vectors are already normalized to unit length, cosine similarity reduces to a plain dot product.
def cosine_similarity(a, b):
return float(np.dot(a, b))
def retrieve(query, chunks, chunk_embeddings, vocab, idf, top_k=2):
query_embedding = embed(query, vocab, idf)
scores = [cosine_similarity(query_embedding, emb) for emb in chunk_embeddings]
ranked = sorted(zip(scores, chunks), reverse=True)
return ranked[:top_k]
query = "How does RAG use a vector database to ground an LLM's answer?"
top_matches = retrieve(query, document_chunks, chunk_embeddings, vocabulary, idf, top_k=5)
for score, chunk in top_matches:
print(f"{score:.3f} {chunk}")0.236 Retrieval-augmented generation grounds a language model's answer in retrieved documents.
0.224 Vector databases store embeddings and support fast approximate nearest neighbor search.
0.000 Polygenic risk scores aggregate many small-effect variants into a single predictive score.
0.000 GWAS identifies associations between genetic variants and traits across the genome.
0.000 Fine-mapping narrows a GWAS locus down to the most likely causal variant.The two chunks about RAG and vector databases score well above the three unrelated genetics chunks, which score exactly zero – they share no distinctive vocabulary with the query once stopwords are removed. This is the retrieval step of a RAG pipeline in miniature: the same query-embed-and-rank mechanism, at a scale of five chunks instead of millions, and with TF-IDF vectors standing in for a trained embedding model’s dense vectors.
8.4 Assembling the full pipeline
The retrieved chunks are then handed to a generator, exactly as in the earlier chain examples – a prompt template that includes the retrieved context, piped into a model.
rag_llm = FakeListLLM(responses=[
"RAG grounds an LLM's answer by first retrieving relevant passages -- often from a "
"vector database using similarity search -- and then conditioning generation on both "
"the query and those retrieved passages, rather than relying on the model's training "
"data alone."
])
rag_prompt = PromptTemplate.from_template(
"Answer the question using only the context below.\n\n"
"Context:\n{context}\n\nQuestion: {question}"
)
rag_chain = rag_prompt | rag_llm | StrOutputParser()
retrieved_context = "\n".join(chunk for _, chunk in top_matches[:2])
answer = rag_chain.invoke({"context": retrieved_context, "question": query})
print(answer)RAG grounds an LLM's answer by first retrieving relevant passages -- often from a vector database using similarity search -- and then conditioning generation on both the query and those retrieved passages, rather than relying on the model's training data alone.9 Best Practices
- Build against LCEL, not the legacy
Chainclasses. As demonstrated above,LLMChainand its relatives are gone from the current package, so code written against them fails immediately rather than merely warning. - Choose memory strategy by conversation length and cost, not by default. Full buffer memory is simplest but grows unbounded; window or summary strategies trade some fidelity for a bounded cost.
- Ground factual or domain-specific responses in retrieval. A RAG step reduces (not eliminates) the risk of a confident, unsupported answer.
- Match the vector store to the deployment, not the other way around. A locally-run prototype and a production service under load have different requirements, and the “best” vector database differs accordingly.
- Test each component of a pipeline independently before composing it. A
Runnablecan be invoked and inspected on its own before being piped into a longer chain. - Keep prompts concise and retrieval focused. Retrieving more chunks than the model needs increases cost and can dilute the most relevant context.
- Treat this fast-moving ecosystem as fast-moving. As this notebook’s own deprecation check demonstrates, code and tutorials more than a year or so old are a reasonable place to start but should be verified against the current API before being relied on.
10 Limitations
- Retrieval quality bounds answer quality. A generator conditioned on irrelevant or outdated retrieved chunks will produce a fluent but ungrounded answer regardless of how good the underlying model is.
- Full conversation memory has a cost ceiling. Storing complete history without bound eventually exceeds context limits or becomes prohibitively expensive to resend on every call.
- Composed pipelines are harder to test. A single model call is straightforward to evaluate; a multi-step chain with retrieval, memory, and several model calls has many more places a failure can originate from.
- Latency compounds across steps. Each additional retrieval or model call in a pipeline adds to total response time.
- The framework itself changes quickly. As shown directly above, APIs that were standard as recently as a year or two ago have since been removed, not just superseded – any documentation or tutorial, including this one eventually, is a snapshot rather than a permanent reference.
11 Summary
LangChain provides a common Runnable interface – composed with the | operator via LCEL – that unifies prompts, models, retrievers, and output parsers into a single composition system, and this has superseded the older, separate Chain classes (LLMChain, SimpleSequentialChain, SequentialChain) entirely; this notebook confirmed their removal directly rather than assuming it. Memory has undergone the same shift, from the deprecated ConversationBufferMemory family to LangGraph’s checkpointing system, which persists conversation state outside the chain and was demonstrated here with a minimal two-turn example. Retrieval-Augmented Generation was built end-to-end at small scale – chunking, a hand-built TF-IDF embedding, cosine-similarity retrieval, and a generation step conditioned on the retrieved context – which exercises the same retrieve-then-generate mechanism a production system built on a real embedding model and vector database would use, just with a transparent, dependency-free stand-in for the embedding step itself.
12 References
- Lewis et al., “Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks” (2020) – arXiv:2005.11401
- Karpukhin et al., “Dense Passage Retrieval for Open-Domain Question Answering” (2020) – arXiv:2004.04906
- Vaswani et al., “Attention Is All You Need” (2017) – arXiv:1706.03762
- Gao et al., “Retrieval-Augmented Generation for Large Language Models: A Survey” (2023) – arXiv:2312.10997
- LangChain documentation – docs.langchain.com
A correction to a common citation. Material on this topic sometimes cites arXiv:2301.12652 as “A Survey on Retrieval-Augmented Generation” by Karpukhin et al. (2023). That identifier actually belongs to a different paper (REPLUG, on retrieval-augmented black-box language models), and Karpukhin et al.’s well-known 2020 paper is on dense passage retrieval, not a RAG survey. The reference list above cites the actual survey (Gao et al., 2023) and the correctly attributed Karpukhin et al. paper separately.
13 Try It Yourself
- Modify the router chain’s conditions so a query containing “refund” is also routed to the billing chain, and confirm the routing with a new test query.
- Extend the sequential pipeline (translate to summarize to detect language to follow-up) with a fifth step that scores the summary’s length against a target, reusing the
RunnablePassthrough.assign()pattern. - Add two more document chunks to the RAG retrieval example – one relevant to the existing query, one clearly not – and confirm the ranking places them where expected.
- Look up LangChain’s current documentation for one component used in this notebook (LCEL, LangGraph checkpointers, or
RunnableBranch) and note anything that has changed since this notebook was written.
14 Solutions
Worked solutions to the four exercises above. Each was run against the same installed LangChain version confirmed at the top of this notebook.
14.1 1. Routing “refund” queries to billing
Adding a third or condition to the billing branch’s lambda is enough – no new branch is needed, since a refund question is a billing question.
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.language_models.fake import FakeListLLM
from langchain_core.runnables import RunnableBranch
billing_llm = FakeListLLM(responses=["This looks like a billing question -- routing to the billing team."])
technical_llm = FakeListLLM(responses=["This looks like a technical question -- routing to the engineering team."])
general_llm = FakeListLLM(responses=["Routing to general support."])
billing_chain = PromptTemplate.from_template("{query}") | billing_llm | StrOutputParser()
technical_chain = PromptTemplate.from_template("{query}") | technical_llm | StrOutputParser()
general_chain = PromptTemplate.from_template("{query}") | general_llm | StrOutputParser()
router = RunnableBranch(
(lambda x: "invoice" in x["query"].lower() or "charge" in x["query"].lower() or "refund" in x["query"].lower(), billing_chain),
(lambda x: "error" in x["query"].lower() or "crash" in x["query"].lower(), technical_chain),
general_chain, # default
)
for query in ["Why was I charged twice on my invoice?",
"The pipeline crashes with a segmentation fault.",
"What are your support hours?",
"Can I get a refund for last month's subscription?"]:
print(query, "->", router.invoke({"query": query}))Why was I charged twice on my invoice? -> This looks like a billing question -- routing to the billing team.
The pipeline crashes with a segmentation fault. -> This looks like a technical question -- routing to the engineering team.
What are your support hours? -> Routing to general support.
Can I get a refund for last month's subscription? -> This looks like a billing question -- routing to the billing team.The new query about a refund is correctly routed to the billing chain alongside the existing invoice/charge queries.
14.2 2. Scoring summary length against a target
A fifth RunnablePassthrough.assign() step adds a length_score field, following the exact pattern the rest of the pipeline already uses – the new step reads summary from the running dictionary and reports how far its word count is from a target.
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.language_models.fake import FakeListLLM
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
translate_llm = FakeListLLM(responses=[
"The GamersTech laptops impress with their exceptional performance and elegant design."
])
summarize_llm = FakeListLLM(responses=[
"GamersTech laptops balance strong gaming performance with a sleek, portable design."
])
language_llm = FakeListLLM(responses=["French"])
followup_llm = FakeListLLM(responses=[
"Merci pour ce retour tres positif sur nos ordinateurs portables GamersTech."
])
review = (
"Les ordinateurs portables GamersTech impressionnent par ses performances "
"exceptionnelles et son design elegant."
)
translate_chain = (
PromptTemplate.from_template("Translate the following review to English:\n\n{review}")
| translate_llm | StrOutputParser()
)
summarize_chain = (
PromptTemplate.from_template("Summarize the following review in one sentence:\n\n{english_review}")
| summarize_llm | StrOutputParser()
)
language_chain = (
PromptTemplate.from_template("What language is the following review written in?\n\n{review}")
| language_llm | StrOutputParser()
)
followup_chain = (
PromptTemplate.from_template(
"Write a follow-up response to this summary, in the specified language.\n\n"
"Summary: {summary}\nLanguage: {language}"
)
| followup_llm | StrOutputParser()
)
# New fifth step: score the summary's length against a target word count,
# reusing the RunnablePassthrough.assign() pattern from the rest of the pipeline.
TARGET_WORDS = 12
def score_summary_length(inputs):
word_count = len(inputs["summary"].split())
return {
"word_count": word_count,
"target_words": TARGET_WORDS,
"delta": word_count - TARGET_WORDS,
}
length_score_step = RunnableLambda(score_summary_length)
pipeline = (
RunnablePassthrough.assign(english_review=translate_chain)
| RunnablePassthrough.assign(summary=summarize_chain, language=language_chain)
| RunnablePassthrough.assign(followup_message=followup_chain)
| RunnablePassthrough.assign(length_score=length_score_step)
)
output = pipeline.invoke({"review": review})
for key, value in output.items():
print(f"{key}:\n {value}\n")review:
Les ordinateurs portables GamersTech impressionnent par ses performances exceptionnelles et son design elegant.
english_review:
The GamersTech laptops impress with their exceptional performance and elegant design.
summary:
GamersTech laptops balance strong gaming performance with a sleek, portable design.
language:
French
followup_message:
Merci pour ce retour tres positif sur nos ordinateurs portables GamersTech.
length_score:
{'word_count': 11, 'target_words': 12, 'delta': -1}
The generated summary comes in one word under the 12-word target – delta: -1. All five outputs, including the original input, remain available in the final dictionary, which is the main advantage of the nested-assign() pattern over a plain linear chain.
14.3 3. Adding two more chunks to the retrieval example
One new chunk is written to be relevant to the existing query (about cosine similarity ranking), and one is written to be clearly unrelated (about heritability, matching the theme of the other irrelevant chunks already in the set).
import numpy as np
import re
from collections import Counter
STOPWORDS = {"a", "an", "the", "is", "are", "in", "on", "of", "to", "and", "or",
"with", "how", "does", "do", "use", "s", "its", "this", "that",
"for", "by", "into", "across"}
def tokenize(text):
tokens = re.findall(r"[a-z0-9]+", text.lower())
return [t for t in tokens if t not in STOPWORDS]
def build_vocabulary(chunks):
vocab = sorted(set(tok for chunk in chunks for tok in tokenize(chunk)))
return {tok: i for i, tok in enumerate(vocab)}
def inverse_document_frequency(chunks, vocab):
n_chunks = len(chunks)
doc_freq = np.zeros(len(vocab))
for chunk in chunks:
for tok in set(tokenize(chunk)):
doc_freq[vocab[tok]] += 1
return np.log((n_chunks + 1) / (doc_freq + 1)) + 1
def embed(text, vocab, idf):
vec = np.zeros(len(vocab))
for tok, count in Counter(tokenize(text)).items():
if tok in vocab:
vec[vocab[tok]] = count * idf[vocab[tok]]
norm = np.linalg.norm(vec)
return vec / norm if norm > 0 else vec
def cosine_similarity(a, b):
return float(np.dot(a, b))
def retrieve(query, chunks, chunk_embeddings, vocab, idf, top_k=2):
query_embedding = embed(query, vocab, idf)
scores = [cosine_similarity(query_embedding, emb) for emb in chunk_embeddings]
ranked = sorted(zip(scores, chunks), reverse=True)
return ranked[:top_k]
document_chunks = [
"GWAS identifies associations between genetic variants and traits across the genome.",
"Fine-mapping narrows a GWAS locus down to the most likely causal variant.",
"Polygenic risk scores aggregate many small-effect variants into a single predictive score.",
"Retrieval-augmented generation grounds a language model's answer in retrieved documents.",
"Vector databases store embeddings and support fast approximate nearest neighbor search.",
# New: one relevant to the query below, one clearly not.
"Cosine similarity ranks retrieved chunks by comparing their embedding vectors to the query vector.",
"Heritability partitions phenotypic variance into genetic and environmental components.",
]
vocabulary = build_vocabulary(document_chunks)
idf = inverse_document_frequency(document_chunks, vocabulary)
chunk_embeddings = [embed(chunk, vocabulary, idf) for chunk in document_chunks]
query = "How does RAG use a vector database to ground an LLM's answer?"
top_matches = retrieve(query, document_chunks, chunk_embeddings, vocabulary, idf, top_k=len(document_chunks))
for score, chunk in top_matches:
print(f"{score:.3f} {chunk}")0.261 Retrieval-augmented generation grounds a language model's answer in retrieved documents.
0.170 Vector databases store embeddings and support fast approximate nearest neighbor search.
0.165 Cosine similarity ranks retrieved chunks by comparing their embedding vectors to the query vector.
0.000 Polygenic risk scores aggregate many small-effect variants into a single predictive score.
0.000 Heritability partitions phenotypic variance into genetic and environmental components.
0.000 GWAS identifies associations between genetic variants and traits across the genome.
0.000 Fine-mapping narrows a GWAS locus down to the most likely causal variant.The new relevant chunk lands third, just behind the two chunks it shares vocabulary with (“vector”, “database”), and the new irrelevant chunk scores exactly zero alongside the other genetics chunks, since it shares no non-stopword vocabulary with the query. The ranking places both new chunks exactly where expected.
14.4 4. What’s changed in the current documentation
Checked two components used in this notebook against LangChain’s current reference docs:
RunnableBranchis unchanged. The current reference lists it as available since v0.1 with no deprecation notice, so the routing pattern used above is still the recommended one, not a stopgap.- LangGraph checkpointing has grown a companion concept since this notebook’s core pattern was written. The current persistence docs now distinguish two systems: checkpointers (like
InMemorySaver, used above) for short-term, thread-scoped memory, and a separateStore(for exampleInMemoryStore) for long-term, cross-thread memory such as user preferences or facts that should persist beyond a single conversation thread. The docs also now explicitly flagInMemorySaveras suitable for debugging and testing only, recommendingPostgresSaverfor production use, a distinction not called out above.
For the two-turn example in this notebook, both points are consistent with what’s shown – InMemorySaver is exactly right for a scoped, in-memory demo, and there was no cross-thread memory need that would have called for a Store.