How We Built a RAG‑Powered Knowledge Base That Actually Understands Source Code

Why FAQ Systems Fail

A production incident during a major e‑commerce promotion showed that the standard FAQ answer "check payment status" was useless because the real cause lay in a code path involving a WAIT_COMPENSATION saga state, a Redis lock timeout, and a version‑specific lock lease change. The relevant information was scattered across Git commits, design documents, release notes, and post‑mortems, illustrating that valuable answers reside in engineering artifacts rather than static FAQs.

Four Types of Knowledge in a Source‑Code Knowledge Base

Descriptive Knowledge : product docs, runbooks, incident post‑mortems, design reviews.

Executable Knowledge : source code, configs, SQL, scripts, CI/CD pipelines.

Structural Knowledge : module dependencies, call graphs, class hierarchies, state machines.

Temporal Knowledge : branch diffs, version changes, gray‑release flags, commit timestamps.

Relying only on dense vector similarity leads to three typical failures: returning generic status docs instead of the compensation code, missing ACL‑related logic for a tenant_id field, and mixing evidence from different versions.

Overall Architecture: Online and Offline Pipelines

The system is split into an online request path (milliseconds‑scale answer generation) and an offline pipeline that continuously transforms code, docs, and change events into searchable, versioned knowledge snapshots.

+-----------------------+
|  API Gateway / SSO    |
+-----------+-----------+
            |
   +--------v--------+
   | Query Orchestrator |
   | rewrite / policy   |
   +----+----------+-----+
        |          |
+-------v--+   +--v----------------+
| Retrieval |   | Answer Guardrail |
| Coordinator|   | citation / risk |
+---+----+-----+   +---------+---------+
    |                     |
    +---------------------+

The offline pipeline builds snapshots:

repo webhook -> parser -> semantic chunk -> embedding -> indexing
   document sync -> metadata enrich -> snapshot publish -> rollback

Four Design Planes

Control Plane : manage knowledge source ingestion, index rules, model routing, snapshot publishing, rollback.

Execution Plane : query rewriting, retrieval, re‑ranking, context assembly, generation, streaming.

State Plane : store document chunks, code chunks, symbol tables, dependency graphs, snapshots, feedback.

Governance Plane : ACL filtering, audit logging, risk interception, observability, cost governance.

Online Request Flow (7 Steps)

Problem normalization – e.g.

domain=order, symptom=processing_timeout, expected_artifacts=code+runbook+postmortem

Permission pruning – filter by repo, doc, env scopes.

Multi‑modal retrieval – dense vector, BM25 full‑text, symbol lookup, dependency‑graph expansion.

Cross‑rank – combine intent, evidence type, freshness, permissions, version consistency.

Context assembly – stitch code snippets, config explanations, incident conclusions into a coherent context.

Trusted generation – force the LLM to answer only from evidence; return “insufficient evidence” otherwise.

Result logging – store snapshot_id, cited fragments, latency, risk tags, user feedback for offline evaluation.

Key Code: Retrieval Coordinator

class RetrievalRequest:
    query: str
    principal_id: str
    snapshot_id: str
    repo_scopes: list[str]
    top_k: int

class RetrievalCoordinator:
    def __init__(self, dense, sparse, symbol, graph, reranker):
        ...
    async def retrieve(self, req: RetrievalRequest) -> list[dict]:
        filters = {
            "snapshot_id": req.snapshot_id,
            "repo": {"$in": req.repo_scopes},
            "acl": {"$allow": req.principal_id},
        }
        dense_hits = await self.dense.search(req.query, top_k=40, filters=filters)
        sparse_hits = await self.sparse.search(req.query, top_k=40, filters=filters)
        symbol_hits = await self.symbol.search(req.query, top_k=20, filters=filters)
        graph_seeds = [hit["chunk_id"] for hit in symbol_hits[:5]]
        graph_hits = await self.graph.expand(seeds=graph_seeds, snapshot_id=req.snapshot_id,
                                            principal_id=req.principal_id, depth=1)
        merged = self._rrf_merge(dense_hits, sparse_hits, symbol_hits, graph_hits)
        return await self.reranker.rank(req.query, merged[:80])[: req.top_k]

Answer Service with Evidence Guard

class AnswerService:
    def __init__(self, retriever, llm_gateway, risk_guard, citation_builder):
        ...
    async def answer(self, query, principal, snapshot_id, repo_scopes):
        hits = await self.retriever.retrieve(RetrievalRequest(
            query=query,
            principal_id=principal.user_id,
            snapshot_id=snapshot_id,
            repo_scopes=repo_scopes,
            top_k=6,
        ))
        if not hits:
            return {"status": "insufficient_evidence", "answer": "当前知识快照中没有足够证据支撑回答...", "citations": []}
        context = self.citation_builder.build_context(hits, max_tokens=12000)
        prompt = self._build_prompt(query, context)
        draft = await self.llm_gateway.generate(prompt, temperature=0.1)
        checked = self.risk_guard.verify(draft=draft, evidence=hits)
        return {"status": checked.status, "answer": checked.answer, "citations": [...]}

The service refuses to answer when evidence is missing and always returns citations pointing to the exact snapshot_id, file path, symbol, and line range.

Snapshot‑Aware Configuration

# config/knowledge_engine.yaml
snapshot:
  active_snapshot: "kb-2026-07-14-02"
  publish_strategy: "blue_green"
  rollback_window: 3
retrieval:
  dense_top_k: 40
  sparse_top_k: 40
  symbol_top_k: 20
  rerank_top_k: 12
  final_top_k: 6
  timeout_ms: 450
filters:
  enforce_acl: true
  require_snapshot_consistency: true
  allowed_artifact_types:
    - markdown
    - source_code
    - yaml
    - sql
    - postmortem
generation:
  max_context_tokens: 12000
  refuse_without_evidence: true
  high_risk_keywords:
    - drop
    - delete
    - truncate
    - rm -rf
cache:
  semantic_ttl_seconds: 300
  answer_ttl_seconds: 120

Key flags ensure that every answer comes from a single snapshot and that high‑risk operations are blocked unless explicitly allowed.

Source‑Code Chunking Strategy

# pipeline/source_chunker.py
@dataclass
class CodeChunk:
    chunk_id: str
    artifact_id: str
    repo: str
    path: str
    symbol: str
    language: str
    content: str
    summary: str
    start_line: int
    end_line: int
    snapshot_id: str
    acl_tags: list[str]
    metadata: dict

class SourceChunker:
    def __init__(self, parser_registry, summarizer):
        ...
    def chunk_file(self, repo: str, file_path: Path, snapshot_id: str, acl_tags: list[str]) -> Iterable[CodeChunk]:
        parser = self.parser_registry.for_path(file_path)
        tree = parser.parse(file_path.read_text())
        for symbol in tree.iter_symbols():
            if symbol.kind not in {"function", "method", "class", "handler"}:
                continue
            source = symbol.source_text()
            summary = self.summarizer.summarize_symbol(name=symbol.qualified_name,
                                                      docstring=symbol.docstring,
                                                      source=source)
            yield CodeChunk(
                chunk_id=f"{snapshot_id}:{repo}:{file_path}:{symbol.start_line}",
                artifact_id=f"{repo}:{file_path}",
                repo=repo,
                path=str(file_path),
                symbol=symbol.qualified_name,
                language=parser.language,
                content=source,
                summary=summary,
                start_line=symbol.start_line,
                end_line=symbol.end_line,
                snapshot_id=snapshot_id,
                acl_tags=acl_tags,
                metadata={"kind": symbol.kind, "imports": symbol.imports,
                          "calls": symbol.calls, "owners": symbol.owners},
            )

Chunks are created per function, class, or handler, preserving symbol boundaries and attaching a generated summary to improve vector retrieval.

Offline Snapshot Publishing

# pipeline/snapshot_publisher.py
class SnapshotPublisher:
    def __init__(self, evaluator, registry, router):
        ...
    def publish(self, snapshot_id: str) -> None:
        report = self.evaluator.evaluate(snapshot_id)
        if report.empty_hit_ratio > 0.18:
            raise RuntimeError("empty hit ratio exceeds threshold")
        if report.faithfulness_score < 0.82:
            raise RuntimeError("faithfulness score too low")
        self.registry.mark_ready(snapshot_id)
        self.router.shift_read_traffic(snapshot_id, strategy="blue_green")

Publishing aborts if quality metrics fall below thresholds, ensuring only vetted snapshots receive traffic.

Common Pitfalls and Mitigations

Code‑only indexing : treat source as plain text; fix by symbol‑level chunking and graph expansion.

Cross‑version contamination : enforce a single snapshot_id per request and display the snapshot version in answers.

Latency under load : cache frequent queries, limit re‑rank candidate size, apply per‑retriever timeouts, and use rate‑limiting/fuse at the LLM gateway.

Missing ACL enforcement : filter at retrieval time and strip unauthorized citations before rendering.

Kubernetes Deployment Blueprint

apiVersion: apps/v1
kind: Deployment
metadata:
  name: retrieval-coordinator
spec:
  replicas: 4
  selector:
    matchLabels:
      app: retrieval-coordinator
  template:
    metadata:
      labels:
        app: retrieval-coordinator
    spec:
      containers:
        - name: app
          image: registry.internal/kb/retrieval-coordinator:v2.4.1
          ports:
            - containerPort: 8080
          env:
            - name: ACTIVE_SNAPSHOT
              valueFrom:
                configMapKeyRef:
                  name: knowledge-engine
                  key: active_snapshot
          resources:
            requests:
              cpu: "2"
              memory: "4Gi"
            limits:
              cpu: "4"
              memory: "8Gi"
          readinessProbe:
            httpGet:
              path: /readyz
              port: 8080
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: retrieval-coordinator
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: retrieval-coordinator
  minReplicas: 4
  maxReplicas: 20
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 65

Key recommendations: separate embedding, re‑ranking, and generation into distinct services for independent scaling; use a snapshot controller for hot‑swap of active_snapshot without restarts; keep old snapshots for quick rollback.

Essential Monitoring Metrics

Retrieval empty‑result rate

Citation missing rate

Faithfulness score

Positive feedback vs. follow‑up rate

Query‑rewrite trigger frequency

P95/P99 latency for each retrieval sub‑pipeline

Quality drift after snapshot switch

Average token cost per query

These metrics let operators distinguish between "no evidence", "mis‑ranked evidence", and "generation drift".

Evolution Roadmap

Start with a single‑repo, single‑doc Q&A loop to validate user intent and evidence gaps.

Introduce source‑code parsing and multi‑index retrieval once questions target methods, state‑machine branches, or error codes.

Add snapshot publishing, automated evaluation, and blue‑green rollout for production stability.

Mature into an enterprise knowledge engine supporting on‑call assistance, onboarding, and cross‑team knowledge sharing.

The final takeaway is that the real challenge of RAG lies not in model size but in building a robust engineering loop that guarantees version‑consistent, permission‑aware, evidence‑backed answers at scale.