Chapter 6: Retrieval-Augmented Generation: Building a Knowledge-Base-Driven Agent

If an Agent relies solely on the LLM's built-in knowledge, it's like a student who only relies on memory—knowledge is limited and becomes outdated. RAG (Retrieval-Augmented Generation) allows an Agent to retrieve professional, real-time information from an external knowledge base, significantly improving the accuracy and depth of its answers. This chapter will systematically explain RAG technology, from full-process analysis to advanced strategies.

6.1 Full RAG Process Analysis: Data Cleaning, Chunking, and Index Construction

The Core RAG Process

Offline Phase: Documents -> Cleaning -> Chunking -> Embedding -> Vector Database
Online Phase: Question -> Embedding -> Retrieval -> Insert into Prompt -> LLM Generation

Data Cleaning

Raw documents are often chaotically formatted and contain noise. Cleaning quality directly affects retrieval results:

import re

class DocumentCleaner:
    """Document cleaning utility"""

    def clean(self, text: str) -> str:
        text = self._remove_boilerplate(text)
        text = self._normalize_whitespace(text)
        text = self._remove_special_chars(text)
        return text.strip()

    def _remove_boilerplate(self, text: str) -> str:
        """Remove boilerplate text like headers, footers, and copyright notices"""
        patterns = [
            r"Copyright.*?All rights reserved",
            r"This document is for reference only.*?does not constitute any advice",
            r"Page\s*\d+\s*/\s*\d+",
        ]
        for pattern in patterns:
            text = re.sub(pattern, "", text, flags=re.IGNORECASE)
        return text

    def _normalize_whitespace(self, text: str) -> str:
        """Normalize whitespace characters"""
        text = re.sub(r"\n{3,}", "\n\n", text)  # Compress multiple newlines to two
        text = re.sub(r"[ \t]+", " ", text)       # Compress multiple spaces to one
        return text

    def _remove_special_chars(self, text: str) -> str:
        """Remove invisible characters and garbled text"""
        text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", text)
        return text

Text Chunking Strategies

Chunking is the most critical step in RAG. Chunks that are too large reduce retrieval precision; chunks that are too small lack complete context.

Strategy 1: Fixed-Length Chunking

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,       # 500 characters per chunk
    chunk_overlap=50,     # 50-character overlap between chunks
    separators=["\n\n", "\n", "。", "！", "？", "；", " "],
)

Strategy 2: Semantic Chunking

Chunk by semantic boundaries (paragraphs, sections) rather than mechanical cuts:

from langchain.text_splitter import MarkdownHeaderTextSplitter

# Chunk by Markdown heading levels
md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "h1"),
        ("##", "h2"),
        ("###", "h3"),
    ]
)
chunks = md_splitter.split_text(markdown_doc)
# Each chunk automatically carries its heading hierarchy information

Strategy 3: Parent-Child Chunking (Small-to-Big Retrieval)

Use small chunks for retrieval (high precision) and large chunks for return (complete context):

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

# Child splitter (for retrieval)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)
# Parent splitter (for return)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1000)

vectorstore = Chroma(embedding_function=embeddings)
docstore = InMemoryStore()  # Stores parent document originals

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=docstore,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

Chunking Strategy	Advantages	Disadvantages	Use Cases
Fixed-Length	Simple and controllable	May cut off semantics	General scenarios
Semantic	Preserves semantic integrity	Depends on document structure	Markdown/HTML documents
Parent-Child	Balances precision and context	Complex implementation	High-quality RAG

Index Construction

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Build index from documents
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./knowledge_base",
    collection_metadata={"hnsw:space": "cosine"}  # Use cosine similarity
)

Reference: LangChain Text Splitters

6.2 Advanced Retrieval Strategies: Hybrid Search and Re-ranking Techniques

Limitations of Basic Vector Retrieval

Pure vector retrieval has a fundamental problem: semantic similarity does not equal answer relevance. For example, if a user asks "How to install Python," vector retrieval might return an article titled "Various Methods to Install Python" with outdated content, while missing a tutorial with a less relevant title but precise content.

Hybrid Search: Vector + Keyword

Hybrid search combines vector retrieval (semantic matching) and BM25 keyword retrieval (exact matching) to complement each other:

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# Vector retriever
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})

# BM25 keyword retriever
bm25_retriever = BM25Retriever.from_documents(chunks, k=10)

# Ensemble retriever
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.5, 0.5]  # 50% weight each for vector and keyword
)

results = ensemble_retriever.invoke("Python installation tutorial")

Re-ranking

Retrieved documents are sorted by similarity, but the most similar may not be the most relevant. A re-ranking model can more accurately assess document relevance to the query:

from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank

# Use Cohere re-ranking model
compressor = CohereRerank(model="rerank-v3.5", top_n=5)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=ensemble_retriever
)

# First retrieve a large number of candidates, then re-rank to filter the top 5 most relevant
results = compression_retriever.invoke("How to optimize RAG retrieval performance?")

Query Rewriting and Expansion

A user's original query may not be precise enough. Using an LLM to rewrite or expand the query can improve retrieval recall:

class QueryRewriter:
    def __init__(self, llm):
        self.llm = llm

    def expand_query(self, original_query: str) -> list[str]:
        """Expand the original query into multiple sub-queries"""
        prompt = f"""
Original query: {original_query}

Please generate 3 sub-queries from different angles to help retrieve more comprehensive information:
1. A synonym-replaced version
2. A more specific, detailed version
3. A broader, higher-level version
"""
        response = self.llm.invoke(prompt)
        return [original_query] + self._parse_queries(response.content)

    def hyde_query(self, original_query: str) -> str:
        """HyDE: Have the LLM generate a hypothetical answer first, then use the answer for retrieval"""
        prompt = f"Please answer the following question (give your best guess even if unsure): {original_query}"
        hypothetical_answer = self.llm.invoke(prompt).content
        return hypothetical_answer  # Use the hypothetical answer as the retrieval query

Reference Paper: HyDE: Precise Zero-Shot Dense Retrieval

6.3 Combining Knowledge Graphs and Agents: Utilizing Structured Data

Blind Spots of Vector Retrieval

Vector retrieval excels at semantic matching but struggles with entity relationships and structured queries. For example, "What other companies were founded by the people who co-founded PayPal with Elon Musk?" — this type of multi-hop relational query is almost impossible for vector retrieval to handle.

Knowledge Graph-Enhanced RAG

from langchain_community.graphs import Neo4jGraph
from langchain.chains import GraphCypherQAChain

# Connect to knowledge graph
graph = Neo4jGraph(
    url="bolt://localhost:7687",
    username="neo4j",
    password="your_password"
)

# Use LLM to generate Cypher queries
chain = GraphCypherQAChain.from_llm(
    llm=ChatOpenAI(model="gpt-4o", temperature=0),
    graph=graph,
    verbose=True,
)

result = chain.run("What other companies were founded by the people who co-founded PayPal with Elon Musk?")
# LLM automatically generates: MATCH (p:Person)-[:COFOUNDED]->(c:Company {name: 'PayPal'})<-[:COFOUNDED]-(other:Person)
#                MATCH (other)-[:COFOUNDED]->(other_c:Company)
#                RETURN other.name, collect(other_c.name)

GraphRAG: Microsoft's Graph-Enhanced Solution

Microsoft's GraphRAG solution process:

1. Extraction: Extract entities and relationships from documents
2. Construction: Build a community graph to discover entity clusters
3. Indexing: Generate summaries for each community
4. Retrieval: First locate relevant communities, then retrieve within them

# Simplified entity extraction
def extract_entities_and_relations(text: str, llm) -> dict:
    prompt = f"""Extract entities and relations from the following text, returning in JSON format:
Text: {text}

Format: {{
    "entities": [{{"name": "entity name", "type": "type"}}],
    "relations": [{{"source": "entity1", "target": "entity2", "relation": "relationship"}}]
}}"""
    response = llm.invoke(prompt)
    return json.loads(response.content)

Reference: Microsoft GraphRAG

6.4 Solving RAG Pain Points: Lost in the Middle and Multi-Hop Reasoning

Pain Point 1: Lost in the Middle

Research shows that LLMs pay the least attention to information in the middle of the context. When retrieval returns multiple document fragments, those placed in the middle are easily overlooked.

Solution: Document Reordering

def relevance_ordered_placement(documents: list[str], strategy: str = "descending") -> list[str]:
    """Reorder documents by relevance to avoid placing important information in the middle"""
    # Option 1: Descending order - most relevant first
    if strategy == "descending":
        return documents  # Keep retrieval order

    # Option 2: Alternating order - alternate most and least relevant
    if strategy == "alternating":
        result = []
        left, right = 0, len(documents) - 1
        while left <= right:
            result.append(documents[left])
            if left != right:
                result.append(documents[right])
            left += 1
            right -= 1
        return result

    return documents

Pain Point 2: Multi-Hop Reasoning

"What is the birthplace of the wife of the 46th President of the United States?" — This requires two-hop reasoning: Who is the 46th President -> Who is his wife -> What is her birthplace.

Solution: Iterative Retrieval

class MultiHopRetriever:
    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm

    def retrieve(self, query: str, max_hops: int = 3) -> list[str]:
        all_docs = []
        current_query = query

        for hop in range(max_hops):
            docs = self.retriever.invoke(current_query)
            all_docs.extend(docs)

            # Determine if further retrieval is needed
            judge_prompt = f"""
Original question: {query}
Retrieved information: {[d.page_content[:200] for d in all_docs]}

Can the question be fully answered now?
If not, please generate the sub-question needed for the next retrieval step.
"""
            response = self.llm.invoke(judge_prompt).content

            if "yes" in response.lower() and "complete" in response.lower():
                break

            # Extract the sub-question for the next retrieval step
            current_query = response.split("Sub-question:")[-1].strip() if "Sub-question:" in response else query

        return all_docs

Pain Point 3: Information Conflicts

Information from different sources may contradict each other:

def resolve_conflicts(documents: list[str], query: str, llm) -> str:
    """Handle information conflicts"""
    conflict_prompt = f"""
Question: {query}
Retrieved information (may contain conflicts):
{chr(10).join(f'Source {i+1}: {d}' for i, d in enumerate(documents))}

Please analyze:
1. Which pieces of information conflict with each other?
2. What are the possible reasons for each conflict (timeliness, source reliability, etc.)?
3. Which version do you tend to adopt? Why?
"""
    return llm.invoke(conflict_prompt).content

6.5 Dynamic Knowledge Base Updates: Incremental Indexing and Version Management Strategies

Why Dynamic Updates Are Needed

A knowledge base is not static. Product documentation gets updated, policies and regulations change, and technical solutions evolve. An outdated knowledge base is more dangerous than having none at all—it will provide incorrect information.

Incremental Indexing

class IncrementalIndexer:
    def __init__(self, vectorstore, embeddings):
        self.vectorstore = vectorstore
        self.embeddings = embeddings
        self.doc_hashes: dict[str, str] = {}  # doc_id -> content_hash

    def _content_hash(self, content: str) -> str:
        import hashlib
        return hashlib.md5(content.encode()).hexdigest()

    def update(self, documents: list) -> dict:
        """Incrementally update the index"""
        added, updated, unchanged = 0, 0, 0

        for doc in documents:
            doc_id = doc.metadata.get("doc_id", str(id(doc)))
            new_hash = self._content_hash(doc.page_content)

            if doc_id not in self.doc_hashes:
                # New document
                self.vectorstore.add_documents([doc])
                self.doc_hashes[doc_id] = new_hash
                added += 1
            elif self.doc_hashes[doc_id] != new_hash:
                # Document changed, delete old version and add new version
                self.vectorstore.delete(ids=[doc_id])
                self.vectorstore.add_documents([doc])
                self.doc_hashes[doc_id] = new_hash
                updated += 1
            else:
                unchanged += 1

        return {"added": added, "updated": updated, "unchanged": unchanged}

Version Management Strategy

class VersionedKnowledgeBase:
    def __init__(self, vectorstore):
        self.vectorstore = vectorstore
        self.versions: dict[str, list[dict]] = {}  # doc_id -> version list

    def add_version(self, doc_id: str, content: str, metadata: dict = None):
        """Add a new version of a document"""
        version = {
            "content": content,
            "metadata": metadata or {},
            "timestamp": datetime.now().isoformat(),
            "version": len(self.versions.get(doc_id, [])) + 1
        }
        self.versions.setdefault(doc_id, []).append(version)

        # Only write the latest version to the vector store (marked as current)
        self.vectorstore.add_documents([{
            "page_content": content,
            "metadata": {**(metadata or {}), "doc_id": doc_id, "version": version["version"], "status": "current"}
        }])

    def rollback(self, doc_id: str, target_version: int):
        """Roll back to a specified version"""
        if doc_id not in self.versions:
            return False

        versions = self.versions[doc_id]
        target = next((v for v in versions if v["version"] == target_version), None)
        if not target:
            return False

        # Delete current version, restore target version
        self.add_version(doc_id, target["content"], target["metadata"])
        return True

Automatic Update Trigger Mechanism

import hashlib
from datetime import datetime

class AutoUpdater:
    """Monitor source file changes and automatically trigger index updates"""

    def __init__(self, indexer: IncrementalIndexer, source_dir: str):
        self.indexer = indexer
        self.source_dir = source_dir
        self.file_hashes: dict[str, str] = {}

    def scan_and_update(self) -> dict:
        """Scan the source directory, detect changes, and update"""
        changes = {"added": [], "modified": [], "deleted": []}

        # Scan current files
        current_files = {}
        for root, dirs, files in os.walk(self.source_dir):
            for f in files:
                if f.endswith(('.md', '.txt', '.pdf', '.docx')):
                    filepath = os.path.join(root, f)
                    with open(filepath, 'rb') as file:
                        file_hash = hashlib.md5(file.read()).hexdigest()
                    current_files[filepath] = file_hash

        # Detect additions and modifications
        for filepath, file_hash in current_files.items():
            if filepath not in self.file_hashes:
                changes["added"].append(filepath)
            elif self.file_hashes[filepath] != file_hash:
                changes["modified"].append(filepath)

        # Detect deletions
        for filepath in self.file_hashes:
            if filepath not in current_files:
                changes["deleted"].append(filepath)

        # Execute updates
        # ... (load changed files, chunk, write to index)

        self.file_hashes = current_files
        return changes

Chapter Summary

RAG Stage	Key Technology	Core Point
Data Cleaning	Regex denoising, format normalization	Garbage in = garbage out; cleaning is the foundation
Chunking Strategy	Fixed-length / Semantic / Parent-Child	Parent-child chunking balances precision and context
Hybrid Retrieval	Vector + BM25 + Re-ranking	Semantic matching and exact matching complement each other
Query Optimization	Rewriting / Expansion / HyDE	Use LLM to optimize retrieval queries
Knowledge Graph	Cypher + GraphRAG	Structured relationships + multi-hop reasoning
Pain Point Solutions	Document reordering / Iterative retrieval / Conflict resolution	Lost in the middle / Multi-hop / Information conflicts
Dynamic Updates	Incremental indexing / Version management / Automatic scanning	A knowledge base is not a one-time project

In the next chapter, we will move from single Agents to multi-Agents—enabling multiple intelligent agents to work together to solve more complex systemic problems.