RAG vs. GraphRAG Comparison Guide

Document: RAG_VS_GRAPHRAG_COMPARISON_GUIDE.md Version: 2.0.0 Last Updated: January 30, 2026 Scope: Retrieval Architecture, Knowledge Graph Integration, Hallucination Reduction, MIZ OKI Enhancement Roadmap

Executive Summary

This guide provides practical decision criteria for choosing between standard Vector RAG and GraphRAG retrieval strategies, with deep integration into MIZ OKI's existing dual-store retrieval architecture. It documents the current implementation, identifies improvement opportunities, and provides concrete enhancement blueprints.

Part 1: Core Concepts

Vector RAG

Retrieves by semantic similarity—finding the nearest embedding chunks to the query.

Strengths: - Fast and simple to implement - Excellent for "find the passage" tasks - Works well when answers live in a single chunk - Lower infrastructure complexity

Limitations: - Cannot reason across entities or time - Misses causal relationships - Struggles with multi-step inference

GraphRAG

Retrieves by relationships and paths in a Knowledge Graph, enabling who-did-what-when-why reasoning.

Strengths: - Multi-hop reasoning across entities - Causal and temporal awareness - Explainable retrieval paths - Handles complex entity relationships

Limitations: - Higher infrastructure complexity - Requires well-structured KG schema - More expensive to maintain

Part 2: Decision Rules

Rule 1: Choose Vector RAG When...

The question is a "find the passage" task where one hop of context suffices.

Examples: - Definitions and glossary lookups - Policy and procedure retrieval - Code/API documentation Q&A - FAQ-style support queries - Simple fact retrieval

Signal: Answers mostly live in a single chunk; no entity joins required.

Rule 2: Choose GraphRAG When...

The question is about links across entities, time, or causality.

Examples: - "Which campaigns drove uplift via audience X after event Y?" - "What led from A → B → C?" - "Who influences whom under condition Z?" - "Why did metric M change after action A?" - "Trace the attribution path from click to conversion"

Signals: - Multi-step paths required - Entity ↔ Event ↔ Metric joins needed - Explanations require "because..." reasoning - Temporal ordering matters

Rule 3: Choose Hybrid/Adaptive When...

The question requires both semantic context and structural relationships.

Examples: - "Find documents about campaign optimization that mention the audience segment we created last week" - "What's the best practice for bidding on keywords similar to our top performers?" - Complex analytical queries with both content and relationship aspects

Part 3: MIZ OKI Current Architecture

3.1 DualStoreRetriever (dual_store_retriever.py)

The DualStoreRetriever is MIZ OKI's core hybrid retrieval engine combining vector search and KG traversal.

Location: miz-oki-adk-agents/boss/dual_store_retriever.py (960 lines)

Retrieval Modes

class RetrievalMode(str, Enum):
    VECTOR_ONLY = "vector_only"   # Pure semantic search
    KG_ONLY = "kg_only"           # Pure graph traversal
    DUAL = "dual"                 # Both with fusion
    ADAPTIVE = "adaptive"         # Auto-select based on query

Fusion Strategies

class FusionStrategy(str, Enum):
    RECIPROCAL_RANK = "reciprocal_rank"     # RRF fusion (default)
    WEIGHTED_AVERAGE = "weighted_average"   # Fixed weight blend
    MAX_SCORE = "max_score"                 # Take highest score per result
    LEARNED = "learned"                     # ML-based fusion (future)

Current Implementation (Reciprocal Rank Fusion):

# From dual_store_retriever.py lines 420-445
def _fuse_results_rrf(self, vector_results: List, kg_results: List, k: int = 60) -> List:
    """
    Reciprocal Rank Fusion: score = Σ 1/(k + rank_i) across sources
    """
    scores = defaultdict(float)
    for rank, result in enumerate(vector_results):
        scores[result.id] += 1.0 / (k + rank + 1)
    for rank, result in enumerate(kg_results):
        scores[result.id] += 1.0 / (k + rank + 1)
    # Sort by fused score
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)

Graph-Aware Caching

The GraphAwareCache provides intelligent caching with edge-based invalidation:

class GraphAwareCache:
    """Cache with KG-aware invalidation.

    When an edge is updated in the KG, any cached paths
    traversing that edge are automatically invalidated.
    """
    def __init__(self, max_size: int = 10000, ttl_seconds: int = 3600):
        self.cache = {}
        self.edge_to_paths = defaultdict(set)  # Track which paths use which edges

    def invalidate_edge(self, edge_id: str):
        """Invalidate all cached paths containing this edge."""
        affected_paths = self.edge_to_paths.get(edge_id, set())
        for path_key in affected_paths:
            self.cache.pop(path_key, None)

3.2 StepwiseKGReasoner (stepwise_kg_reasoner.py)

Multi-paradigm KG reasoning with Chain-of-Thought (CoT), Tree-of-Thought (ToT), and Graph-of-Thought (GoT).

Location: miz-oki-adk-agents/boss/stepwise_kg_reasoner.py (1133 lines)

Reasoning Paradigms

class ReasoningParadigm(str, Enum):
    COT = "chain_of_thought"    # Linear step-by-step reasoning
    TOT = "tree_of_thought"     # Branching exploration with pruning
    GOT = "graph_of_thought"    # Full graph traversal with cycles

Node and Edge Types

class NodeType(str, Enum):
    ENTITY = "entity"           # Concrete entities (campaigns, products)
    CONCEPT = "concept"         # Abstract concepts
    EVENT = "event"             # Time-bounded occurrences
    METRIC = "metric"           # Measurable values
    DOCUMENT = "document"       # Text sources
    FACT = "fact"               # Verified statements

class EdgeType(str, Enum):
    IS_A = "is_a"               # Taxonomy
    CAUSES = "causes"           # Causal relationship
    RELATED_TO = "related_to"   # General association
    CONTAINS = "contains"       # Composition
    PRECEDES = "precedes"       # Temporal ordering
    ATTRIBUTED_TO = "attributed_to"  # Attribution

GraphRAG Extraction

The GraphRAGExtractor implements Microsoft-style community detection and summarization:

class GraphRAGExtractor:
    """Extract community summaries for hierarchical retrieval."""

    async def extract_communities(self, subgraph: KnowledgeGraph) -> List[Community]:
        """
        1. Run Leiden clustering on subgraph
        2. Generate summary for each community
        3. Create hierarchical index for retrieval
        """
        communities = self._leiden_cluster(subgraph)
        for community in communities:
            community.summary = await self._generate_summary(community.nodes)
        return communities

3.3 CausalQueryRouter (causal_query_router.py)

Intelligent routing of causal queries to specialized expert cells.

Location: miz-oki-adk-agents/boss/causal_query_router.py (905 lines)

10 Causal Query Types

class CausalQueryType(str, Enum):
    CAUSE_ATTRIBUTION = "cause_attribution"       # "What caused X?"
    COUNTERFACTUAL = "counterfactual"             # "What if X hadn't happened?"
    INTERVENTION = "intervention"                 # "What happens if we do X?"
    MEDIATION = "mediation"                       # "Does X affect Y through Z?"
    EFFECT_PREDICTION = "effect_prediction"       # "What will happen if X?"
    CAUSAL_DISCOVERY = "causal_discovery"         # "What are the causal relationships?"
    ATE_ESTIMATION = "ate_estimation"             # Average Treatment Effect
    CATE_ESTIMATION = "cate_estimation"           # Conditional ATE
    MECHANISM_EXPLANATION = "mechanism_explanation" # "How does X cause Y?"
    TEMPORAL_CAUSATION = "temporal_causation"     # Time-ordered causality

Dual-Stage Classification

class QueryClassifier:
    """Two-stage query classification for routing."""

    async def classify(self, query: str) -> ClassificationResult:
        # Stage 1: Pattern matching (fast)
        pattern_match = self._pattern_classify(query)
        if pattern_match.confidence > 0.85:
            return pattern_match

        # Stage 2: Embedding similarity (accurate)
        query_embedding = await self._embed_query(query)
        return self._embedding_classify(query_embedding)

Pattern Keywords by Query Type:

Expert Cell Routing

class ExpertCell(str, Enum):
    CELL_03 = "cell03"      # KG Brain - primary knowledge graph
    CELL_05 = "cell05"      # Causal inference specialist
    CELL_13 = "cell13"      # Causal API service
    CELL_23 = "cell23"      # Causal discovery
    BOSS_AGENT = "boss_agent"  # Fallback orchestrator

# Routing matrix: query_type -> [primary, secondary, fallback]
ROUTING_MATRIX = {
    CausalQueryType.CAUSE_ATTRIBUTION: [ExpertCell.CELL_05, ExpertCell.CELL_03],
    CausalQueryType.COUNTERFACTUAL: [ExpertCell.CELL_05, ExpertCell.CELL_13],
    CausalQueryType.INTERVENTION: [ExpertCell.CELL_13, ExpertCell.CELL_05],
    CausalQueryType.CAUSAL_DISCOVERY: [ExpertCell.CELL_23, ExpertCell.CELL_03],
    # ...
}

Part 4: Retrieval Hygiene for Hallucination Reduction

These practices apply to both RAG approaches and significantly reduce hallucinations.

1. Tight Scopes

Filter by org/project/time/window before embedding search.

# MIZ OKI Implementation in DualStoreRetriever
async def search(self, query: str, filters: Optional[Dict] = None) -> List[Result]:
    # Apply pre-filters before any retrieval
    effective_filters = {
        "org_id": self.current_org,
        "time_window": filters.get("time_window", self.default_window),
        "entity_types": filters.get("entity_types", None),
        **filters
    }
    return await self._execute_search(query, effective_filters)

2. Typed Metadata

3. De-duplication and Chunking

# Current deduplication in DualStoreRetriever
def _deduplicate_results(self, results: List[Result]) -> List[Result]:
    seen_ids = set()
    deduped = []
    for result in results:
        content_hash = hashlib.md5(result.content.encode()).hexdigest()
        if content_hash not in seen_ids:
            seen_ids.add(content_hash)
            deduped.append(result)
    return deduped

4. Attribution-On Policy

Always return source IDs and snippets alongside the context:

# Required response format for all retrieval operations
@dataclass
class RetrievalResponse:
    answer: str
    sources: List[Source]
    confidence: float
    reasoning_path: Optional[List[str]] = None  # For GraphRAG

@dataclass
class Source:
    doc_id: str
    snippet: str
    confidence: float
    node_id: Optional[str] = None  # KG node reference
    edge_path: Optional[List[str]] = None  # For multi-hop

5. Evidence-Based Ranking

# Current ranking formula in DualStoreRetriever
def compute_final_score(self, result: Result) -> float:
    return (
        result.similarity_score * 0.4 +
        result.recency_weight * 0.2 +
        result.authority_score * 0.2 +
        result.confidence * 0.2
    )

6. Causal Guardrails (GraphRAG-specific)

Only traverse edges with proper metadata:

Part 5: Improvement Proposals

Based on analysis of the current implementation, these are high-priority enhancements:

5.1 Adaptive Weight Learning (HIGH PRIORITY)

Gap: Current fusion uses fixed weights (0.5 vector, 0.5 KG). Weights should adapt based on query type and historical performance.

Current State:

# dual_store_retriever.py - Fixed weights
self.vector_weight = 0.5
self.kg_weight = 0.5

Proposed Enhancement:

class AdaptiveWeightLearner:
    """Learn optimal fusion weights per query type."""

    def __init__(self):
        self.weights_by_query_type: Dict[str, Tuple[float, float]] = {}
        self.outcome_buffer: List[QueryOutcome] = []

    async def get_weights(self, query: str, query_type: CausalQueryType) -> Tuple[float, float]:
        """Get learned weights for this query type."""
        if query_type in self.weights_by_query_type:
            return self.weights_by_query_type[query_type]

        # Default weights by query type
        defaults = {
            CausalQueryType.CAUSE_ATTRIBUTION: (0.3, 0.7),  # Favor KG
            CausalQueryType.COUNTERFACTUAL: (0.2, 0.8),     # Strong KG
            "factual": (0.7, 0.3),                          # Favor vector
            "mixed": (0.5, 0.5),                            # Balanced
        }
        return defaults.get(query_type, (0.5, 0.5))

    async def record_outcome(self, query: str, query_type: str,
                             vector_contributed: float, kg_contributed: float,
                             user_feedback: float):
        """Record retrieval outcome for learning."""
        self.outcome_buffer.append(QueryOutcome(
            query_type=query_type,
            vector_contribution=vector_contributed,
            kg_contribution=kg_contributed,
            feedback_score=user_feedback
        ))

        # Batch update weights periodically
        if len(self.outcome_buffer) >= 100:
            await self._update_weights()

    async def _update_weights(self):
        """Update weights using gradient descent on feedback."""
        for query_type in set(o.query_type for o in self.outcome_buffer):
            outcomes = [o for o in self.outcome_buffer if o.query_type == query_type]

            # Simple gradient: increase weight for source that correlated with positive feedback
            vector_gradient = sum(o.vector_contribution * o.feedback_score for o in outcomes)
            kg_gradient = sum(o.kg_contribution * o.feedback_score for o in outcomes)

            total = vector_gradient + kg_gradient
            if total > 0:
                self.weights_by_query_type[query_type] = (
                    vector_gradient / total,
                    kg_gradient / total
                )

        self.outcome_buffer = []

MCP Tool:

# New MCP tool for adaptive weights
async def retriever_set_adaptive_weights(
    query_type: str,
    vector_weight: float,
    kg_weight: float
) -> Dict:
    """Configure adaptive fusion weights for a query type."""

5.2 Cross-Store Reranking (HIGH PRIORITY)

Gap: Current RRF fusion treats both sources equally. Need reranking that considers semantic coherence between vector and KG results.

Proposed Enhancement:

class CrossStoreReranker:
    """Rerank fused results considering cross-store coherence."""

    def __init__(self, embedding_model: str = "text-embedding-3-large"):
        self.embedder = EmbeddingModel(embedding_model)

    async def rerank(
        self,
        query: str,
        vector_results: List[Result],
        kg_results: List[Result],
        top_k: int = 10
    ) -> List[RankedResult]:
        """
        Rerank considering:
        1. Query-result similarity
        2. Cross-source agreement (results found in both get boost)
        3. Path coherence for KG results
        4. Novelty penalty for redundant results
        """
        query_embedding = await self.embedder.embed(query)

        all_results = []
        seen_content = set()

        for result in vector_results + kg_results:
            # Skip near-duplicates
            content_sig = self._signature(result.content)
            if content_sig in seen_content:
                continue
            seen_content.add(content_sig)

            # Compute rerank score
            score = await self._compute_rerank_score(
                query_embedding,
                result,
                vector_results,
                kg_results
            )
            all_results.append(RankedResult(result=result, score=score))

        return sorted(all_results, key=lambda x: x.score, reverse=True)[:top_k]

    async def _compute_rerank_score(
        self,
        query_emb: np.ndarray,
        result: Result,
        vector_results: List[Result],
        kg_results: List[Result]
    ) -> float:
        # Base similarity
        result_emb = await self.embedder.embed(result.content)
        similarity = cosine_similarity(query_emb, result_emb)

        # Cross-source agreement bonus
        in_both = (
            any(r.id == result.id for r in vector_results) and
            any(r.id == result.id for r in kg_results)
        )
        agreement_bonus = 0.15 if in_both else 0.0

        # Path coherence for KG results
        path_coherence = 0.0
        if hasattr(result, 'path') and result.path:
            path_coherence = self._compute_path_coherence(result.path)

        return similarity * 0.6 + agreement_bonus + path_coherence * 0.25

5.3 Query-Expert Affinity Matrix (HIGH PRIORITY)

Gap: Static routing matrix doesn't learn from outcomes. Need dynamic affinity learning.

Current State:

# causal_query_router.py - Static routing
ROUTING_MATRIX = {
    CausalQueryType.CAUSE_ATTRIBUTION: [ExpertCell.CELL_05, ExpertCell.CELL_03],
    # ...
}

Proposed Enhancement:

class QueryExpertAffinityMatrix:
    """Learn and maintain query-expert affinity scores."""

    def __init__(self):
        # Initialize with prior beliefs
        self.affinity: Dict[Tuple[CausalQueryType, ExpertCell], float] = {}
        self.success_counts: Dict[Tuple[CausalQueryType, ExpertCell], int] = defaultdict(int)
        self.total_counts: Dict[Tuple[CausalQueryType, ExpertCell], int] = defaultdict(int)

    def get_routing(self, query_type: CausalQueryType, top_k: int = 3) -> List[ExpertCell]:
        """Get top-k experts for query type, ranked by affinity."""
        affinities = []
        for expert in ExpertCell:
            key = (query_type, expert)
            # Thompson sampling: sample from Beta distribution
            alpha = self.success_counts[key] + 1
            beta = self.total_counts[key] - self.success_counts[key] + 1
            sample = np.random.beta(alpha, beta)
            affinities.append((expert, sample))

        affinities.sort(key=lambda x: x[1], reverse=True)
        return [e for e, _ in affinities[:top_k]]

    def record_outcome(
        self,
        query_type: CausalQueryType,
        expert: ExpertCell,
        success: bool,
        response_quality: float  # 0-1 score
    ):
        """Update affinity based on outcome."""
        key = (query_type, expert)
        self.total_counts[key] += 1
        if success and response_quality > 0.7:
            self.success_counts[key] += 1

5.4 Intelligent Path Pruning (MEDIUM PRIORITY)

Gap: Current graph traversal explores all paths up to max_hops. Need intelligent pruning based on relevance.

Proposed Enhancement:

class IntelligentPathPruner:
    """Prune low-relevance paths during graph traversal."""

    def __init__(self, relevance_threshold: float = 0.3):
        self.threshold = relevance_threshold
        self.pruning_stats = {"pruned": 0, "kept": 0}

    async def should_expand(
        self,
        current_path: List[str],
        next_edge: Edge,
        query_embedding: np.ndarray
    ) -> bool:
        """Decide whether to expand this path."""
        # Check edge confidence
        if next_edge.confidence < self.threshold:
            self.pruning_stats["pruned"] += 1
            return False

        # Check semantic relevance of path so far
        path_text = self._path_to_text(current_path + [next_edge])
        path_embedding = await self._embed(path_text)
        relevance = cosine_similarity(query_embedding, path_embedding)

        if relevance < self.threshold:
            self.pruning_stats["pruned"] += 1
            return False

        self.pruning_stats["kept"] += 1
        return True

    def _path_to_text(self, path: List) -> str:
        """Convert path to natural language for embedding."""
        parts = []
        for i, item in enumerate(path):
            if isinstance(item, Node):
                parts.append(item.name)
            elif isinstance(item, Edge):
                parts.append(f"--[{item.type}]-->")
        return " ".join(parts)

5.5 Temporal Reasoning Enhancement (MEDIUM PRIORITY)

Gap: Limited support for "before/after" and time-bounded queries.

Proposed Enhancement:

class TemporalReasoner:
    """Enhanced temporal reasoning for time-bounded queries."""

    TEMPORAL_PATTERNS = [
        (r"before (\d{4}-\d{2}-\d{2})", "before"),
        (r"after (\d{4}-\d{2}-\d{2})", "after"),
        (r"between (.+) and (.+)", "between"),
        (r"in the last (\d+) (days?|weeks?|months?)", "relative"),
        (r"since (.+)", "since"),
    ]

    async def extract_temporal_constraints(self, query: str) -> TemporalConstraint:
        """Extract temporal constraints from query."""
        for pattern, constraint_type in self.TEMPORAL_PATTERNS:
            match = re.search(pattern, query, re.IGNORECASE)
            if match:
                return self._parse_constraint(constraint_type, match)
        return TemporalConstraint(type="none")

    async def filter_by_temporal(
        self,
        results: List[Result],
        constraint: TemporalConstraint
    ) -> List[Result]:
        """Filter results by temporal constraint."""
        if constraint.type == "none":
            return results

        filtered = []
        for result in results:
            if self._satisfies_temporal(result, constraint):
                filtered.append(result)
        return filtered

    async def order_by_causality(
        self,
        path: List[Node],
        edges: List[Edge]
    ) -> List[Tuple[Node, Edge]]:
        """Order path elements by causal/temporal sequence."""
        # Sort by timestamp, respecting PRECEDES edges
        timestamped = []
        for i, node in enumerate(path):
            ts = self._get_timestamp(node, edges[i] if i < len(edges) else None)
            timestamped.append((node, edges[i] if i < len(edges) else None, ts))

        return sorted(timestamped, key=lambda x: x[2] or datetime.max)

5.6 Pattern Mining Pipeline (LOW PRIORITY)

Gap: No automated discovery of common query patterns for optimization.

Proposed Enhancement:

class QueryPatternMiner:
    """Mine common query patterns for optimization."""

    def __init__(self, min_support: float = 0.05):
        self.min_support = min_support
        self.patterns: List[QueryPattern] = []
        self.query_log: List[QueryLogEntry] = []

    async def mine_patterns(self, lookback_days: int = 30) -> List[QueryPattern]:
        """Mine frequent patterns from query log."""
        # Load recent queries
        queries = await self._load_query_log(lookback_days)

        # Extract features
        feature_sequences = [self._extract_features(q) for q in queries]

        # Run frequent pattern mining
        patterns = self._apriori(feature_sequences, self.min_support)

        # Rank by support and performance correlation
        ranked = []
        for pattern in patterns:
            support = self._compute_support(pattern, feature_sequences)
            perf_correlation = await self._compute_performance_correlation(pattern)
            ranked.append(QueryPattern(
                features=pattern,
                support=support,
                performance_impact=perf_correlation
            ))

        return sorted(ranked, key=lambda x: x.support * x.performance_impact, reverse=True)

    def _extract_features(self, query: QueryLogEntry) -> List[str]:
        """Extract feature set from query."""
        features = []
        features.append(f"type:{query.query_type}")
        features.append(f"mode:{query.retrieval_mode}")
        features.append(f"hops:{query.max_hops}")
        if query.entity_types:
            for et in query.entity_types:
                features.append(f"entity:{et}")
        return features

Part 6: Implementation Roadmap

Phase 1: Foundation Enhancements (Weeks 1-2)

Phase 2: Advanced Retrieval (Weeks 3-4)

Phase 3: Learning & Optimization (Weeks 5-6)

Part 7: MCP Tools for Retrieval

Existing Tools

Proposed New Tools

Part 8: Best Practices Summary

Part 9: Virtuoso Routing Integration

The Boss Agent's Virtuoso routing leverages this guide:

# In boss_agent_core.py ACTION_KEYWORDS
ACTION_KEYWORDS = {
    # Existing keywords...
    "rag_retrieval": ["find", "lookup", "search for", "what is", "definition"],
    "graphrag_retrieval": ["trace", "path", "caused", "led to", "why", "how connected"],
    "hybrid_retrieval": ["analyze", "investigate", "comprehensive", "full picture"],
    "causal_query": ["because", "resulted in", "influenced", "what if", "intervene"],
}

Routing Decision Tree:

Query arrives
    │
    ├─ Contains causal keywords? ──────► CausalQueryRouter.classify()
    │       │                                    │
    │       │                                    ├─ CAUSE_ATTRIBUTION ─► Cell 05
    │       │                                    ├─ COUNTERFACTUAL ────► Cell 05
    │       │                                    ├─ INTERVENTION ──────► Cell 13
    │       │                                    └─ other ─────────────► KG Brain
    │       │
    ├─ Simple factual query? ──────────► DualStoreRetriever.VECTOR_ONLY
    │
    ├─ Relationship query? ────────────► StepwiseKGReasoner (auto-paradigm)
    │
    └─ Complex/unknown? ───────────────► DualStoreRetriever.ADAPTIVE

References

Internal Documentation

• MIZ OKI Knowledge Graph Framework: docs/KNOWLEDGE_GRAPH_FRAMEWORK_REPORT.md
• Boss Agent Architecture: docs/BOSS_AGENT_ARCHITECTURE_V5.md
• Causal Reasoning Systems: docs/archive/2025-12-24-session.md

Implementation Files

External Research

• Microsoft GraphRAG: Community-based summarization for hierarchical retrieval
• Reciprocal Rank Fusion: Cormack et al., SIGIR 2009
• Thompson Sampling for Multi-Armed Bandits: Russo et al., 2018

Version History