Evaluate a RAG system

This end-to-end guide evaluates a retrieval-augmented generation (RAG) system: an AnythingLLM workspace answering questions about the NIST AI Risk Management Framework from an indexed document corpus. You will create a RAG project, connect AnythingLLM as the system under test, supply a Q&A golden dataset (either your own or one generated for you), pick retrieval and answer-quality metrics, and run a hosted evaluation that calls the live endpoint and scores every answer.

The workflow reuses the building blocks documented elsewhere — this guide wires them together and links each step to its reference section: Projects, Systems Under Test, Datasets, Golden Datasets, Evaluation Configs, and Runs & Results. Start by connecting the SDK.

import os
import aip_sdk as aip

aip.init("https://api.trials.aip-v2.resarodev.ai", api_key="<your-api-key>")
workspace_name = os.environ.get("AIP_WORKSPACE_NAME")
ws = aip.Workspace.get_by_name(workspace_name) if workspace_name else aip.Workspace.default()

Create the RAG project

RAG is canonically the (gdi_text_v1, single_turn_rag) pairing — there is no separate RAG schema. The single_turn_rag task type is what tells the platform a row carries retrieved context alongside the prompt and answer, and it must match the RAG SUT you connect next. Use get_or_create so a reran notebook or CI job provisions the project once and reuses it thereafter. A Dimension declares a categorical column the platform slices your data and results by; here a single topic dimension is enough. Note that task_type="single_turn_rag" is required for gdi_text_v1 and must match the RAG SUT. See Projects for the full create surface. With the SDK:

rag_project, created = aip.Project.get_or_create(
    name="AnythingLLM RAG Demo (NIST AI RMF)",
    schema="gdi_text_v1",
    task_type="single_turn_rag",
    dimensions=[aip.Dimension(name="topic", column="topic", values=["nist_ai_rmf"])],
    workspace_id=ws.id,
)

Register the AnythingLLM RAG SUT

A system under test (SUT) is the model or service you evaluate; it is registered under the project and then given a connection — the HTTP endpoint AIP calls. Register (or reuse) the SUT first. get_or_register returns a (sut, created) tuple, so reruns are idempotent. See Systems Under Test for the one-call aip.sut.register(...) shortcut and the full connection surface. With the SDK:

sut, created = aip.Sut.get_or_register(
    name="anythingllm-rag-nist",
    version="1.0",
    project_id=rag_project.id,
)

Add the connection and RAG adapter

AnythingLLM's workspace-chat endpoint is POST /api/v1/workspace/{slug}/chat, and it speaks a native format rather than the OpenAI chat shape: the request body is {message, mode} and the response is {textResponse, sources}. Because it is not OpenAI-compatible, the built-in openai_chat template does not fit; the platform's generic text builder, generic_json, does. Point the connection's base_url at the full chat endpoint — the platform POSTs there directly — and set model_params so generic_json renames the outgoing prompt field to message and sends mode: "query" — AnythingLLM's retrieval (RAG) mode rather than plain chat.

AnythingLLM expects the Authorization header value to already include the Bearer prefix, and the platform forwards auth_header_value verbatim, so include it yourself. With the SDK:

import os

endpoint = f"{os.environ['ANYTHINGLLM_BASE_URL']}/api/v1/workspace/{os.environ['RAG_WORKSPACE_SLUG']}/chat"

conn = sut.add_connection(
    base_url=endpoint,
    label="anythingllm-rag-nist",
    auth_type="bearer",
    auth_header_name="Authorization",
    auth_header_value=f"Bearer {os.environ['ANYTHINGLLM_API_KEY']}",
    sut_protocol="generic_json",
    gdi_schema="gdi_text_v1",
    task_type="single_turn_rag",
    model_params={"prompt_field": "message", "mode": "query"},
)

The adapter is where a RAG connection differs from a plain LLM one: it maps the raw response into two GDI output columns rather than one:

• $.textResponse becomes the sut_response — the generated answer.
• $.sources[*].text extracts just the chunk-text strings from each source object into retrieved_context — the retrieved passages the RAG metrics score against. ($.sources returns full source objects {id, text, score, ...}; the trailing [*].text is what turns them into the List[str] that retrieved_context expects.)

Passing template_name=None supplies this mapping directly instead of starting from a built-in template. With the SDK:

conn.set_adapter(
    template_name=None,
    mapping_config={
        "sut_response": "$.textResponse",
        "retrieved_context": "$.sources[*].text",
    },
)

Validate the mapping against a sample payload and run a live connection test before relying on it — both are covered in Systems Under Test.

Provide the evaluation dataset

The evaluation needs Q&A pairs: a prompt (question), an expected_output (reference answer), and per-row reference_contexts (the source chunks a correct answer should draw on, used by the retrieval metric rag.hit_at_k). There are two ways to get them, and you run exactly one. Both paths end the same way — an uploaded, quality-checked dataset promoted to golden — so every step after this is identical regardless of which you choose.

Path	When to use
Option A — Bring your own Q&A dataset	You already have question/answer pairs with reference contexts.
Option B — Generate with Synthetic Data Generation	You want the platform to generate Q&A pairs from your document corpus.

Option A — Bring your own Q&A dataset

Load your Q&A pairs into a DataFrame with the gdi_text_v1 columns input_id, prompt, and expected_output, then attach reference_contexts — a list of source chunks per row drawn from the same corpus the AnythingLLM workspace indexes, which the retrieval metric rag.hit_at_k scores against. Upload it under the project, run quality checks, upload the check results, and promote the version to golden. Promotion requires a PASS quality verdict; if the latest verdict is a WARN, acknowledge it with force=True and a reason (forced promotions are audited). The CSV read here supplies the input_id, prompt, and expected_output columns. See Datasets and Golden Datasets for the full lifecycle. With the SDK:

import time
import pandas as pd

qa_df = pd.read_csv("nist_qa_pairs.csv")

chunks = pd.read_csv("nist_chunks.csv")
reference_contexts = [c.strip() for c in chunks["chunk"].dropna().astype(str) if c.strip()]
qa_df["reference_contexts"] = [list(reference_contexts) for _ in range(len(qa_df))]

rag_ds = rag_project.upload_dataset(qa_df, name=f"rag-qa-pairs-{int(time.time())}")
rag_v = rag_ds.latest_version()

# RAG Q&A pairs have no canonical schema yet, so the selection is row-count-only.
report = rag_ds.run_checks(rag_v.id, label="qa-pairs-import", checks=["row_count"])

try:
    rag_golden = rag_ds.promote(rag_v.id)
except Exception:
    rag_golden = rag_ds.promote(rag_v.id, force=True, reason="Dataset acknowledged for evaluation.")
rag_v = rag_golden

The try/except promotes on a PASS verdict, and falls back to a forced promotion with a reason when the verdict is a WARN.

Option B — Generate Q&A pairs with Synthetic Data Generation

If you have a document corpus but no Q&A pairs, let the platform generate them with Synthetic Data Generation (SDG). You upload the corpus as a seed dataset, run a RAG SDG job (ragas_single_hop generation followed by an open-book quality check), and the platform produces an augmented dataset version of query / answer / context rows with full lineage back to the corpus. The full SDG mechanics — job graphs, generator types, quality control, and running an adapter — are covered in the Guide: Synthetic Data Generation; this guide only shows where its output plugs in, as the augmented DataFrame df_sdg_qa below.

Once the SDG run completes, map the generated columns onto the gdi_text_v1 eval columns (query → prompt, answer → expected_output, context → reference_contexts), then upload, quality-check, and promote it to golden exactly as in Option A — landing on the same rag_ds / rag_v this guide uses downstream. With the SDK:

if "prompt" not in df_sdg_qa and "query" in df_sdg_qa:
    df_sdg_qa["prompt"] = df_sdg_qa["query"]
if "expected_output" not in df_sdg_qa and "answer" in df_sdg_qa:
    df_sdg_qa["expected_output"] = df_sdg_qa["answer"]
if "reference_contexts" not in df_sdg_qa and "context" in df_sdg_qa:
    df_sdg_qa["reference_contexts"] = df_sdg_qa["context"].apply(
        lambda x: [str(v) for v in x] if hasattr(x, "__iter__") and not isinstance(x, str) else [str(x)]
    )

rag_ds = rag_project.upload_dataset(df_sdg_qa, name=f"rag-sdg-eval-{int(time.time())}")
rag_v = rag_ds.promote(rag_ds.latest_version().id, force=True, reason="SDG-generated QA pairs for evaluation.")

Download and inspect the golden dataset

With a golden version in place, download it and confirm its shape before spending a run on it. dataset.download() streams the rows from object storage as a pandas DataFrame (see Golden Datasets). For a RAG set you should see prompt, expected_output, and a reference_contexts list on each row. With the SDK:

df_golden = rag_ds.download()
print(len(df_golden), list(df_golden.columns))
print("reference chunks on row 0:", len(df_golden["reference_contexts"].iloc[0]))

It is worth confirming the golden data against the schema contract. gdi_text_v1 is a rag-family schema whose required columns include input_id, prompt, and sut_response (the last is injected by the platform at evaluation time, when it calls the SUT), and whose optional columns include retrieved_context and reference_contexts.

Dimensions live on the project, not the dataset, and drive how results are sliced. The project already declares topic; you can add more when a corresponding column exists in the golden data — for example a scenario slice — by patching the project's dimension list. Keeping dimensions in sync with the data you actually uploaded is what lets Runs & Results break scores down per slice.

Select metrics

A RAG system generates text, so both RAG-specific retrieval metrics and general LLM text metrics apply — which frames the NIST AI RMF evaluation as covering two axes: did the system retrieve the right NIST passages, and is the answer itself faithful and well-formed. From the SDK, discover the metrics that accept your schema, then split them by family:

metrics = aip.ops.list_metrics(schema="gdi_text_v1")
rag_ops = [m for m in metrics if m["name"].startswith("rag.")]
llm_ops = [m for m in metrics if m["name"].startswith("llm.")]
for m in rag_ops + llm_ops:
    print(m["name"], "-", m.get("description", ""))

Two metrics run without any external judge key and make a good default pairing: rag.hit_at_k measures retrieval overlap against each row's reference_contexts, and llm.bleu measures answer overlap with expected_output. rag.hit_at_k takes a k parameter, passed through the metric's test_params. With the SDK:

selected_metrics = ["rag.hit_at_k", "llm.bleu"]
metric_configs = {
    "rag.hit_at_k": {"test_params": {"metric_params": {"k": 5}}},
}

To go deeper into answer quality and trustworthiness, opt into judge-backed metrics such as rag.faithfulness (is the answer grounded in the retrieved context?), rag.context_precision, and llm.toxicity. These are scored by an LLM judge, so they need a judge model — either the worker's global judge credential or a per-metric override supplied through metric_configs (see the judge-model wiring in Runs & Results). Opt into these optional judge-backed metrics from the SDK:

selected_metrics += ["rag.faithfulness", "rag.context_precision", "llm.toxicity"]
metric_configs.update(
    {
        "rag.faithfulness": {"judge_model": "gpt-4o-mini"},
        "rag.context_precision": {"judge_model": "gpt-4o-mini"},
        "llm.toxicity": {"judge_model": "gpt-4o-mini"},
    }
)

Publish the evaluation config

Capture the project, golden dataset (pinned to its version for reproducibility), SUT connection, and metric selection as a reusable evaluation config, then publish it to the platform. Pinning the dataset to @v{version} keeps the run reproducible. This makes the run definition versionable and shareable; the same config can be pulled back or authored as YAML — see Evaluation Configs for the full schema. With the SDK:

import time
from aip_core.schemas.eval_config import EvalConfig

cfg = EvalConfig(
    project=str(rag_project.id),
    dataset=f"{rag_ds.id}@v{rag_v.version}",
    sut_id=sut.id,
    connection_id=conn.id,
    metrics=selected_metrics,
)

cfg_id = aip.publish_config(
    cfg,
    name=f"rag-qa-nist-eval-{int(time.time())}",
    description="RAG (NIST AI RMF docs) demo evaluation config",
    workspace_id=ws.id,
)

Run the hosted evaluation

Now run it. Passing connection_id to aip.run(...) (here alongside sut_id, which is attribution-only) selects the hosted runner: AIP executes the whole evaluation for you — it calls the live AnythingLLM endpoint for every question, extracts retrieved_context through the connection adapter, injects the sut_response, and scores the selected metrics — while you only poll for completion. poll_status blocks until the run reaches a terminal state. Because the hosted runner does the SUT calls, you do not download the golden set or upload results yourself. See Runs & Results for the external-runner alternative and the full Run surface. With the SDK:

with aip.run(
    project=rag_project.id,
    dataset=f"{rag_ds.id}@v{rag_v.version}",
    sut_id=sut.id,
    connection_id=conn.id,
    metrics=selected_metrics,
    metric_configs=metric_configs,
) as run:
    run.poll_status(interval=5, timeout=900)
    print(run.status, run.url)
    run_id = run.id

Read the results

Fetch per-row scores and aggregates from the completed run. run.results() returns a RunResultsPage: .results holds one ScoreResult per (input_id, scorer) pair (input_id, scorer, score, and an optional explanation), and .metrics holds one aggregated Metric per scorer (mean, std, count, pass_rate). You can read a run created earlier with aip.get_run(run_id). From the SDK, the three snippets below read, in order, the per-row scores, the aggregated per-metric summary, and the score breakdown by dimension slice:

run = aip.get_run(run_id)
page = run.results(page=1, page_size=100)

for r in page.results[:12]:
    print(r.input_id, r.scorer, "=", r.score)

for m in page.metrics:
    print(f"{m.scorer}: mean={m.mean:.3f} pass_rate={m.pass_rate:.0%} (n={m.count})")

analysis = run.analysis(group_by="topic")

The scores tell you how well the system did; to see what it did, read the run's scored output. The hosted run writes a Parquet file that carries, next to each metric column, the SUT's sut_response (the answer AnythingLLM returned) and the adapter-extracted retrieved_context (the passages it retrieved) — so you can inspect the answer and the evidence behind every score together. The loaded df_out carries input_id, prompt, sut_response, retrieved_context, and one column per metric. Over REST, request a presigned URL for the run's scored Parquet output with curl:

curl -sS "$AIP_API_URL/runs/<run_id>/parquet-url" \
  -H "Authorization: Bearer $AIP_TOKEN"

Or fetch the same URL and read the Parquet directly with the SDK:

import io, requests
from aip_sdk._context import get_default_client

url = get_default_client().get(f"/runs/{run_id}/parquet-url")["url"]
df_out = pd.read_parquet(io.BytesIO(requests.get(url, timeout=60).content))

That closes the loop: the AnythingLLM RAG system has been evaluated end to end against a NIST AI RMF golden set, with per-answer retrieval and quality scores you can slice, export, and compare across runs.