Building an Unbreakable RAG Knowledge Base: Lazy Initialization, Idempotent Ingestion, and Graceful Degradation
This page has been translated by machine translation. View original
Introduction
I incorporated RAG (Retrieval-Augmented Generation) functionality into an in-house AI automation tool. The system takes Markdown documents into a vector store and answers questions in chat with sources attached.
However, RAG is fragile. I'll introduce the problems encountered in production operations—such as vector store initialization failures, embedding API timeouts, and chunk duplication on restart—along with the patterns established to address them.
Prerequisites & Environment
- Python 3.12, FastAPI
- Vector store: ChromaDB (local persistence)
- Embedding: OpenAI
text-embedding-3-small - Chunking: LangChain
MarkdownHeaderTextSplitter+RecursiveCharacterTextSplitter
Pattern 1: Lazy Initialization
This is a pattern where vector store initialization is performed at the time of the first request, rather than at application startup.
def get_chroma(collection: str = KB_COLLECTION) -> Chroma:
pdir = _persist_dir(collection)
os.makedirs(pdir, exist_ok=True)
try:
vec = Chroma(
collection_name=collection,
persist_directory=pdir,
embedding_function=get_embeddings(),
)
# Log document count (best effort)
count = None
try:
underlying = getattr(vec, "_collection", None)
if underlying and hasattr(underlying, "count"):
count = int(underlying.count())
except Exception:
count = None
logger.info(json.dumps({
"message": "chroma_initialized",
"persist_directory": pdir,
"doc_count": count,
}))
return vec
except Exception as e:
logger.error(json.dumps({
"message": "chroma_init_failed",
"error": str(e),
}))
raise
Lazy ingestion is performed on the chat agent side.
# Ingest only if collection is empty
try:
if count_docs(KB_COLLECTION) <= 0:
yield ChatEvent("log", {"step": "kb", "message": "KB empty; ingesting"})
ingest_kb(KB_COLLECTION)
except Exception as exc:
logger.warning("kb_ingest_failed: %s", exc)
# Continue without RAG if ingestion fails
Why use lazy initialization:
- Even if the embedding API is unavailable at app startup, other features (chat, Jira integration, etc.) can work normally
- Vector store setup takes a few seconds to tens of seconds, so startup time can be reduced
- Users who don't use the knowledge base feature incur no initialization cost
Pattern 2: Idempotent Ingestion (Deterministic Chunk IDs)
When documents are re-ingested on every restart, the same chunks get stored as duplicates. The same content appears multiple times in search results, wasting the context window.
The solution is deterministic chunk IDs. A SHA1 hash is generated from the filename, heading path, and byte offset, so that chunks with the same content always get the same ID.
import hashlib
def _deterministic_id(md: dict[str, Any]) -> str:
"""Generate a deterministic ID from chunk metadata."""
basis = (
f"{md.get('filename', '')}|"
f"{md.get('heading_path', '')}|"
f"{md.get('start_index', '')}|"
f"{md.get('end_index', '')}"
)
# SHA1 is sufficient for data identification, not security purposes
return hashlib.sha1(basis.encode("utf-8")).hexdigest()
This ID is used to add documents to ChromaDB during ingestion.
docs = [Document(page_content=c["text"], metadata=c["metadata"]) for c in chunks]
ids = [_deterministic_id(d.metadata) for d in docs]
vec.add_documents(docs, ids=ids)
Since ChromaDB overwrites (upserts) when a document with the same ID is added, chunks will not be duplicated even after a restart.
Pattern 3: Chunking That Preserves Heading Hierarchy
Chunking Markdown by simply "splitting text at fixed lengths" is not sufficient. By preserving the hierarchical structure of headings, context remains in the search results.
from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter
def split_markdown_file(path: str, chunk_size: int = 800, chunk_overlap: int = 120) -> list[dict[str, Any]]:
with open(path, encoding="utf-8") as f:
raw = f.read()
fm, body = parse_frontmatter(raw)
# Step 1: Logically split by headings
md_splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=[
("#", "Header 1"),
("##", "Header 2"),
("###", "Header 3"),
]
)
header_docs = md_splitter.split_text(body)
# Step 2: Further split each section by character count
char_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
add_start_index=True,
)
out = []
filename = os.path.basename(path)
for doc in header_docs:
heading_path = _build_heading_path(doc.metadata)
chunks = char_splitter.create_documents([doc.page_content])
for c in chunks:
meta = {"filename": filename}
if heading_path:
meta["heading_path"] = heading_path
# Save byte offset (used for deterministic ID generation)
start_index = c.metadata.get("start_index")
if isinstance(start_index, int):
meta["start_index"] = start_index
meta["end_index"] = start_index + len(c.page_content)
# Propagate frontmatter metadata
if fm.get("source_url"):
meta["source_url"] = fm["source_url"]
out.append({"text": c.page_content, "metadata": meta})
return out
Building the heading path:
def _build_heading_path(md_meta: dict[str, str]) -> str:
"""Build a path like Header 1 > Header 2 > Header 3."""
keys = [k for k in md_meta if k.lower().startswith("header ")]
keys.sort(key=lambda k: int(k.split()[1]))
parts = [str(md_meta[k]).strip() for k in keys if md_meta[k].strip()]
return " > ".join(parts)
This attaches information like heading_path: "Setup > Environment Setup > Docker Configuration" to chunk metadata. It can be used for displaying search results and providing context to the LLM.
Pattern 4: Graceful Degradation
No matter where a failure occurs in each stage of RAG (initialization → ingestion → retrieval → context injection), the chat functionality itself keeps running.
File level: Don't let one file's failure stop everything
for path in files:
try:
chunks = split_markdown_file(path)
# ... ingestion processing
except Exception as e:
logger.error(json.dumps({
"message": "split_markdown_failed",
"path": path,
"error": str(e),
}))
continue # Skip this file and move to the next
Agent level: Respond even without RAG
# KB agent processing flow
GROUNDS = "" # RAG context (may be empty)
if rag_is_ready():
results = rag_retrieve(content, k=5)
# ... Store retrieval results in GROUNDS
# ... Emit sources/chunks events
# LLM can respond even if GROUNDS is empty (becomes a general answer without knowledge base)
gen = gen_fn(
prompt=f"{GROUNDS}\n\n{prompt}" if GROUNDS else prompt,
model=meta.model,
)
Degradation stages:
| Failure Point | Behavior |
|---|---|
| Embedding API | General answer without RAG |
| Parsing of some files | RAG with only successfully parsed files |
| ChromaDB initialization | General answer without RAG |
| Zero search results | General answer without RAG |
Pattern 5: Source Citation
When passing search results to the LLM, specify a citation format to ensure source transparency.
# Build previews from search results
previews = []
for c in chunks_payload[:5]:
loc = ""
if c.get("startLine") is not None and c.get("endLine") is not None:
loc = f":{c['startLine']}-{c['endLine']}"
previews.append(f"[{c['filename']}{loc}] {c['preview']}")
# Context injection into LLM
GROUNDS = (
"Use the following knowledge base excerpts when relevant.\n"
"Cite inline using [filename:startLine-endLine]. If unsure, say so.\n"
+ "\n".join(previews)
)
Search result metadata is also sent to the frontend as SSE events to display source cards in the UI.
# Send source information via SSE
yield ChatEvent("sources", sources_payload)
yield ChatEvent("chunks", {
"chunks": chunks_payload,
"document_set": "knowledge-base",
"strategy": "vector",
})
Payload for each chunk:
{
"id": chunk_id, # SHA1 hash
"filename": "setup.md",
"url": "https://...", # source_url from frontmatter
"heading_path": "Setup > Docker",
"score": 0.87, # similarity score
"preview": "Docker configuration is...", # first 240 characters
}
Summary
| Pattern | Purpose | Implementation |
|---|---|---|
| Lazy initialization | Reduce startup time, fault isolation | Ingest at first request |
| Idempotent ingestion | Prevent chunk duplication on restart | Deterministic chunk IDs via SHA1 |
| Heading hierarchy preservation | Preserve context in search results | Two-stage chunking (headings → character count) |
| Graceful degradation | Continue chat even when RAG fails | try-except + fallback at each stage |
| Source citation | Transparency of answers | [filename:startLine-endLine] format |
The most important lesson is that RAG is not just about "retrieving and injecting"—the core design is deciding what to do when retrieval fails. By designing the system so that it can return a response to the user no matter which of the vector store, embedding API, or file parsing breaks down, RAG can be operated stably as a feature that is "useful when available, but still works without it."
