Scalable Knowledge Base with High‑Concurrency Crawling and Vector Search

Why a Knowledge‑Base Platform Is Needed

Many teams start with a simple pipeline: fetch web pages or PDFs, extract text, chunk, embed, and write vectors to Milvus or Elasticsearch, then expose a Q&A API. This works for small demos, but in real organizations the problem shifts from "can we retrieve?" to "can we retrieve reliably and trustably over time?".

Enterprise data sources are heterogeneous (internal wikis, CRM cases, product manuals, regulatory sites) and have common traits: varied formats, frequent updates, permission sensitivity, noisy quality, and complex retrieval goals (finding the right version). A single‑machine script fails in four ways:

Uncontrolled collection chain : incremental fetching, deduplication, rate‑limiting, and traceability become hard.

Unstable cleaning results : changing PDF templates break extraction, causing random vector quality.

Unexplained index visibility : missing steps (cleaning, chunking, indexing, version switch) make answers disappear.

RAG quality cannot be operated : stale context, wrong version, or permission errors lead to "sometimes correct, but not reliable" feedback.

Therefore the real need is a knowledge‑base middle platform that answers five questions:

How are new documents discovered, fetched, queued, and processed?

How are multiple updates of the same document versioned and tracked?

How can cleaning, chunking, and vectorisation failures be retried locally rather than re‑processing the whole corpus?

How are tenant, department, confidentiality, and version filters applied at retrieval time?

How does user feedback drive re‑crawling, re‑cleaning, and index back‑filling?

If you only need a static FAQ for a tiny team, a simple script suffices. Upgrade to a platform when any of the following holds:

Data sources add new items daily.

Document count grows to hundreds of thousands or millions.

Permission isolation across organizations or roles is required.

Business expects minute‑level visibility after a document change.

Auditable provenance of each answer is needed.

Overall Architecture: Microservices + Event‑Driven + State Governance

The system is not just "crawler → vector store → RAG". It consists of four responsibility layers:

Collection layer : discover changes, fetch raw files, deduplicate, schedule.

Processing layer : clean, extract structure, chunk, embed, build indexes.

Retrieval layer : hybrid recall, permission filtering, re‑ranking, answer generation.

Governance layer : versioning, state tracking, audit, feedback, rollout, gray‑scale switching.

┌────────────────────────────┐
                │      Source Registry       │
                │  数据源配置 / 抓取策略 / ACL │
                └────────────┬───────────────┘
                             │
                             ▼
┌───────────────┐           ┌────────────────────┐
│ Scheduler     │  task     │ Crawler Workers    │
│ 调度/限速/重试│──────────▶│ 拉取网页/PDF/附件 │
└──────┬────────┘           └──────────┬─────────┘
       │                                 │ raw file
       │ state                           ▼
       │                         ┌────────────────────┐
       │                         │ Object Storage     │
       │                         │ MinIO/S3 原始文档 │
       │                         └──────────┬─────────┘
       │                                   │ event
       ▼                                   ▼
┌──────────────────┐             ┌────────────────────┐
│ Ingest Event Bus │────────────▶│ Cleaner Pipeline    │
│ Kafka/Pulsar     │             │ 解析/去噪/结构标准化 │
└────────┬─────────┘             └──────────┬─────────┘
          │                                 │ cleaned doc
          │                                 ▼
          │                         ┌────────────────────┐
          │                         │ Chunk Builder       │
          │                         │ 标题切块/表格保留   │
          │                         └──────────┬─────────┘
          │                                   │ chunk batch
          ▼                                   ▼
┌──────────────────┐             ┌────────────────────┐
│ State Store      │◀────────────│ Embedding Workers  │
│ MySQL/Postgres   │ status      │ 批量向量化/幂等写入 │
└────────┬─────────┘             └──────────┬─────────┘
          │                                   │
          │                                   ├──────────────┐
          │                                   │              │
          ▼                                   ▼              ▼
┌──────────────────┐             ┌────────────────┐ ┌────────────────┐
│ Search Metadata  │             │ Vector Store   │ │ Sparse Index    │
│ 文档版本/ACL/状态 │             │ Milvus/pgvector│ │ ES/OpenSearch │
└────────┬─────────┘             └──────┬─────────┘ └──────┬─────────┘
          │                                 │            │
          └───────────────┬─────────────────┴────────────┬───────┘
                        ▼                              ▼
                ┌──────────────────────────────────────┐
                │ Retrieval Gateway                     │
                │ 权限过滤 / 混合召回 / 重排 / RAG    │
                └──────────────────────────────────────┘

The most easily overlooked yet critical component is state governance , not Kafka or Milvus.

2.1 Why a State Layer Is Mandatory

Many prototypes fail because each component only does its own job without a global view. A document’s lifecycle typically passes through nine explicit states, from source change detection to old‑version retirement. Modeling these as a state machine turns an "occasionally successful" system into an "operable" one.

Data source detects a change.

Scheduler creates a fetch task.

Fetcher stores raw file in object storage.

Cleaning succeeds, producing standardized text.

Chunking succeeds, generating a chunk set.

Vectorisation succeeds, writing to the vector index.

Sparse index succeeds, writing to the full‑text index.

New version is switched to queryable.

Old version is retired or kept for rollback.

2.2 Document Lifecycle Data Model

Two core tables are required: kb_document for document metadata and kb_document_job for per‑version processing jobs.

CREATE TABLE kb_document (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  tenant_id BIGINT NOT NULL,
  source_id BIGINT NOT NULL,
  external_doc_key VARCHAR(256) NOT NULL,
  title VARCHAR(512) NOT NULL,
  content_hash CHAR(64) NOT NULL,
  current_version INT NOT NULL DEFAULT 0,
  visibility VARCHAR(32) NOT NULL DEFAULT 'PRIVATE',
  acl_scope JSON NOT NULL,
  latest_job_id BIGINT NULL,
  created_at DATETIME NOT NULL,
  updated_at DATETIME NOT NULL,
  UNIQUE KEY uk_source_doc (tenant_id, source_id, external_doc_key)
);

CREATE TABLE kb_document_job (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  tenant_id BIGINT NOT NULL,
  document_id BIGINT NOT NULL,
  version_no INT NOT NULL,
  trigger_type VARCHAR(32) NOT NULL,
  status VARCHAR(32) NOT NULL,
  raw_object_key VARCHAR(512) NULL,
  clean_object_key VARCHAR(512) NULL,
  error_code VARCHAR(64) NULL,
  error_message VARCHAR(1024) NULL,
  retry_count INT NOT NULL DEFAULT 0,
  started_at DATETIME NULL,
  finished_at DATETIME NULL,
  created_at DATETIME NOT NULL,
  updated_at DATETIME NOT NULL,
  UNIQUE KEY uk_doc_version (document_id, version_no)
);

This model solves version tracking, independent job retries, safe index switches, and per‑version visibility.

2.3 Event‑Driven Over Direct Service Calls

Instead of a monolithic HTTP chain, each stage emits an event (e.g., raw.doc.created, doc.cleaned) that downstream workers consume. Benefits include independent scaling, isolated retries, and the ability to back‑pressure upstream when downstream queues back up.

Core Engineering Details

3.1 Distributed Crawling Scheduler

Enterprise crawlers must handle internal APIs, authenticated pages, recursive links, and periodic re‑crawls. The scheduler must address:

Priority (regulatory updates vs low‑value pages).

Deduplication at both URL and content levels.

Politeness and quota per domain, tenant, or system.

Explicit status reporting for success, failure, skip, or delay.

Double‑layer deduplication uses source_id + external_doc_key for task uniqueness and content_hash after cleaning to decide whether a new version is needed.

3.2 Cleaning Pipeline – Stability Over Cleverness

The goal is to preserve the semantic structures needed for downstream retrieval, not merely to extract raw text. Three semantic layers are retained:

Structural semantics (headings, tables, lists, references).

Source semantics (origin system, publish time, department).

Explainability semantics (page, section, version provenance).

Typical failure modes include double‑column PDFs producing interleaved text, tables flattened into long strings, and header/footer noise. The recommended cleaning stages are:

Format detection (PDF, DOCX, HTML, scanned images).

Structure restoration (retain headings, tables, sections).

Noise removal (strip headers/footers, ads, repeated boilerplate).

Semantic enrichment (add image captions, table explanations when needed).

Standardised output as Markdown or JSON with metadata.

Do not use LLMs to rewrite content during cleaning; this introduces audit risk and non‑deterministic chunks.

3.3 Chunking Strategy – Semantic Units, Not Fixed Tokens

Instead of blindly tuning token length, decide whether the model should see a "text fragment" or a "self‑contained knowledge unit". Typical units in an enterprise knowledge base are:

Policy clauses.

FAQ Q&A pairs.

Tables with explanations.

Procedural step groups.

Technical explanations with sub‑headings.

Chunk boundaries should follow semantic breaks (headings, list items, table rows) and avoid overly small fragments that lose context. Recommended heuristics:

Markdown/Wiki: chunk by heading hierarchy.

FAQ/Support tickets: chunk by Q&A pair.

Long reports: sliding window over sentences but keep chapter context.

Tables: treat the whole table as one chunk with header metadata.

Images: bind caption text to the adjacent paragraph.

3.4 Vectorisation & Hybrid Retrieval

Pure vector search cannot handle exact terms, structured filters, or stale versions. A production pipeline typically performs:

Query normalisation (spelling, synonym expansion, time extraction).

ACL and version filtering.

Sparse BM25 recall for keyword hits.

Dense embedding recall for semantic similarity.

Reciprocal Rank Fusion (RRF) to merge results.

Cross‑Encoder or business‑rule re‑ranking.

Context de‑duplication, conflict resolution, and final assembly for LLM input.

The engineering rule is: filter first, recall second, then generate.

3.5 Governance – The Real Bottleneck

When scale grows, hidden failures dominate: outdated versions staying live, missing ACL fields leaking data, non‑idempotent retries polluting indexes, silent cleaning failures, and feedback loops that never trigger re‑processing. A robust system records user feedback (likes, dislikes), links low‑quality answers to specific chunks and versions, and automatically triggers re‑crawling or manual review for problematic sources.

Key Code Implementations

4.1 Crawl Task Enqueue (Python)

# scheduler/enqueue.py
from __future__ import annotations
import hashlib, json
from dataclasses import dataclass
from datetime import datetime, timezone
import redis

r = redis.Redis(host="redis", port=6379, decode_responses=True)

@dataclass
class CrawlTask:
    tenant_id: int
    source_id: int
    external_doc_key: str
    url: str
    priority: int
    trigger_type: str
    trace_id: str

    @property
    def dedupe_key(self) -> str:
        raw = f"{self.tenant_id}:{self.source_id}:{self.external_doc_key}"
        return hashlib.sha256(raw.encode("utf-8")).hexdigest()

def enqueue_crawl(task: CrawlTask) -> bool:
    task_key = f"crawl:task:{task.dedupe_key}"
    queue_key = "crawl:queue"
    payload = {
        "tenant_id": task.tenant_id,
        "source_id": task.source_id,
        "external_doc_key": task.external_doc_key,
        "url": task.url,
        "priority": task.priority,
        "trigger_type": task.trigger_type,
        "trace_id": task.trace_id,
        "enqueued_at": datetime.now(timezone.utc).isoformat(),
    }
    with r.pipeline() as pipe:
        while True:
            try:
                pipe.watch(task_key)
                if pipe.exists(task_key):
                    pipe.unwatch()
                    return False
                pipe.multi()
                pipe.set(task_key, json.dumps(payload), ex=1800)
                pipe.zadd(queue_key, {json.dumps(payload): task.priority})
                pipe.execute()
                return True
            except redis.WatchError:
                continue

Key design points: idempotent key tenant_id+source_id+external_doc_key, short TTL for deduplication, and a trace ID propagated downstream.

4.2 Cleaning Worker State Machine (Python)

# cleaner/worker.py
from __future__ import annotations
import io, json
from contextlib import contextmanager
from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
from minio import Minio
from db import SessionLocal
from models import DocumentJob
from parser import parse_document, normalize_markdown

minio_client = Minio("minio:9000", access_key="replace-me", secret_key="replace-me", secure=False)

@contextmanager
def db_session():
    session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

def mark_job(session, job_id: int, status: str, **fields) -> DocumentJob:
    job = session.query(DocumentJob).filter(DocumentJob.id == job_id).with_for_update().one()
    job.status = status
    for key, value in fields.items():
        setattr(job, key, value)
    return job

async def run_cleaner():
    consumer = AIOKafkaConsumer("raw.doc.created", bootstrap_servers="kafka:9092")
    producer = AIOKafkaProducer(bootstrap_servers="kafka:9092")
    await consumer.start()
    await producer.start()
    try:
        async for msg in consumer:
            event = json.loads(msg.value)
            job_id = event["job_id"]
            try:
                with db_session() as session:
                    mark_job(session, job_id, "CLEANING")
                    response = minio_client.get_object("kb-raw", job.raw_object_key)
                    raw_bytes = response.read()
                parsed = parse_document(io.BytesIO(raw_bytes), event["file_type"])
                cleaned = normalize_markdown(parsed)
                clean_key = f"{event['tenant_id']}/{event['document_id']}/{event['version_no']}/clean.md"
                minio_client.put_object(
                    "kb-clean",
                    clean_key,
                    io.BytesIO(cleaned.encode("utf-8")),
                    length=len(cleaned.encode("utf-8")),
                    content_type="text/markdown",
                )
                with db_session() as session:
                    mark_job(session, job_id, "CLEANED", clean_object_key=clean_key, error_code=None, error_message=None)
                await producer.send_and_wait(
                    "doc.cleaned",
                    json.dumps({
                        "job_id": job_id,
                        "tenant_id": event["tenant_id"],
                        "document_id": event["document_id"],
                        "version_no": event["version_no"],
                        "clean_object_key": clean_key,
                    }).encode("utf-8"),
                )
            except Exception as exc:
                with db_session() as session:
                    mark_job(session, job_id, "CLEAN_FAILED", error_code=exc.__class__.__name__, error_message=str(exc)[:1000])
                raise
    finally:
        await consumer.stop()
        await producer.stop()

Important aspects: state transition before work, error codes persisted, and downstream event carries document_id + version_no + job_id for full traceability.

4.3 Chunk & Vector Ingestion (Python)

# embedding/ingest.py
from __future__ import annotations
import hashlib
from typing import Iterable
from pymilvus import Collection
from chunking import build_chunks
from embeddings import embed_texts

collection = Collection("kb_chunk_index")

def stable_chunk_id(document_id: int, version_no: int, chunk_seq: int) -> str:
    raw = f"{document_id}:{version_no}:{chunk_seq}"
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()

def ingest_document(document_id: int, version_no: int, markdown_text: str, tenant_id: int, acl_tags: list[str]) -> None:
    chunks = build_chunks(markdown_text)
    vectors = embed_texts([c.text for c in chunks], batch_size=64)
    entities = []
    for chunk_seq, (chunk, vector) in enumerate(zip(chunks, vectors)):
        entities.append({
            "chunk_id": stable_chunk_id(document_id, version_no, chunk_seq),
            "document_id": document_id,
            "version_no": version_no,
            "chunk_seq": chunk_seq,
            "tenant_id": tenant_id,
            "visibility_state": "STAGING",
            "acl_tags": acl_tags,
            "heading_path": " / ".join(chunk.heading_path),
            "text": chunk.text,
            "vector": vector,
        })
    collection.upsert(entities)

Two engineering rules: never treat the vector store as the source of truth (metadata lives in the relational DB) and never overwrite an active version; instead use a STAGING → READY → ACTIVE → RETIRED flow.

4.4 Retrieval Gateway (FastAPI)

# retrieval/gateway.py
from __future__ import annotations
from fastapi import FastAPI
from pydantic import BaseModel
from authz import build_acl_filter
from query_rewrite import normalize_query
from rerank import rerank_contexts
from search import dense_search, sparse_search, reciprocal_rank_fusion

app = FastAPI()

class AskRequest(BaseModel):
    tenant_id: int
    user_id: str
    department: str
    question: str

@app.post("/ask")
async def ask(req: AskRequest):
    normalized_query = normalize_query(req.question)
    acl_filter = build_acl_filter(
        tenant_id=req.tenant_id,
        department=req.department,
        user_id=req.user_id,
    )
    dense_hits = dense_search(query=normalized_query, top_k=30, acl_filter=acl_filter, visibility_state="ACTIVE")
    sparse_hits = sparse_search(query=normalized_query, top_k=30, acl_filter=acl_filter, visibility_state="ACTIVE")
    fused = reciprocal_rank_fusion(dense_hits, sparse_hits, k=60)
    ranked_contexts = rerank_contexts(normalized_query, fused[:20])[:8]
    return {
        "query": normalized_query,
        "contexts": [
            {
                "document_id": item.document_id,
                "version_no": item.version_no,
                "score": item.score,
                "heading_path": item.heading_path,
                "snippet": item.text[:240],
            }
            for item in ranked_contexts
        ],
    }

The gateway performs query normalisation, ACL filtering, dense + sparse recall, RRF fusion, and a second‑stage rerank before returning evidence snippets. Answer generation is deliberately left out to keep retrieval testable.

4.5 Index Switch (Version Activation)

After dense, sparse, and metadata indexes are ready, a separate switch task marks the version as ACTIVE, retires the previous version, emits a doc.version.activated event, and records the operator and rollback point. This guarantees that a version becomes visible atomically.

Production Pitfalls & Tuning

5.1 Kafka Back‑Pressure

Back‑pressure is needed when parsing cost spikes (e.g., scanned PDFs causing OCR blow‑up) or source schema changes. Metrics to monitor include topic lag for raw.doc.created, cleaning success rate, embedding worker queue time, and vector‑store write latency. When thresholds are crossed, automatically reduce low‑priority fetch quotas, throttle non‑core tenants, or pause high‑cost augmentations.

5.2 Vector Retrieval Latency Jitter

Latency jitter often stems from mismatched write‑read patterns, shared resource pools, or noisy low‑quality chunks. Mitigations: separate real‑time write nodes from query nodes, use micro‑batches for minute‑level visibility, prune noisy chunks, and apply metadata filtering before ANN search.

5.3 Model Inference Bottleneck

Embedding, OCR, image summarisation, query rewrite, and rerank all compete for GPU resources. Allocate dedicated GPU pools for offline batch jobs and a low‑latency pool for online Q&A. Provide independent circuit‑breakers for optional steps (image summarisation, query expansion).

5.4 MinIO Small‑File Explosion

Store only raw documents and cleaned versions per document version. Keep chunks in the index tables, not as separate objects. Enable lifecycle policies to delete temporary debugging artefacts.

5.5 Permission & Auditing

Enforce ACL filtering before retrieval, retain tenant/department/clearance tags at chunk level, optionally suppress answer generation for sensitive docs, mask user queries in logs, and always attach source document version and trace ID to generated answers.

Evolution Roadmap

Stage 1 – Single‑machine proof of concept : validate cleaning stability, chunk quality, and RAG value.

Stage 2 – Service decomposition : split into scheduler, cleaner, chunker/embedding, indexers, and retrieval gateway; enable independent scaling and replay.

Stage 3 – Multi‑tenant & permission layer : add tenant isolation, ACL filters, audit logs, per‑tenant quotas, version gray‑rollout.

Stage 4 – Feedback loop & knowledge ops : collect likes/dislikes, analyse hot questions, auto‑re‑process low‑quality docs, run offline evaluation suites.

Conclusion

Building an enterprise knowledge base is not about converting PDFs to vectors; it is about constructing a production‑grade data pipeline that guarantees versioned ingestion, stateful processing, fine‑grained permission enforcement, and a closed feedback loop. When these four pillars—version, state, permission, and governance—are solid, further optimisation of models, re‑ranking, and answer presentation yields sustainable value.