High‑Concurrency RAG: When to Use Classic, Graph or Agentic Architecture

Problem Overview

When a Retrieval‑Augmented Generation (RAG) service moves from a low‑traffic demo to a high‑concurrency production environment (e.g., customer service, risk control, after‑sales knowledge bases), the failure modes shift from model‑centric issues to system‑centric problems such as latency spikes, fragmented answers, incomplete multi‑step reasoning, and uncontrolled cost.

Typical Failure Types

Type 1: Retrieval returns relevant fragments but none of them cover the whole task (e.g., a query about tax differences between Hong Kong and South‑China warehouses requires warehouse‑level, tax‑policy, and status information).

Type 2: Knowledge is constantly updated, yet the index remains stale, causing answer drift.

Type 3: The user request becomes a multi‑step task (e.g., locate the nearest return point, verify supplier qualifications, generate a work‑order summary) that a plain RAG pipeline cannot satisfy.

Core Insight

RAG is not the hard part; turning a demo into a production‑ready pipeline is. The article proposes a unified engineering framework that decides, based on request characteristics, whether to use Classic RAG, Graph RAG, or Agentic RAG.

Four Production Pillars

Online Chain : request normalization, ACL filtering, vector retrieval, reranking, context assembly, LLM generation, answer citation.

Offline Chain : document change detection, chunking, entity & relation extraction, graph merge, embedding, community detection, summary generation, vector index build, snapshot publish.

Reasoning Orchestration : decide when a fixed pipeline suffices, when to invoke graph traversal, and when a stateful agent is required.

State Governance : versioned knowledge, cache keys bound to tenant_id, knowledge_base_id, index_version, acl_scope_hash, and snapshot version; token budgets, step limits, and graceful degradation.

Classic RAG – The Fast Path

Classic RAG excels at high‑frequency, low‑ambiguity FAQ‑style queries. Production checklist includes:

Query standardization (extract tenant, language, domain, time window).

Pre‑filter ACL before retrieval.

Multi‑layer caching: query‑normalize cache, embedding cache, retrieval result cache, rerank cache, final answer cache.

Cache keys must contain tenant_id, knowledge_base_id, index_version, rewrite_version, and acl_scope_hash to avoid cross‑tenant leakage.

Prefer caching at the query level for hot questions and at the embedding level for long‑tail queries.

Example Retrieval Service (Python)

from dataclasses import dataclass
from typing import List
import hashlib, orjson

@dataclass
class RetrievalRequest:
    tenant_id: str
    knowledge_base_id: str
    user_id: str
    query: str
    index_version: str
    top_k: int = 8

class RetrievalService:
    def __init__(self, vector_store, acl_service, reranker, redis_client):
        self.vector_store = vector_store
        self.acl_service = acl_service
        self.reranker = reranker
        self.redis = redis_client

    async def retrieve(self, req: RetrievalRequest) -> List[dict]:
        acl_scope = await self.acl_service.resolve_scope(
            tenant_id=req.tenant_id,
            user_id=req.user_id,
            knowledge_base_id=req.knowledge_base_id,
        )
        cache_key = self._cache_key(req, acl_scope.scope_hash)
        cached = await self.redis.get(cache_key)
        if cached:
            return orjson.loads(cached)
        embedding = await self._get_or_build_embedding(req.query)
        raw_hits = await self.vector_store.search(
            embedding=embedding,
            top_k=req.top_k * 3,
            filters={
                "tenant_id": req.tenant_id,
                "knowledge_base_id": req.knowledge_base_id,
                "index_version": req.index_version,
                "doc_acl_tags": {"$in": acl_scope.allowed_tags},
                "status": "published",
            },
        )
        reranked = await self.reranker.rank(req.query, raw_hits)
        final_chunks = self._assemble_context(reranked[: req.top_k])
        await self.redis.setex(cache_key, 300, orjson.dumps(final_chunks))
        return final_chunks

    async def _get_or_build_embedding(self, query: str) -> List[float]:
        key = "emb:" + hashlib.sha256(query.encode("utf-8")).hexdigest()
        cached = await self.redis.get(key)
        if cached:
            return orjson.loads(cached)
        vector = await embedding_client.embed(query)
        await self.redis.setex(key, 1800, orjson.dumps(vector))
        return vector

    def _assemble_context(self, hits: List[dict]) -> List[dict]:
        merged = []
        seen = set()
        for hit in hits:
            dedupe_key = (hit["doc_id"], hit["section_id"])
            if dedupe_key in seen:
                continue
            seen.add(dedupe_key)
            merged.append({
                "doc_id": hit["doc_id"],
                "title": hit["title"],
                "content": hit["content"],
                "score": hit["score"],
                "source_url": hit["source_url"],
            })
        return merged

    def _cache_key(self, req: RetrievalRequest, scope_hash: str) -> str:
        raw = "|".join([
            req.tenant_id,
            req.knowledge_base_id,
            req.index_version,
            scope_hash,
            req.query,
        ])
        return "rag:retrieve:" + hashlib.sha256(raw.encode("utf-8")).hexdigest()

Graph RAG – Adding Structured Reasoning

When a query requires multi‑hop entity relationships (e.g., “Which supplier belongs to which region and does its tax rule conflict with the standard policy?”), Classic RAG alone cannot provide a complete answer. Graph RAG runs parallel text retrieval and graph traversal, then merges the results.

Entity extraction must be versioned and normalized (dictionary mapping, rule‑based normalization, human whitelist).

Edges store provenance: source chunk ID, extraction model version, confidence, and snapshot version.

Community summaries are versioned alongside the graph to avoid stale explanations.

Snapshot publishing ties vector index, graph snapshot, and community summary under a single snapshot_version for consistent reads.

Incremental Graph Ingestion (Python)

from dataclasses import dataclass
from typing import Iterable

@dataclass
class ChangeEvent:
    doc_id: str
    chunk_id: str
    tenant_id: str
    snapshot_version: str
    content: str

class GraphIngestionPipeline:
    def __init__(self, entity_extractor, graph_repo, vector_repo, summary_repo):
        self.entity_extractor = entity_extractor
        self.graph_repo = graph_repo
        self.vector_repo = vector_repo
        self.summary_repo = summary_repo

    async def handle_events(self, events: Iterable[ChangeEvent]):
        for event in events:
            extraction = await self.entity_extractor.extract(
                text=event.content,
                snapshot_version=event.snapshot_version,
            )
            await self.graph_repo.merge_chunk_entities(
                tenant_id=event.tenant_id,
                chunk_id=event.chunk_id,
                snapshot_version=event.snapshot_version,
                entities=extraction.entities,
                relations=extraction.relations,
            )
            await self.vector_repo.upsert_chunk(
                tenant_id=event.tenant_id,
                chunk_id=event.chunk_id,
                snapshot_version=event.snapshot_version,
                content=event.content,
            )
            affected_communities = await self.graph_repo.find_affected_communities(
                snapshot_version=events[-1].snapshot_version
            )
            for community in affected_communities:
                summary = await self.summary_repo.build_summary(community)
                await self.graph_repo.save_community_summary(
                    community_id=community.community_id,
                    snapshot_version=community.snapshot_version,
                    summary=summary,
                )

Agentic RAG – Controlled Multi‑Step Execution

Agentic RAG is reserved for tasks that need sequential tool calls, dynamic planning, and stateful reasoning (e.g., retrieve policy, check ERP order status, then draft a handling suggestion). The key is a deterministic state machine rather than an open‑ended “thinking” loop.

Maximum steps, per‑step timeout, and token budget are hard limits.

Tool whitelist and input validation prevent accidental privileged actions.

All intermediate results are persisted (Redis for hot state, object storage for audit logs).

Failure paths include explicit statuses: success, failed, timeout, skipped.

Agent Session & Runtime Definition (Python)

from typing import Literal, TypedDict, List

class AgentStep(TypedDict):
    step_id: str
    action: str
    tool_name: str
    tool_input: dict
    tool_output: dict
    status: Literal["success", "failed", "timeout", "skipped"]

class AgentSession(TypedDict):
    session_id: str
    tenant_id: str
    user_id: str
    snapshot_version: str
    budget_tokens: int
    used_tokens: int
    max_steps: int
    current_step: int
    steps: List[AgentStep]
    final_answer: str
    state: Literal["running", "waiting_human", "finished", "failed"]

class AgentRuntime:
    async def run(self, session: AgentSession) -> AgentSession:
        while session["current_step"] < session["max_steps"]:
            plan = await planner.plan(session)
            tool = tool_registry.resolve(plan["tool_name"])
            validated_input = tool.validate_input(plan["tool_input"])
            result = await tool.invoke(validated_input, timeout_ms=tool.timeout_ms)
            session["steps"].append({
                "step_id": plan["step_id"],
                "action": plan["action"],
                "tool_name": plan["tool_name"],
                "tool_input": validated_input,
                "tool_output": result,
                "status": "success",
            })
            session["current_step"] += 1
            session["used_tokens"] += plan["estimated_tokens"]
            if await verifier.enough(session):
                session["final_answer"] = await summarizer.finalize(session)
                session["state"] = "finished"
                return session
        if session["used_tokens"] >= session["budget_tokens"]:
            session["state"] = "failed"
            session["final_answer"] = "budget exceeded"
            return session
        session["state"] = "failed"
        session["final_answer"] = "max steps exceeded"
        return session

Governance & State Consistency

Production RAG must treat knowledge, graph, and agent state as a single versioned entity. A snapshot registry records snapshot_version and all components (vector index, graph, community summary) are read atomically from that snapshot. Cache keys embed the snapshot version, and any cache miss forces a fresh read from the current snapshot.

Cache invalidation on knowledge updates: include snapshot_version in the key, publish a new snapshot, and let old caches expire naturally.

Graph‑summary sync: regenerate community summaries only for affected sub‑graphs, not the whole graph.

Agent session persistence: store hot state in Redis, archive full step logs to object storage, and keep a relational record of session metadata for audit.

Budget governance: enforce per‑session token caps, per‑tool timeouts, and global concurrency limits to protect core resources.

Production Checklist

Online Chain

Query classification & standardization.

Pre‑filter ACL before retrieval.

Separate timeout budgets for retrieval, rerank, and generation.

Define degradation paths for Classic, Graph, and Agentic routes.

Offline Pipeline

Clear chunking rules and incremental update support.

Snapshot‑based publishing with rollback capability.

State Governance

Cache keys bound to knowledge version.

Graph and summary share the same snapshot ID.

Agent sessions are recoverable after pod restarts.

Token, step, and tool budgets are enforced.

Observability

Per‑stage latency breakdown.

Version‑driven answer quality metrics.

Agent step audit logs.

Feedback loop for answer quality.

Evolution Roadmap

Stage 1 – Classic RAG Ops: stable offline chunking, ACL, caching, versioned indexes, monitoring.

Stage 2 – Graph RAG: add graph construction for queries that Classic RAG cannot answer fully.

Stage 3 – Agentic RAG: enable low‑volume, high‑value multi‑step tasks with strict governance.

Decision Matrix (Simplified)

Dimension – Classic RAG : solves high‑frequency FAQ; lowest latency; low‑to‑medium engineering complexity; medium offline build cost; medium governance difficulty; default entry point yes; strong state management needed.

Dimension – Graph RAG : solves relation reasoning; medium latency; medium‑to‑high engineering complexity; high offline build cost; high governance difficulty; not a default entry point; more state management needed.

Dimension – Agentic RAG : solves multi‑step tasks; highest latency; high engineering complexity; medium offline build cost; very high governance difficulty; not a default entry point; most state management needed.

Conclusion

In high‑concurrency production, the bottleneck moves from model capability to system reliability. Classic RAG provides the fast, stable base; Graph RAG adds structured reasoning; Agentic RAG handles complex, multi‑tool workflows. Success depends on four fundamentals: a short and stable online path, an incremental offline pipeline, clear boundaries for reasoning orchestration, and robust state governance that enables gray‑scale releases, rollbacks, and auditability.