Wire loaders, retrievers, prompts, and model calls into a grounded Q&A flow.
Load PDFs, plain-text files, and web pages into LangChain `Document` objects, then split them into retrieval-sized chunks using `RecursiveCharacterTextSplitter`. You'll wire a multi-source loader pipeline for the chatbot's document corpus.
How to load PDFs, text files, and web pages into LangChain Document objects and split them into retrieval-sized chunks.
Why this matters: This is the first step of your RAG chatbot pipeline — without clean, correctly-sized chunks, every downstream stage (embedding, retrieval, generation) degrades.
Your chatbot can't answer from raw files. Every source — PDF, text file, webpage — must become a object. This struct holds page content plus metadata: source path, page number, URL.
A handles one source type. It reads raw bytes and wraps text in a with metadata. Call .load() and get back a list of Documents.
After loading, a cuts each Document into retrieval-sized . Too-large chunks dilute relevance. Too-small chunks lose context. This boundary is your first design decision.
You're building a support chatbot for a SaaS product. The corpus has three source types: PDF manual, plain-text changelog, and help-center webpage. Each needs a different loader. Output must be one flat list of before embedding.
PyPDFLoader — one Document per page, includes page number metadataTextLoader — one Document for the whole fileWebBaseLoader — one Document per URL, includes source URL metadataAfter loading, RecursiveCharacterTextSplitter runs over all three lists. It splits on paragraph breaks first, then sentences, then words — preserving natural boundaries before falling back to character cuts.
The overlap parameter (10–15% of chunk_size) repeats a tail of each chunk at the start of the next. A sentence straddling a boundary appears in both chunks and won't be lost.
from langchain.document_loaders import PyPDFLoader loader = PyPDFLoader("manual.pdf") docs = loader.load() # Attempt to use the whole doc as one chunk print(docs[0].page_content[:200]) print("Total docs:", len(docs)) print("Total chars in doc 0:", len(docs[0].page_content))
PyPDFLoader("manual.pdf").load()docs[0].page_contentThis loads the PDF correctly but skips splitting — each page becomes one Document, and page 1 alone may be 3 000+ characters. Passing these raw pages to a retriever means every search returns a full page, burying the relevant sentence in noise.
PyPDFLoader returns 50 Documents — one per page. A dense manual page typically runs 2 000–4 000 characters, far above the 512–768 token sweet spot. Passing these unsplit pages to a retriever means the top-K results each contain multiple topics, and the model's grounding suffers. The fix: always split after loading.
from langchain.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter loader = PyPDFLoader("manual.pdf") pages = loader.load() # 50 Documents splitter = RecursiveCharacterTextSplitter( chunk_size=768, chunk_overlap=80, ) chunks = splitter.split_documents(pages) print(f"{len(pages)} pages → {len(chunks)} chunks")
RecursiveCharacterTextSplitter(chunk_size=768, chunk_overlap=80)splitter.split_documents(pages)Adding RecursiveCharacterTextSplitter converts 50 large pages into many smaller, retrieval-sized . Each chunk inherits the source Document's metadata (page number, source path), so you can cite the origin later.
Expect roughly 180–220 chunks. Each 3 000-char page splits into ~4 chunks (3000 / 768 ≈ 3.9), so 50 pages × ~4 = ~200. The exact count varies with paragraph boundaries. Output: '50 pages → 203 chunks' (your number will differ slightly).
from langchain.document_loaders import PyPDFLoader, TextLoader, WebBaseLoader from langchain.text_splitter import RecursiveCharacterTextSplitter loaders = [ PyPDFLoader("manual.pdf"), TextLoader("changelog.txt"), WebBaseLoader("https://help.example.com/faq"), ] raw_docs = [] for ldr in loaders: raw_docs.extend(ldr.load()) splitter = RecursiveCharacterTextSplitter(chunk_size=768, chunk_overlap=80) all_chunks = splitter.split_documents(raw_docs) print(f"Total chunks ready for embedding: {len(all_chunks)}")
TextLoader("changelog.txt")WebBaseLoader("https://...")raw_docs.extend(ldr.load())splitter.split_documents(raw_docs)This is the complete ingestion pipeline: three loaders, one splitter, one flat list of ready to be embedded. The loop pattern scales — add a new loader to the list and the rest of the pipeline is unchanged.
Add PyPDFLoader("appendix.pdf"), to the loaders list (line 5–8). That's the only change — the loop, splitter, and output variable are untouched. Changed line: insert ' PyPDFLoader("appendix.pdf"),' after the WebBaseLoader entry. This is the crux: the pipeline is open for extension without touching the split or output logic.
Drag to see how chunk_size affects retrieval precision and context completeness for your chatbot corpus.
Three failure modes hit most teams in their first pipeline build:
docs[0].page_content == "" — silently. You get the right number of Documents but zero content. Fix: run OCR (e.g. pytesseract) before loading, or use a loader with built-in OCR.chunk_overlap=800, chunk_size=768 throws: ValueError: chunk_overlap must be less than chunk_size. Keep overlap at 10–15% of chunk_size.chunks[0].page_content manually — if it starts with 'Accept cookies', your splitter is ingesting noise that will pollute retrieval.chunks[0].page_content and chunks[-1].page_content — confirm they contain real content, not empty strings or HTML artifacts.chunks[0].metadata — it must include 'source' (file path or URL) so citations work downstream.max_length instead of chunk_size).You now have a flat list of -sized Documents, each carrying source metadata. This is the exact input the next stage expects.
The next module converts each chunk into a dense numeric vector — an — using an embedding model. It writes vectors into a for meaning-based search at query time. Search quality depends on your chunk boundaries.
Convert your split chunks into dense vectors with an embedding model and persist them in a vector store such as Chroma or FAISS. You'll build the indexing half of the chatbot pipeline — the offline step that makes fast similarity search possible at query time.
Convert document chunks into embedding vectors and persist them in a Chroma or FAISS vector store — the offline indexing step that makes fast similarity search possible.
Why this matters: This is the foundation of your chatbot's retrieval system — without a correctly built and persisted index, every query either re-embeds from scratch or returns wrong results.
Decision this forces: Which vector store backend (Chroma for persistence, FAISS for in-memory speed) fits your deployment, and which embedding model matches your query/document length ratio?
Before reading on, predict from memory: what two parameters did you set on RecursiveCharacterTextSplitter in module 1, and why does chunk size matter?
chunk_size controls characters per ; chunk_overlap prevents mid-sentence cuts. Smaller chunks enable precise retrieval. Larger chunks provide more context per result. This tradeoff repeats when choosing an embedding model.You now have objects. The next problem: how does the chatbot find three relevant chunks out of ten thousand in milliseconds?
Convert every chunk into a dense numeric vector — an — and store those vectors in a that searches by geometric closeness.
An maps text to a fixed-length list of floats (typically 768–3072 numbers) where related meanings land geometrically close.
During indexing, every chunk is embedded once and stored. At query time, the user's question is embedded with the same model family. The store runs a (typically cosine similarity) and returns nearest chunk vectors.
Critical constraint: query and documents must use the same embedding model. Mixing models produces meaningless distances. The failure is silent — you get results, just wrong ones.
text-embedding-3-small and BGE variants train on query–passage pairs for this mismatch. Symmetric models (trained on passage–passage pairs) underperform with short queries and long chunks.You've split a 200-page HR handbook into 1,400 chunks. Now make those chunks searchable before the chatbot goes live.
Indexing runs offline once, before any user query. Embed all 1,400 chunks in one batch call (seconds, small API fee). Persist vectors to a Chroma directory on disk.
When the server starts, reload the Chroma store — no re-embedding, no API cost. A user asks "What's the parental leave policy?" The question is embedded on the fly. The store returns top-3 chunks in under 50 ms.
Key insight: embedding is a one-time indexing cost, not a per-query cost. Persisting to disk makes this economics work.
from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma # chunks = list of Document objects from module 1 embedding_model = OpenAIEmbeddings(model="text-embedding-3-small") vector_store = Chroma.from_documents( documents=chunks, embedding=embedding_model, persist_directory="./chroma_db", ) print(f"Indexed {vector_store._collection.count()} chunks")
OpenAIEmbeddings(model=...)Chroma.from_documents(...)persist_directory="./chroma_db"vector_store._collection.count()This creates a Chroma store from your chunk list in one call — it embeds every document and writes vectors to disk at ./chroma_db.
The final print confirms how many vectors were stored — a quick sanity check that no chunks were silently dropped.
Indexed 1400 chunks — the collection count matches the number of Documents you passed. If the number is lower, chunks were silently dropped, usually because some had empty page_content strings.
# On a fresh server start — no re-embedding needed loaded_store = Chroma( persist_directory="./chroma_db", embedding_function=embedding_model, ) results = loaded_store.similarity_search( query="What is the parental leave policy?", k=3, ) for doc in results: print(doc.metadata["source"], doc.page_content[:120])
Chroma(persist_directory=..., embedding_function=...)similarity_search(query, k=3)doc.metadata["source"]Reloading passes the same embedding model so the store can embed incoming queries — but it reads vectors from disk, not from your chunk list.
similarity_search returns the top- chunks by cosine distance; printing the source and a snippet lets you verify the right passages surfaced.
No error. The store will return results, but they'll be semantically random — the query vector lives in a different space than the stored vectors, so cosine distances are meaningless. This is a silent failure: you must catch it by inspecting the returned chunks, not by reading an exception.
from langchain_community.vectorstores import FAISS # Same embedding model and chunks as before faiss_store = FAISS.from_documents( documents=chunks, embedding=embedding_model, ) # TODO: persist the FAISS index to disk so it survives a restart # Hint: FAISS uses a different method than Chroma — check the class API # Reload on next run: reloaded = FAISS.load_local( "faiss_index", embedding_model, allow_dangerous_deserialization=True )
FAISS.from_documents(...)save_local("faiss_index")FAISS.load_local(..., allow_dangerous_deserialization=True)Stop — attempt the TODO before revealing. The missing call is the one FAISS-specific step that Chroma handles automatically.
Hints: the method name mirrors the reload call below it, and it takes a folder path as its only required argument.
faiss_store.save_local("faiss_index") ← this is the changed line.
Why it differs: Chroma persists automatically via persist_directory; FAISS is in-memory by default and requires an explicit save_local() call. If you skip it, the index is lost when the process exits — no error, just gone. The reload call (load_local) is already shown; it mirrors save_local exactly.
| Option | Persistence | Query speed | When to choose | Cost | Complexity |
|---|---|---|---|---|---|
| Chroma | Writes to disk automatically; reload with from_existing(). | Slightly slower than FAISS for large corpora due to SQLite overhead. | You need the index to survive restarts and want to reload without re-embedding — the default for a production chatbot. | Disk I/O on write; fast reads from SQLite-backed store. | Low — one extra persist() call; runs embedded or as a server. |
| FAISS | In-memory by default; must call save_local() explicitly or the index is lost on restart. | Fastest option for pure vector search; uses optimised ANN indexes. | You need maximum in-memory search speed for a large corpus and can afford to re-embed on restart (or save/load the index file manually). | RAM-only by default; fast ANN search at scale. | Medium — requires explicit save_local() / load_local() calls; no built-in metadata server. |
similarity_search returns results, but they're topically unrelated. Fix: log the model name alongside the index and assert it matches on reload.page_content="". They pollute the index and surface as top results. Fix: filter before indexing — chunks = [c for c in chunks if c.page_content.strip()].save_local(). On the next run, load_local() raises FileNotFoundError. Fix: always call save_local() immediately after from_documents().vector_store._collection.count() (Chroma) or faiss_store.index.ntotal (FAISS). Confirm the count equals your chunk list length.similarity_search("a query you know the answer to", k=3) and read the returned chunks. Off-topic results mean the model or chunks are wrong.With a verified, persisted store, wrap it as a LangChain using .as_retriever(). Tune search_type, k, and metadata filters to control what evidence reaches the model.
Turn the vector store into a LangChain `Retriever` by calling `.as_retriever()`, then tune `search_type`, `k`, and metadata filters to control what evidence reaches the prompt. You'll revisit the embedding model choice from Module 2 to understand why retrieval quality depends on it.
Configure a LangChain vector store retriever with search type, top-k, and metadata filters to control what evidence reaches the prompt.
Why this matters: The retriever is the gatekeeper between your indexed documents and the LLM — tuning it directly determines whether your chatbot finds the right evidence or hallucinates.
Decision this forces: What top-k value and search_type balance recall against prompt-window cost for your query distribution?
Answer: the decides the geometry of the space. Chunks that the model considers semantically similar cluster near each other; chunks it considers unrelated sit far apart.
The you configure in this module inherits that geometry. If the embedding model misrepresents your domain vocabulary, no amount of tuning k or search_type will rescue a fundamentally misaligned index.
That dependency is why retrieval quality and embedding choice are inseparable — and why this module revisits that decision before you tune anything else.
Calling .as_retriever() on a wraps it in a standard LangChain interface. It accepts a query string and returns a ranked list of objects.
Two parameters control what comes back: search_type picks the ranking algorithm. caps how many chunks reach the prompt.
Higher k improves recall. It reduces missed answers. But it inflates prompt cost and can dilute the signal with off-topic chunks.
The retriever sits between the vector store and the prompt. It is the last place you can filter, rank, and cap evidence before the LLM sees it.
# Assume `vector_store` was built in Module 2 retriever = vector_store.as_retriever( search_type="similarity", search_kwargs={"k": 4}, ) # Invoke with a plain query string docs = retriever.invoke("What is the refund policy?") print(len(docs), docs[0].page_content[:120])
.as_retriever(...)search_type="similarity"search_kwargs={"k": 4}retriever.invoke(...)This is the minimal retriever setup: one call on the vector store you persisted in Module 2, with k=4 and the default similarity search type.
The output line prints how many chunks came back and the first 120 characters of the top result — a quick sanity check before wiring the retriever into the chain.
It prints the full 90-character sentence — Python slicing to [:120] on a shorter string just returns the whole string without error. You'll see the complete sentence, not a truncated version.
retriever = vector_store.as_retriever( search_type="mmr", search_kwargs={ "k": 4, "fetch_k": 20, "filter": {"source": "employee_handbook.pdf"}, }, ) docs = retriever.invoke("What is the parental leave policy?") print([d.metadata["source"] for d in docs])
search_type="mmr""fetch_k": 20"filter": {"source": "employee_handbook.pdf"}d.metadata["source"]This stage adds two things to Stage 1: search for diversity, and a metadata filter that scopes retrieval to a single source file.
The fetch_k=20 parameter tells MMR to pull 20 candidates first, then re-rank and return the 4 most diverse and relevant ones.
['employee_handbook.pdf', 'employee_handbook.pdf', 'employee_handbook.pdf', 'employee_handbook.pdf'] — all four returned chunks carry the filtered source. If you see any other filename, the filter is not being applied (check that your vector store backend supports metadata filtering).
# Your corpus chunks have metadata: {"source": str, "year": int} # Goal: retrieve the 5 most diverse chunks from 2024 only. retriever = vector_store.as_retriever( search_type=_______________, # TODO 1: pick the right type search_kwargs={ "k": 5, "fetch_k": 25, "filter": _______________, # TODO 2: filter to year == 2024 }, ) docs = retriever.invoke("Key product launches") assert all(d.metadata["year"] == 2024 for d in docs), "Filter failed!"
assert all(...)d.metadata["year"]Stop — attempt both TODOs before revealing the answer. You want diversity across 2024 chunks, and the assert will catch a broken filter immediately.
Hint 1: you need the search type that re-ranks for diversity. Hint 2: the filter dict mirrors the metadata key-value pair exactly.
TODO 1: "mmr" — you need diversity across 2024 results, so MMR is the right type.
TODO 2: {"year": 2024} — the filter matches the metadata key 'year' with the integer value 2024.
Changed lines vs Stage 2: search_type stays "mmr" (same), fetch_k bumped to 25 to match the larger k=5, and the filter key changed from 'source' to 'year' with an integer value instead of a string. The assert is new — it enforces the filter in code rather than just printing.
Drag to see how top-k shifts the recall/cost tradeoff. Most production chatbots land between k=3 and k=6.
| Option | Result diversity | Handles redundant chunks | When to choose | Cost | Complexity |
|---|---|---|---|---|---|
| similarity | Returns the k most similar chunks, which may all be near-duplicates from the same passage. | No deduplication — repeated content in the corpus surfaces repeatedly. | Factual look-ups where the top-ranked chunk is almost always the right one (e.g., 'What is the refund policy?'). | Lowest — one nearest-neighbour pass. | None — default behaviour, no extra parameters. |
| mmr | Balances relevance and diversity — each successive chunk must be both relevant and different from already-selected ones. | Penalises near-duplicate chunks, so overlapping passages don't crowd out distinct evidence. | Multi-facet questions or corpora with overlapping chunks (e.g., 'Summarise our Q1 and Q2 results') where you need breadth. | Slightly higher — fetches fetch_k then re-ranks for diversity. | One extra parameter: fetch_k (candidates to re-rank, typically 3–5× k). |
Three failure patterns account for most retrieval misses in production chatbots.
fetch_k ≥ 3×k."Source" vs "source", or an int vs a string. Symptom: the LLM hallucinates or says it found nothing. Removing the filter returns results. Fix: print docs[0].metadata from an unfiltered query to confirm the exact key names and value types before filtering.fetch_k > k whenever search_type="mmr". A generated config often sets them equal. That defeats MMR. (2) filter keys match your actual metadata field names exactly. Print a sample doc's .metadata to confirm. (3) the assert pattern from the practice block is present. That makes filter failures surface immediately.With a solid retriever in place, the next module — Engineer the RAG Prompt Template — shows you how to pack those retrieved chunks into a structured prompt that grounds the LLM's answer.
Build a `ChatPromptTemplate` that injects retrieved context and the user question into a grounding instruction, then test it with a hardcoded context block before wiring it to the live retriever. You'll write the system instruction that tells the model to cite sources and refuse out-of-scope questions.
Build a ChatPromptTemplate that injects retrieved context and a grounding instruction into every model call.
Why this matters: The prompt template is the contract between your retriever and the model — it determines whether answers stay grounded in your documents or drift into hallucination.
Decision this forces: How should the system instruction handle questions the retrieved context doesn't answer — refuse, hedge, or fall back to parametric knowledge?
A returns a list of objects — each with a page_content string and a metadata dict (source path, page number, etc.).
Module 3 let you tune and search_type to control which chunks surface. This module takes those chunks and shapes them into a prompt the model can actually use.
Your chatbot's answer quality lives or dies in the prompt template. A holds two message slots: a that sets the rules, and a human message that carries the retrieved evidence plus the user's question.
The system message is where you enforce — the instruction that tells the model to answer only from the provided context, cite its sources, and refuse or hedge when the context doesn't cover the question.
The human message uses two placeholders: {context} (the formatted chunk text) and {question} (the raw user query). At runtime, LangChain fills both before the message reaches the model.
Imagine your retriever returns three chunks about a refund policy from different pages of a PDF. The model can't consume a Python list — you need to flatten those objects into a single readable string before slotting them into {context}.
A clean pattern: join each chunk's page_content with a separator like "\n\n---\n\n", and optionally prepend the source filename so the model can cite it. That gives the model clear boundaries between chunks and a citation anchor for each one.
You'll test this with a hardcoded context block first — no live retriever yet. That isolates prompt logic from retrieval logic, so you can verify the template independently before wiring them together.
from langchain_core.prompts import ChatPromptTemplate SYSTEM = """ You are a helpful assistant. Answer ONLY from the context below. Cite the source filename for every claim you make. If the context does not contain the answer, say: "I don't have enough information in the provided documents." """ prompt = ChatPromptTemplate.from_messages([ ("system", SYSTEM), ("human", "Context:\n{context}\n\nQuestion: {question}"), ])
ChatPromptTemplate.from_messages([...])("system", SYSTEM){context} / {question}This template encodes the grounding contract in the system message and reserves two runtime slots in the human message. The model sees the system instruction on every call — it never drifts to parametric answers without an explicit label.
A list of message objects (BaseMessage instances): one SystemMessage and one HumanMessage, each with its content already filled in. LangChain passes this list directly to the chat model.
def format_docs(docs): return "\n\n---\n\n".join( f"[Source: {d.metadata.get('source', 'unknown')}]\n{d.page_content}" for d in docs ) # Hardcoded test — no live retriever yet from langchain_core.documents import Document fake_docs = [ Document(page_content="Refunds take 5–7 days.", metadata={"source": "policy.pdf"}), Document(page_content="Contact support@co.com for help.", metadata={"source": "faq.pdf"}), ] formatted = format_docs(fake_docs) print(prompt.format_messages(context=formatted, question="How long do refunds take?"))
d.metadata.get('source', 'unknown')"\n\n---\n\n".join(...)prompt.format_messages(context=..., question=...)format_docs flattens the retriever's Document list into a single string with source labels and chunk separators. Testing with hardcoded docs lets you confirm the template renders correctly before the live retriever is wired in.
Context:
[Source: policy.pdf]
Refunds take 5–7 days.
---
[Source: faq.pdf]
Contact support@co.com for help.
Question: How long do refunds take?
When the joined chunks exceed the model's context window, content near the end is silently dropped or deprioritized. You'll see answers that reference only the first chunk even when the answer is in chunk 4. Fix: cap so total chunk tokens stay well under the model's limit. A safe rule is ≤ 50% of the context window for retrieved text.
Without an explicit "cite the source" line in the system message, the model blends retrieved facts with training knowledge. It presents both as equally certain. The symptom is answers that sound right but reference details not in any chunk. Fix: add a mandatory citation line and spot-check answers against the raw formatted string.
If the system message says "only answer from context" but doesn't define what to do when context is absent, the model either hallucinates or refuses every borderline question. Fix: give it an explicit fallback phrase, as in the template above. Then it knows exactly what to say when the context falls short.
{context} and {question} appear in the human message, (2) the system message contains an explicit citation instruction, (3) the fallback phrase is present, and (4) format_docs includes source metadata. A generated version often drops one of these silently.from langchain_core.prompts import ChatPromptTemplate STRICT_SYSTEM = """ You are a compliance assistant. Answer ONLY from the context below. Cite the source filename for every claim. # TODO: add the hard-refuse instruction for out-of-scope questions """ strict_prompt = ChatPromptTemplate.from_messages([ ("system", STRICT_SYSTEM), ("human", "Context:\n{context}\n\nQuestion: {question}"), ]) # Test it: what should the model say when context is empty?
# TODO: add the hard-refuse instructionThis is a variation of the working template adapted for a compliance use case that requires a hard refuse. One line is missing — supply it before revealing the answer.
Replace the TODO with:
If the context does not contain the answer, respond only with: "I cannot answer this question from the provided documents."
Changed line vs. the earlier template: the fallback phrase is now a hard refusal with no partial answer offered — no 'here's what I do know'. This removes the model's escape hatch to blend in parametric knowledge, which is the right call for compliance domains where a confident wrong answer carries legal risk.
Compose the retriever, prompt template, and LLM into a single LCEL chain using the pipe operator (`|`), then invoke it with a user question to get a grounded answer. You'll complete the chatbot's query path — from raw question in to cited answer out — and add a source-citation pass-through so the response includes document references.
Compose the retriever, prompt, and LLM into a single LCEL chain and invoke it end-to-end to get a grounded, cited answer.
Why this matters: This is the assembly step that turns your separate RAG components into a working chatbot query path — nothing answers questions until this chain exists.
Decision this forces: Should the chain return a single answer string or a structured object that includes both the answer and the source documents?
ChatPromptTemplate with two variables. What were they, and which one carries the retrieved passages?Answer: the template held {context} (the retrieved chunks) and {question} (the user's query). You tested it with a hardcoded context block — now you'll wire the live into that slot and close the loop.
This module is the assembly step: you take the retriever (Module 3), the prompt (Module 4), and an LLM, then pipe them into a single that answers any question end-to-end.
LangChain Expression Language () lets you compose any runnable components with the | operator — each component's output becomes the next one's input.
A RAG follows the path: retriever → prompt → LLM → output parser. Every step is a Runnable, so they all share the same .invoke() / .stream() interface.
The key wiring detail: the retriever returns a list of objects, but the prompt expects a plain string for {context}. You need a small formatting step between them — that's the only non-obvious join.
from langchain_core.runnables import RunnablePassthrough # Naive attempt — plug retriever directly into prompt naive_chain = retriever | prompt | llm result = naive_chain.invoke("What is the refund policy?") print(result)
retriever | prompt | llmnaive_chain.invoke(...)This looks clean, but it fails because the retriever outputs a list of Document objects while the prompt template expects a plain string for {context}. Running this raises a type error before the LLM is ever called.
LangChain raises a validation error such as: 'Expected a string for variable context, got list'. The prompt template can't serialize a list of Document objects — you must format them into a single string first.
from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser def format_docs(docs): return "\n\n".join(doc.page_content for doc in docs) rag_chain = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | prompt | llm | StrOutputParser() ) print(rag_chain.invoke("What is the refund policy?"))
{"context": retriever | format_docs, "question": RunnablePassthrough()}format_docs(docs)StrOutputParser()The dict step runs two branches in parallel: the retriever fetches chunks and format_docs joins them into a string, while RunnablePassthrough() forwards the raw question unchanged. Both values land in the prompt's {context} and {question} slots.
RunnablePassthrough() forwards the chain's input (the question string) unchanged into the 'question' key. Without it you'd need to manually extract the question from the input dict, or the prompt would receive None for {question}.
# … def get_sources(docs): return [d.metadata.get("source", "unknown") for d in docs] citation_chain = ( { "context": retriever | format_docs, "question": RunnablePassthrough(), "sources": retriever | RunnableLambda(get_sources), } | RunnablePassthrough.assign(answer=prompt | llm | StrOutputParser()) ) result = citation_chain.invoke("What is the refund policy?") print(result["answer"]) print(result["sources"])
RunnableLambda(get_sources)RunnablePassthrough.assign(answer=...)d.metadata.get("source", "unknown")This version returns a structured dict with both answer and sources keys, so the chatbot UI can render citations alongside the response. RunnablePassthrough.assign() merges a new key into the existing dict without discarding the others.
A list of strings — get_sources() extracts only the 'source' field from each Document's metadata dict, so you get e.g. ['docs/policy.pdf', 'docs/faq.txt']. The Documents themselves are not in the output.
from langchain_core.output_parsers import StrOutputParser # Same retriever, prompt, and llm from earlier in the module streaming_chain = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | prompt | llm | # TODO: add the output parser that yields plain text tokens ) for chunk in streaming_chain.stream("What is the refund policy?"): print(chunk, end="", flush=True)
.stream("...")print(chunk, end="", flush=True)StrOutputParser()This is a small variation of Stage 2: everything is wired, but the final parser is missing. Supplying it is the crux — without it, .stream() yields raw LLM message objects instead of text tokens, and the print loop outputs garbled dicts.
Replace the TODO with: StrOutputParser()
Changed line: | StrOutputParser()
Why: StrOutputParser converts each streamed LLM message chunk into its plain string content, so the for-loop receives 'What', ' is', ' the', ... instead of AIMessageChunk objects. Without it, print(chunk) shows something like AIMessageChunk(content='What') on every iteration.
| Option | Citation support | Streaming compatibility | When to choose | Cost | Complexity |
|---|---|---|---|---|---|
| Plain string chain | None — sources are discarded after retrieval | Full — StrOutputParser streams tokens natively | When you only need the answer text — e.g. a simple CLI chatbot or a single-turn API endpoint. | Minimal | Low |
| Structured dict chain | Built-in — sources key carries file paths or IDs | Partial — the answer key can stream, but the full dict only resolves at completion | When the UI needs to render citations, highlight source passages, or log which documents were used. | Minimal extra | Medium |
Three failure patterns show up most often when assembling the end-to-end chain:
retriever appears in both the context and sources branches — so it runs two separate similarity searches. You pay double latency and may get different documents in each branch. Fix: retrieve once with RunnablePassthrough.assign and reuse the result.metadata["source"], get_sources() returns a list of "unknown" strings — citations appear in the UI but are useless. Check docs[0].metadata after retrieval before wiring the chain..stream() on the dict chain yields partial dicts, not token strings — the UI receives {} chunks until the answer key resolves. Stream only the answer sub-chain (prompt | llm | StrOutputParser()) when you need incremental tokens.result["sources"] and confirm the file paths match the documents you ingested — not placeholders or empty strings.retriever.invoke("test question") in isolation and check that len(docs) == k (the k you set in Module 3) — a misconfigured retriever silently returns 0 docs.Build a small evaluation set of queries with expected source IDs, run it against the assembled chain, and check for hallucinations, retrieval misses, and prompt-injection risks. You'll also learn how to verify AI-generated code in your pipeline — spotting silent failures like wrong chunk sizes or misconfigured top-k — before they reach users.
A structured process for testing your RAG pipeline — building an eval set, measuring retrieval hit rate, detecting hallucinations, and hardening against the three most common silent failures.
Why this matters: Your chatbot can produce confident-sounding wrong answers without a single error being raised — this module gives you the concrete checks that catch those failures before users do.
Decision this forces: What minimum hit-rate and groundedness score are acceptable for your use case before you ship?
|) wire together to form the end-to-end chain?Answer: the , the , and the LLM — composed with so a raw question flows in and a grounded answer flows out.
That chain is assembled, but assembly is not the same as correctness. This final module asks: how do you know it retrieves the right documents, stays grounded in them, and resists adversarial input before real users hit it?
An is a list of (query, expected_source_ids) pairs. It's the minimum artifact needed to measure retriever performance.
The key metric is : the fraction of queries where at least one expected source appears in the top- chunks. A hit rate below ~0.7 at k=4 signals you should revisit chunk size or embeddings.
Start with 10–20 hand-crafted cases from real user questions. Don't paraphrase document headings. Each case needs at least one expected source ID that a correct retriever must surface.
eval_set = [
{"query": "What is the refund policy?", "expected_ids": ["policy_doc_p3", "policy_doc_p4"]},
{"query": "How do I reset my password?", "expected_ids": ["support_faq_p1"]},
]
def hit_rate(retriever, cases, k=4):
hits = 0
for case in cases:
docs = retriever.get_relevant_documents(case["query"])[:k]
retrieved_ids = [d.metadata.get("source") for d in docs]
if any(eid in retrieved_ids for eid in case["expected_ids"]):
hits += 1
return hits / len(cases)
print(hit_rate(retriever, eval_set)) # e.g. 0.5 → investigateretriever.get_relevant_documents(query)[:k]d.metadata.get("source")any(eid in retrieved_ids ...)This runs every eval query through the live retriever and checks whether any expected source ID appears in the top-k results. A score of 0.5 on even a small set is a red flag — it means half your queries return the wrong evidence before the LLM ever sees them.
1.0 — both cases are hits, so hits=2, len(cases)=2, result is 1.0.
A hallucination check asks: does the answer contain claims not supported by the retrieved context? This is called a check. You can implement a lightweight version without a second LLM.
Simplest approach: verify every key noun phrase and figure appears verbatim in the chunks. If a number or proper noun is absent, flag it for review.
For a stronger signal, use a judge prompt: feed the answer and context to the LLM. Ask it to rate faithfulness on a 0–1 scale. Ragas automates this if you want a repeatable score.
def is_grounded(answer: str, context_docs: list) -> bool: context_text = " ".join(d.page_content for d in context_docs) # Extract simple numeric/proper-noun tokens as a proxy for key claims import re key_tokens = re.findall(r"\b[A-Z][a-z]+|\d+\.?\d*\b", answer) ungrounded = [t for t in key_tokens if t not in context_text] return len(ungrounded) == 0, ungrounded grounded, missing = is_grounded(answer, retrieved_docs) if not grounded: print("Possible hallucination — tokens not in context:", missing)
re.findall(r"\b[A-Z][a-z]+|\d+\.?\d*\b", answer)[t for t in key_tokens if t not in context_text]return len(ungrounded) == 0, ungroundedThis heuristic flags answers whose capitalized terms or numbers don't appear in the retrieved context — a fast, zero-cost first pass before a judge-LLM call. It won't catch paraphrased hallucinations, but it catches fabricated figures and names reliably.
(False, ['45']) — '45' is extracted as a key token but is absent from the context, so the function flags it as a possible hallucination. The changed line is the ungrounded list comprehension finding '45' missing.
These three failures produce plausible answers with no error raised. You only catch them by running your eval set or inspecting outputs.
len(doc.page_content.split()) and trim if over budget.<context>…</context> and tell the model to treat it as untrusted data.Before trusting AI-generated LangChain code, run these four checks at each stage boundary:
len(docs) and docs[0].metadata. Confirm the source field is populated and page count matches.len(chunks) > len(docs) and spot-check len(chunks[0].page_content.split()). AI-generated splitters often default to chunk_size=1000, which is too large.retriever.get_relevant_documents("test query") directly. Check that k matches your intent and source IDs are in metadata.Run your eval set after each fix. A single change to chunk size or k can shift hit rate by 20+ points.
Before looking at the build order below, reconstruct it from memory: what are the six stages of the RAG pipeline in order, and what does each stage hand off to the next? Sketch the data flow from raw document to final cited answer.
Apply what you learned to A production-ready RAG chatbot that loads documents, retrieves relevant context, and answers questions grounded in that context.
You load a PDF and a set of web pages into the same pipeline. After splitting, you want a single flat list of chunks ready for embedding. Which approach is correct?
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
all_chunks = splitter.split_documents(pdf_docs + web_docs)
split_documents() accepts any list of Document objects regardless of origin, so concatenating the two lists before the call is the idiomatic way to build a single flat chunk list. There is no merge() method — the output is already flat. split_documents() handles a list natively, no loop needed. chunk_overlap controls intra-chunk context continuity and has nothing to do with cross-source duplication.
Your RAG app will run inside a serverless function that is torn down after each request — no disk access is guaranteed between calls. Which vector store backend fits this deployment, and why?
Chroma's persistence model writes to a local directory, which is unreliable in ephemeral serverless containers. FAISS can serialize its index to bytes, store it in object storage (e.g., S3), and reload it at startup — no re-embedding required. The two backends are not equivalent for this deployment pattern. FAISS does not re-embed on load; you serialize the already-embedded index and deserialize it.
A user asks a very specific question but your retriever keeps returning chunks that are topically related but not the ones that actually contain the answer. You suspect the top-k results are dominated by near-duplicate passages. Name the search_type setting you should switch to and explain in one sentence why it helps.
MMR balances relevance against diversity by penalizing chunks that are too similar to already-selected results. Plain similarity search ranks purely by vector distance, so near-duplicate passages cluster at the top and push the genuinely useful chunk down. Switching to MMR directly addresses this retrieval miss pattern.
You are writing the system message for your RAG prompt template. A user asks a question that has no answer in the retrieved context. Which behavior should your system instruction enforce?
When the retrieved context is silent on a question, the safest grounded behavior is an explicit refusal or hedge ('I don't have enough information in the provided documents'). Answering from parametric knowledge breaks the grounding contract and risks hallucination presented as document-backed fact. Returning an empty string is a chain design choice, not a prompt instruction. Retrying with a higher k is a valid pipeline strategy but cannot be encoded in the system message itself.
Look at this LCEL chain assembly:
chain = retriever | prompt | llm
result = chain.invoke({"question": user_query})
What is the most important capability this chain is MISSING for a production RAG chatbot?
Piping retriever | prompt | llm passes the retrieved text into the prompt but drops the Document metadata (source, page, date) that citations depend on. To surface citations you need to pass source metadata through a parallel branch or use a chain type like RetrievalQAWithSourcesChain. LCEL chains accept dict inputs natively, so a string question in a dict is fine. An output parser is optional — without one the chain returns the LLM's raw message object. Retrievers are first-class LCEL runnables and can be piped directly.