Building a multi-agent RAG system with DSPy and GEPA

This article builds a small research prototype with two specialist agents. One searches a diabetes document collection and the other searches a COPD collection. A lead agent can call either specialist, or both, when a question crosses the two collections. We then use GEPA, DSPy’s prompt optimizer, to improve the agents against a labelled evaluation set.

The medical documents make the routing example concrete. The examples below describe an engineering experiment, not medical advice. A retrieval result or an LLM judge is not a clinical authority. Any real clinical system needs domain review, access controls, and an evaluation designed with qualified practitioners.

The complete supporting code and data files are in the dspy_gepa_optimization repository. The examples use a local PDF corpus and persisted FAISS indexes.

Why use retrieval and more than one agent?

Retrieval-Augmented Generation (RAG) gives a model passages from a corpus at answer time. The corpus can be updated independently of the model, and the answer can retain the passage’s source and page metadata. Retrieval still has failure modes: a distance score is not a quality score, a relevant passage can be missing, and a model can misread or overstate evidence.

One agent can often perform the whole workflow. Specialist agents are useful when collections, tools, or instructions have clear boundaries. In this example, the boundaries are the diabetes and COPD indexes. The lead agent makes the routing decision and receives text plus provenance from each specialist.

Setup

The import paths below were checked with DSPy 3.3.1, langchain-community 0.4.2, langchain-huggingface 1.2.2, langchain-text-splitters 1.1.2, pypdf 6.17.0, and faiss-cpu 1.15.0. The model calls and document files still need to be supplied by the runtime.

import json
import os
import random
from pathlib import Path

import dspy
from dspy.evaluate import Evaluate
from langchain_community.document_loaders import PyPDFLoader
from langchain_community.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter


DIABETES_PDFS = [Path("docs/diabetes-1.pdf"), Path("docs/diabetes-2.pdf")]
COPD_PDFS = [Path("docs/copd-1.pdf"), Path("docs/copd-2.pdf")]
DIABETES_INDEX = Path("indexes/diabetes")
COPD_INDEX = Path("indexes/copd")
EMBEDDING_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
CHUNK_SIZE = 400
CHUNK_OVERLAP = 200
RANDOM_SEED = 7

lm = dspy.LM(
    "openrouter/openai/gpt-oss-20b",
    api_base="https://openrouter.ai/api/v1",
    api_key=os.environ["OPENROUTER_API_KEY"],
    model_type="chat",
    cache=False,
    temperature=0.3,
    max_tokens=64000,
)
dspy.configure(lm=lm)


def load_pages(paths: list[Path]):
    pages = []
    for path in paths:
        if not path.exists():
            raise FileNotFoundError(path)
        for document in PyPDFLoader(str(path)).load():
            metadata = dict(document.metadata or {})
            metadata["source"] = path.name
            # PyPDFLoader uses a zero-based page value. Keep it as metadata.
            metadata["page"] = metadata.get("page", 0)
            document.metadata = metadata
            pages.append(document)
    return pages


def chunk_pages(pages):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=CHUNK_SIZE,
        chunk_overlap=CHUNK_OVERLAP,
        separators=["\n\n", "\n", " ", ""],
    )
    return splitter.split_documents(pages)


def build_index(paths: list[Path], index_path: Path):
    chunks = chunk_pages(load_pages(paths))
    embeddings = HuggingFaceEmbeddings(
        model_name=EMBEDDING_MODEL,
        model_kwargs={"device": "cpu"},
    )
    index = FAISS.from_documents(chunks, embeddings)
    index.save_local(str(index_path))
    return index


diabetes_index = build_index(DIABETES_PDFS, DIABETES_INDEX)
copd_index = build_index(COPD_PDFS, COPD_INDEX)

FAISS.similarity_search_with_score returns the document and a similarity distance. With the default FAISS distance used here, lower values mean that the vectors are closer. The value is useful for ranking candidates; it is not a calibrated relevance or answer-quality probability.

For a later run, load the saved index with the same embedding model instead of rebuilding it. When indexes are persisted outside a trusted directory, use the loading options and review the serialization risk documented by the LangChain integration.

Give each specialist a retrieval tool

The tool returns text and carries the source and page beside each passage. Keeping these fields in the tool output makes it possible for a later answerer to cite what it saw.

def make_search_tool(index, name: str):
    def search(query: str, k: int = 3) -> str:
        """Return up to k passages with their retrieval distances and provenance."""
        if not 1 <= k <= 8:
            raise ValueError("k must be between 1 and 8")
        matches = index.similarity_search_with_score(query, k=k)
        passages = []
        for number, (document, distance) in enumerate(matches, start=1):
            metadata = document.metadata or {}
            source = metadata.get("source", "unknown source")
            page = metadata.get("page", "unknown page")
            passages.append(
                f"[PASSAGE {number}; source={source}; page={page}; "
                f"distance={distance:.4f}]\n{document.page_content}"
            )
        return "\n\n".join(passages) or "No passages found."

    search.__name__ = name
    return search


diabetes_search = make_search_tool(diabetes_index, "diabetes_search")
copd_search = make_search_tool(copd_index, "copd_search")

The closure gives each tool a different index while keeping the callable interface the same. A production tool should also apply the caller’s document permissions before returning passages.

Start with a single RAG module

Start with a simple baseline before adding agents. A DSPy signature declares the inputs and outputs, while ChainOfThought supplies the reasoning field internally.

class RAGAnswer(dspy.Signature):
    """Answer the question using only the supplied passages and cite their sources."""

    question: str = dspy.InputField()
    passages: str = dspy.InputField()
    answer: str = dspy.OutputField()


rag = dspy.ChainOfThought(RAGAnswer)
prediction = rag(
    question="Which passages are relevant to this question?",
    passages=diabetes_search("example retrieval query", k=3),
)
print(prediction.answer)

The answer should be inspected together with the source and page values. For this medical corpus, an evaluation can measure retrieval and answer properties without presenting generated text as a treatment instruction.

Make an evaluation set with a fixed split

GEPA needs examples and a metric. Keep the source records in JSON, make the split from the records themselves, and seed the shuffle. The example below expects each record to contain question and answer fields.

def load_examples(path: Path, train_size: int):
    records = json.loads(path.read_text())
    examples = [
        dspy.Example(question=item["question"], answer=item["answer"])
        .with_inputs("question")
        for item in records
    ]
    rng = random.Random(RANDOM_SEED)
    rng.shuffle(examples)
    return examples[:train_size], examples[train_size:]


train_diabetes, dev_diabetes = load_examples(Path("data/diabetes.json"), train_size=20)
train_copd, dev_copd = load_examples(Path("data/copd.json"), train_size=10)
print(len(train_diabetes), len(dev_diabetes))
print(len(train_copd), len(dev_copd))

The recorded diabetes run reported a baseline score of 90.72 over 29 development examples. That score is tied to the source dataset, metric, model, and run. It is not a clinical accuracy figure. The split sizes above preserve the recorded 20-example training split; the source run did not record enough randomization detail to claim bit-for-bit reproduction.

Build specialist ReAct agents

DSPy’s ReAct module can call the tools supplied at construction time. The specialist signature has one input and one answer output. The tool itself supplies the retrieval context, so the signature does not pretend that an unfilled passages field is available.

class SpecialistAnswer(dspy.Signature):
    """Answer the user's question using evidence returned by the available search tool."""

    question: str = dspy.InputField()
    answer: str = dspy.OutputField()


diabetes_agent = dspy.ReAct(
    SpecialistAnswer,
    tools=[diabetes_search],
    max_iters=4,
)
copd_agent = dspy.ReAct(
    SpecialistAnswer,
    tools=[copd_search],
    max_iters=4,
)

One representative interaction is enough to understand the loop:

1. The question arrives at diabetes_agent.
2. The agent calls diabetes_search with a query and k=3.
3. The tool returns passages with source, page, and distance fields.
4. The agent calls its internal finish action and returns an answer tied to those passages.

The tool result and its provenance are the records needed to inspect this interaction.

Evaluate and optimize with GEPA

The metric evaluates the final answer and, when GEPA is inspecting a ReAct predictor, the next tool decision. It returns a dspy.Prediction with a score in [0, 1] and feedback for both Evaluate and GEPA.

class JudgeConsistency(dspy.Signature):
    """Score an answer against a reviewed reference and the supplied question."""

    question: str = dspy.InputField()
    gold_answer: str = dspy.InputField()
    predicted_answer: str = dspy.InputField()
    score: float = dspy.OutputField(desc="a float from 0 to 1")
    justification: str = dspy.OutputField()


class JudgeReactStep(dspy.Signature):
    """Score whether a tool name and JSON arguments fit the question."""

    question: str = dspy.InputField()
    tool_name: str = dspy.InputField()
    tool_args_json: str = dspy.InputField()
    score: float = dspy.OutputField(desc="a float from 0 to 1")
    justification: str = dspy.OutputField()


def _bounded_score(value) -> float:
    try:
        return max(0.0, min(1.0, float(value)))
    except (TypeError, ValueError):
        return 0.0


def _field(value, name, default=""):
    return value.get(name, default) if isinstance(value, dict) else getattr(value, name, default)


def llm_metric_prediction(
    gold,
    pred,
    trace=None,
    pred_name=None,
    pred_trace=None,
):
    """Return a dspy.Prediction for GEPA's program or predictor call."""
    example = gold
    prediction = pred

    if example is None or prediction is None:
        return dspy.Prediction(score=0.0, feedback="Missing example or prediction.")

    if pred_name and (pred_name == "react" or pred_name.endswith(".react")) and pred_trace:
        try:
            _, _, outputs = pred_trace[0]
        except (IndexError, TypeError, ValueError):
            return dspy.Prediction(score=0.0, feedback="Malformed ReAct trace.")
        tool_name = _field(outputs, "next_tool_name")
        tool_args = _field(outputs, "next_tool_args", {})
        args_ok = isinstance(tool_args, dict)
        query_ok = args_ok and bool(str(tool_args.get("query", "")).strip())
        k = tool_args.get("k") if args_ok else None
        k_ok = k is None or isinstance(k, int) and 1 <= k <= 8
        heuristic = sum((0.4 if tool_name not in ("", "finish") else 0.0,
                         0.4 if query_ok else 0.0,
                         0.1 if k_ok else 0.0,
                         0.1 if tool_name != "finish" else 0.0))
        with dspy.settings.context(lm=lm):
            judged = dspy.Predict(JudgeReactStep)(
                question=example.question,
                tool_name=str(tool_name),
                tool_args_json=json.dumps(tool_args, default=str),
            )
        llm_score = _bounded_score(_field(judged, "score", 0.0))
        return dspy.Prediction(
            score=0.5 * heuristic + 0.5 * llm_score,
            feedback=(
                f"ReAct step checks: structural score {heuristic:.2f}; "
                f"judge score {llm_score:.2f}. "
                f"{_field(judged, 'justification', '')}"
            ),
        )

    predicted_answer = str(_field(prediction, "answer") or "")
    if not predicted_answer.strip():
        return dspy.Prediction(score=0.0, feedback="Empty prediction.")
    with dspy.settings.context(lm=lm):
        judged = dspy.Predict(JudgeConsistency)(
            question=example.question,
            gold_answer=example.answer,
            predicted_answer=predicted_answer,
        )
    score = _bounded_score(_field(judged, "score", 0.0))
    feedback = str(_field(judged, "justification") or "")
    return dspy.Prediction(score=score, feedback=feedback)


reflection_lm = dspy.LM(
    "openrouter/openai/gpt-oss-120b",
    api_base="https://openrouter.ai/api/v1",
    api_key=os.environ["OPENROUTER_API_KEY"],
)

evaluator_diabetes = Evaluate(
    devset=dev_diabetes,
    num_threads=32,
    display_progress=True,
    display_table=5,
    provide_traceback=True,
)
diabetes_baseline_eval = evaluator_diabetes(
    diabetes_agent, metric=llm_metric_prediction
)

evaluator_copd = Evaluate(
    devset=dev_copd,
    num_threads=32,
    display_progress=True,
    display_table=5,
    provide_traceback=True,
)
copd_baseline_eval = evaluator_copd(copd_agent, metric=llm_metric_prediction)

diabetes_optimizer = dspy.GEPA(
    metric=llm_metric_prediction,
    max_full_evals=2,
    num_threads=32,
    track_stats=True,
    track_best_outputs=True,
    add_format_failure_as_feedback=True,
    reflection_lm=reflection_lm,
)
optimized_diabetes_agent = diabetes_optimizer.compile(
    student=diabetes_agent,
    trainset=train_diabetes,
    valset=dev_diabetes,
)
copd_optimizer = dspy.GEPA(
    metric=llm_metric_prediction,
    max_full_evals=2,
    num_threads=32,
    track_stats=True,
    track_best_outputs=True,
    add_format_failure_as_feedback=True,
    reflection_lm=reflection_lm,
)
optimized_copd_agent = copd_optimizer.compile(
    student=copd_agent,
    trainset=train_copd,
    valset=dev_copd,
)

GEPA proposes prompt changes from the metric feedback and evaluates candidates. In the recorded diabetes comparison, the selected candidate scored 94.70 on the same evaluation protocol as the 90.72 baseline over 29 development examples. The change is 3.98 percentage points, or about 4.39% relative to the baseline:

baseline = 90.72
optimized = 94.70
percentage_points = optimized - baseline       # 3.98
relative_percent = percentage_points / baseline * 100  # about 4.39

These are results from one task set, model, metric, and selection procedure. A prompt that improves this split can still fail on new questions or retrieve the wrong evidence.

The recorded COPD baseline was 89.44 over 9 development examples under the same evaluator settings. The source article did not report a selected COPD score, so no additional improvement is inferred here.

Let a lead agent call both specialists

Wrap each optimized program in a small tool. The wrappers call the program through its public callable interface and return its answer.

def ask_diabetes(question: str) -> str:
    """Ask the diabetes specialist and return its answer."""
    return str(optimized_diabetes_agent(question=question).answer)


def ask_copd(question: str) -> str:
    """Ask the COPD specialist and return its answer."""
    return str(optimized_copd_agent(question=question).answer)


class LeadAnswer(dspy.Signature):
    """Route a question to one or both specialists.

    Finish only after the required specialist reports arrive. Answer from those
    reports and preserve their source and page citations.
    """

    question: str = dspy.InputField()
    answer: str = dspy.OutputField()


lead_agent = dspy.ReAct(
    LeadAnswer,
    tools=[ask_diabetes, ask_copd],
    max_iters=5,
)

The lead may call one specialist for a focused question and both when the question needs evidence from both collections. It finishes after the required reports have arrived.

The specialist reports should include their source and page metadata if the final answer claims traceable evidence. If the lead rewrites or combines the reports, retain those citations and make clear which collection supplied each claim.

Build and split a joint dataset

When evaluating the lead agent, derive its size from the actual records. The recorded joint run used a 20-example training split and the remaining records for development. If mixed questions are added, keep them in the source file and apply the same seeded split.

def load_joint_examples(path: Path, train_size: int):
    records = json.loads(path.read_text())
    examples = [
        dspy.Example(question=item["question"], answer=item["answer"])
        .with_inputs("question")
        for item in records
    ]
    rng = random.Random(RANDOM_SEED)
    rng.shuffle(examples)
    return examples[:train_size], examples[train_size:]


trainset_joint, devset_joint = load_joint_examples(
    Path("data/joint.json"), train_size=20
)
print(f"loaded {len(trainset_joint) + len(devset_joint)} joint examples")
print(f"joint train size: {len(trainset_joint)}")
print(f"joint development size: {len(devset_joint)}")

evaluator_joint = Evaluate(
    devset=devset_joint,
    num_threads=32,
    display_progress=True,
    display_table=5,
    provide_traceback=True,
)
baseline_lead_eval = evaluator_joint(lead_agent, metric=llm_metric_prediction)

The recorded baseline was EvaluationResult(score=88.79, results=<list of 33 results>). That score covers the 33-example development set from the joint dataset and is an experiment result, not a medical accuracy claim.

lead_optimizer = dspy.GEPA(
    metric=llm_metric_prediction,
    max_full_evals=3,
    num_threads=32,
    track_stats=True,
    track_best_outputs=True,
    add_format_failure_as_feedback=True,
    reflection_lm=reflection_lm,
)
optimized_lead = lead_optimizer.compile(
    student=lead_agent,
    trainset=trainset_joint,
    valset=devset_joint,
)

The best validation score recorded for this run was 0.9242424242424242, or 92.424%. It is the selected candidate’s validation score under the configuration above. The specialist prompts were treated as black boxes during this lead optimization; changing that scope would be a separate experiment.

What this prototype establishes

The experiment demonstrates a design pattern: narrow retrieval tools can be composed behind a coordinator, and GEPA can search prompt candidates against a feedback metric. It does not show that multiple agents always outperform one, that a medical answer is safe, or that the reported score transfers to another corpus, model, or workload.

The next useful checks are retrieval recall, citation coverage, answer correctness against a reviewed set, tool-call errors, latency, and cost. Keep failures separate: an unavailable model or vector index is an infrastructure failure, while a wrong answer after valid retrieval is a model or prompt failure. Reporting both categories makes the result easier to reproduce and act on.

References

• DSPy documentation
• GEPA paper and project links
• Current LangChain vector-store integrations
• Supporting code and data for this experiment