RAG Agent

RAGAgent (Lewis et al., 2020) retrieves relevant chunks before generating an answer, grounding the response in supplied documents rather than parametric memory alone.

Usage

from brain.patterns import RAGAgent

# Simple keyword retriever
corpus = ["The API uses OAuth2.", "Rate limit: 100 req/min."]
retriever = lambda q, k=3: corpus[:k]

agent = RAGAgent(retriever=retriever, top_k=5, include_sources=True)
result = agent.run("What auth method does the API use?")
print(result.answer)
print(result.metadata["sources"])  # retrieved chunks

API Reference

brain.patterns.rag.RAGAgent

RAGAgent(retriever: Retriever | None = None, provider: LLMProvider | None = None, top_k: int = 5, include_sources: bool = True)

Bases: BasePattern

Retrieval-Augmented Generation agent.

Parameters:

Name	Type	Description	Default
retriever	Retriever | None	callable(query, top_k) -> list[str] of text chunks. Defaults to a no-op that returns an empty list (offline-safe).	None
provider	LLMProvider | None	LLMProvider instance; defaults to LocalEchoProvider.	None
top_k	int	number of chunks to retrieve.	5
include_sources	bool	attach retrieved chunks to result metadata.	True

Source code in brain/patterns/rag.py

def __init__(
    self,
    retriever: Retriever | None = None,
    provider: LLMProvider | None = None,
    top_k: int = 5,
    include_sources: bool = True,
) -> None:
    self.retriever: Retriever = retriever or (lambda q, k: [])
    self.top_k = top_k
    self.include_sources = include_sources
    self._provider = provider

run

run(task: str, **kwargs: Any) -> PatternResult

Source code in brain/patterns/rag.py

def run(self, task: str, **kwargs: Any) -> PatternResult:
    provider = self._get_provider()
    steps: list[Step] = []

    # Step 1: Retrieve
    chunks: list[str] = []
    try:
        chunks = self.retriever(task, self.top_k)
        if not isinstance(chunks, list):
            chunks = list(chunks)
    except Exception as exc:  # noqa: BLE001
        logger.warning("RAGAgent: retriever failed: %s", exc)

    retrieval_step = Step(
        index=0,
        action="retrieve",
        action_input=task,
        observation=f"{len(chunks)} chunk(s) retrieved",
    )
    steps.append(retrieval_step)

    # Step 2: Augment + Generate
    if chunks:
        context_text = _format_context(chunks)
        system = _SYSTEM_PROMPT
        user_content = f"Context:\n{context_text}\n\nQuestion: {task}"
    else:
        system = _NO_CONTEXT_SYSTEM
        user_content = task

    try:
        result = provider.generate(
            messages=[
                {"role": "system", "content": system},
                {"role": "user", "content": user_content},
            ],
            tools=[],
            tool_choice="none",
        )
        answer = result.text.strip()
    except Exception as exc:  # noqa: BLE001
        return PatternResult(
            answer="",
            steps=steps,
            iterations=1,
            ok=False,
            error=str(exc),
        )

    generate_step = Step(
        index=1,
        action="generate",
        action_input=user_content[:200],
        observation=answer[:200],
    )
    steps.append(generate_step)

    metadata: dict[str, Any] = {}
    if self.include_sources:
        metadata["sources"] = chunks
        metadata["chunks_retrieved"] = len(chunks)

    return PatternResult(
        answer=answer,
        steps=steps,
        iterations=2,
        metadata=metadata,
    )

Retriever Contract

Retriever = Callable[[str, int], list[str]]
#                     query  top_k   chunks

The retriever receives the task string and top_k; it must return a list of text strings.

SecondBrain VectorStore

Plug in the built-in vector store for vault-grounded answers:

from brain.state import VectorStore
from brain.config import get_settings

settings = get_settings()
vs = VectorStore(settings.paths.chroma_dir, model_name=settings.retrieval.embedding_model)

agent = RAGAgent(
    retriever=lambda q, k=5: [c.text for c in vs.search(q, top_k=k)],
)

When to Use

Situation	Recommendation
Answer must cite specific documents	RAGAgent
Knowledge base is local / private	RAGAgent
Open-ended web research	ReAct + search tool
Self-improving critique loop	Reflexion