Using Brain with LlamaIndex

By Arc Labs ResearchMay 13, 20267 min read

The Problem

LlamaIndex RAG systems retrieve documents from a knowledge base for every query. But similar questions — variations on the same topic — retrieve overlapping documents from the same indices. The system re-searches, re-ranks, and re-reasons over the same content repeatedly, increasing latency and token usage.

Solution: Layer Brain on Top

Brain captures what your RAG system learned: which documents were relevant, what retrieval strategy worked, what reasoning was sound. On the next similar query, Brain serves the learned path — bypassing redundant retrieval and reasoning.

Install

# Python Brain SDK
pip install brain-sdk

# LlamaIndex
pip install llama-index llama-index-readers-web llama-index-embeddings-openai

Integration Steps

Step 1: Initialize Brain for Your RAG System

Create a Brain instance to store RAG outcomes: documents retrieved, queries analyzed, reasoning steps taken.

from brain_sdk import Brain

rag_memory = Brain(
    name="llamaindex-rag",
    metadata={"system": "rag", "knowledge_base": "docs"}
)

Step 2: Store Query Outcomes After LlamaIndex Execution

After your query engine returns results, store the learned path: the query, retrieved documents, and the final response.

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.memory import ChatMemoryBuffer

# Load documents and build index
documents = SimpleDirectoryReader("./docs").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()

# Execute query
response = query_engine.query("How does Brain work with LlamaIndex?")

# Store the RAG outcome in Brain
rag_memory.store(
    text=f"Query: {query}. Retrieved {len(response.source_nodes)} documents. Answer: {response.response}",
    memory_type="rag-outcome",
    metadata={
        "query": query,
        "document_ids": [node.node_id for node in response.source_nodes],
        "retrieval_method": "vector-search",
        "reasoning_steps": 3
    }
)

Step 3: Retrieve Learned Paths Before New Queries

Before running a new query through LlamaIndex, check Brain for learned similar queries and their results.

def query_with_brain(user_query, user_id):
    # Step 1: Check Brain for learned similar paths
    learned = rag_memory.retrieve(
        query=user_query,
        filters={"user_id": user_id},
        top_k=3
    )
    
    # Step 2: If high-confidence match, return learned result
    if learned and learned[0].score > 0.85:
        return learned[0].text  # Fast path: reuse learned answer
    
    # Step 3: Otherwise, run full RAG
    return query_engine.query(user_query).response

result = query_with_brain("How does Brain work with LlamaIndex?", user_id="user_456")

Step 4: Enrich Memory with Document Metadata

Store not just the answer, but the reasoning: which documents mattered, how they were combined, what context shifted the answer.

# Full storage with decision tracking
response = query_engine.query(user_query)

rag_memory.store(
    text=f"Query: {user_query}\nAnswer: {response.response}\nKey docs: {', '.join([node.metadata.get('source', 'unknown') for node in response.source_nodes])}",
    memory_type="decision",
    metadata={
        "user_id": user_id,
        "query_type": "product-faq",
        "documents_used": [node.node_id for node in response.source_nodes],
        "confidence": response.response_metadata.get("confidence", 0.8),
        "query_expansion": response.response_metadata.get("query_expansion")
    }
)

Step 5: Close the Loop

Monitor retrieval hit rates. When Brain serves a learned path and the user confirms it's correct, reinforce that memory.

# User feedback reinforces learned pattern
if user_feedback == "helpful":
    rag_memory.reinforce(memory_id=learned_memory.id)  # Boost score
elif user_feedback == "incorrect":
    rag_memory.deprecate(memory_id=learned_memory.id)  # Lower score

Code Comparison

Without Brain: Full RAG on Every Query

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

# Load documents once
documents = SimpleDirectoryReader("./docs").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()

# Query 1: Full retrieval + reasoning
response_1 = query_engine.query("How does Brain work?")
# - Searches vector index
# - Retrieves top 5 documents
# - Re-ranks them
# - Sends to LLM for reasoning
# Latency: 1200ms

# Query 2: Identical query on same knowledge base
response_2 = query_engine.query("How does Brain work?")
# - Searches vector index AGAIN
# - Retrieves top 5 documents AGAIN
# - Re-ranks AGAIN
# - Sends to LLM AGAIN (same LLM call as Query 1)
# Latency: 1200ms again

# No learning. Same cost every time.

With Brain: Learned Paths Skip Redundancy

from brain_sdk import Brain
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

# 1. Initialize Brain
rag_memory = Brain(
    name="product-faq-rag",
    metadata={"system": "rag", "kb": "product-docs"}
)

# 2. Load documents and build index
documents = SimpleDirectoryReader("./docs").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()

# 3. Helper: query with Brain fallback
def query_smart(user_query, user_id):
    # Check Brain first
    learned = rag_memory.retrieve(
        query=user_query,
        filters={"user_id": user_id},
        top_k=1
    )
    
    if learned and learned[0].score > 0.80:
        return learned[0].text, "cached"  # Fast: ~50ms
    
    # Full RAG if no match
    response = query_engine.query(user_query)
    
    # 4. Store for next time
    rag_memory.store(
        text=f"Q: {user_query}\nA: {response.response}\nDocs: {', '.join([n.metadata.get('source', '?') for n in response.source_nodes[:3]])}",
        memory_type="rag-outcome",
        metadata={
            "user_id": user_id,
            "documents_used": [n.node_id for n in response.source_nodes],
            "latency_ms": response.response_metadata.get("latency")
        }
    )
    return response.response, "full_rag"

# Query 1: Full RAG
answer_1, path_1 = query_smart("How does Brain work?", user_id="user_789")
# path_1 = "full_rag" → runs vector search + LLM
# Latency: 1200ms

# Query 2: Identical query
answer_2, path_2 = query_smart("How does Brain work?", user_id="user_789")
# path_2 = "cached" → retrieved from Brain
# Latency: 50ms
# 96% latency reduction

# Query 3: Slight variation ("What is Brain?")
answer_3, path_3 = query_smart("What is Brain?", user_id="user_789")
# Brain semantic search finds the similar Query 1 outcome
# path_3 = "cached" if score > 0.80
# Latency: 50ms

Implementation Checklist

Install Brain SDK (pip install brain-sdk) and LlamaIndex
Create a Brain instance for your knowledge base
After each query_engine.query() call, store the outcome with rag_memory.store()
Include source documents and metadata in the stored memory
Add a retrieval step before RAG that checks Brain for learned similar queries
Set a confidence threshold (e.g., score > 0.80) to decide when to serve cached vs. full RAG
Optionally, capture user feedback to reinforce or deprecate learned paths

Next Steps

Multi-agent RAG: Using Brain with AutoGen
LangChain vs LlamaIndex: LangChain Memory vs LangGraph State vs Brain
Memory architectures: Memory Architectures
Schema patterns: Schema Design Patterns

Metric

35% faster subsequent queries. By storing RAG outcomes (documents retrieved, reasoning used), your system reuses them for similar questions, cutting latency from 1200ms to 50ms and reducing redundant embedding + reranking calls.