Docs/Guides/Rag System

RAG System Monitoring

A RAG system answers a question in three steps: embed the question, retrieve matching documents, and generate an answer from them. When an answer is wrong, the cause can be in any step. This guide traces each step as its own observation so you can see what was retrieved, what the model was given and what each step cost. Examples use the Python SDK (ants-platform). For other languages see the JS/TS SDK and Java SDK.

The trace you will build

ObservationTypeWhat it records
rag-assistantagentThe question and the final answer. Root of the trace, carries the agent name.
embed-queryembeddingThe embedding model and its input token count, so the embedding call is costed.
retrieveretrieverThe query and top_k as input, the retrieved documents with their IDs and similarity scores as output.
answergenerationThe full prompt including the retrieved context, the answer, the model and token usage.

Install and configure

bash
pip install "ants-platform[openai]" export ANTS_PLATFORM_PUBLIC_KEY="pk-ap-..." export ANTS_PLATFORM_SECRET_KEY="sk-ap-..." export ANTS_PLATFORM_HOST="https://<your-host>" export OPENAI_API_KEY="sk-..."

API keys and the host URL come from Settings > API Keys in your project. See API Keys.

Write the pipeline

Save this as rag.py. It uses a small in-memory document list in place of a vector database so it runs as is.

python
from ants_platform import AntsPlatform from ants_platform.openai import OpenAI ants = AntsPlatform( public_key=os.environ["ANTS_PLATFORM_PUBLIC_KEY"], secret_key=os.environ["ANTS_PLATFORM_SECRET_KEY"], host=os.environ["ANTS_PLATFORM_HOST"], ) llm = OpenAI() EMBEDDING_MODEL = "text-embedding-3-small" DOCS = [ "Refunds are issued to the original payment method within 14 days.", "Standard shipping is free for orders over 50 USD.", "Support is available by chat from 8:00 to 20:00 CET on weekdays.", ] def embed(texts: list[str]): response = llm.embeddings.create(model=EMBEDDING_MODEL, input=texts) return [item.embedding for item in response.data], response.usage.prompt_tokens DOC_VECTORS, _ = embed(DOCS) def cosine(a: list[float], b: list[float]) -> float: dot = sum(x * y for x, y in zip(a, b)) return dot / (math.sqrt(sum(x * x for x in a)) * math.sqrt(sum(y * y for y in b))) def embed_query(question: str) -> list[float]: with ants.start_as_current_observation( name="embed-query", as_type="embedding", model=EMBEDDING_MODEL, input=question ) as embedding: vectors, tokens = embed([question]) embedding.update( output={"dimensions": len(vectors[0])}, usage_details={"input": tokens}, ) return vectors[0] def retrieve(question: str, vector: list[float], top_k: int = 2) -> list[dict]: with ants.start_as_current_observation( name="retrieve", as_type="retriever", input={"query": question, "top_k": top_k} ) as retriever: ranked = sorted( ((cosine(vector, doc_vector), i) for i, doc_vector in enumerate(DOC_VECTORS)), reverse=True, )[:top_k] docs = [{"id": i, "score": round(score, 3), "text": DOCS[i]} for score, i in ranked] retriever.update(output=docs) if not docs or docs[0]["score"] < 0.3: retriever.update(level="WARNING", status_message="no close match") return docs def rag_assistant(question: str, user_id: str, session_id: str) -> str: with ants.start_as_current_observation( name="rag-assistant", as_type="agent", agent_name="rag-assistant", input=question ) as agent: agent.update_trace(user_id=user_id, session_id=session_id, input=question) docs = retrieve(question, embed_query(question)) context = "\n".join(doc["text"] for doc in docs) completion = llm.chat.completions.create( name="answer", model="gpt-4o-mini", temperature=0.2, messages=[ {"role": "system", "content": f"Answer only from this context:\n{context}"}, {"role": "user", "content": question}, ], ) answer = completion.choices[0].message.content agent.update(output=answer) agent.update_trace(output=answer, metadata={"doc_ids": [doc["id"] for doc in docs]}) return answer if __name__ == "__main__": print(rag_assistant("How long do refunds take?", "demo-user", "demo-session")) ants.flush()

Notes:

  • The root observation needs agent_name. Everything inside it inherits the name.
  • The OpenAI drop-in client records chat completions as generations automatically. It does not record embeddings, so embed_query records the embedding call itself with as_type="embedding" and reports usage_details.
  • The document embeddings computed at startup are not traced. Indexing is usually a separate job; trace it with its own agent observation if you want its cost recorded.
  • With a real vector database, keep the same shape: put the search call inside the retriever observation and write the returned IDs, scores and text (or a snippet) to its output.

Run it and open the trace

bash
python rag.py

Open Discover AI > Tracing and select the rag-assistant trace. Select retrieve to see exactly which documents came back and their scores, and answer to see the prompt the model received, its output, token counts and cost. The Timeline view shows how long each step took.

Find retrieval problems

Most wrong answers trace back to retrieval. With the trace above you can check:

  • Nothing relevant retrieved. The retrieve output shows low scores or unrelated text. Filter Tracing by level WARNING to list traces where the retriever flagged a weak match.
  • Right documents, wrong answer. The documents are in the answer prompt but the answer ignores them. That is a prompt or model problem, not retrieval.
  • Context too long. The answer generation shows a large input token count. Lower top_k or trim documents before building the prompt.

See Debugging AI Agents for recording errors and linking logs to traces.

Track cost per step

Cost is calculated per generation and embedding from the model name and usage, using the prices under Settings > Models. If a model has no price, its cost stays empty. See Models & Pricing.

  • Gauge AI > Cost Dashboard shows spend by model, and Top 20 Use Cases (Observation) by Cost splits spend between embed-query and answer. See Cost & Latency Dashboards.
  • Gauge AI > Latency & Analytics shows p50 to p99 latency per span and generation name, so you can see whether retrieval or generation dominates. See Cost & Performance Metrics.

For ways to reduce cost, see the Cost Optimization guide.

Evaluate answer quality

LLM-as-a-Judge on live traces

Set up LLM-as-a-Judge evaluators that read the trace and write a score between 0 and 1. Managed templates that suit RAG:

TemplateVariables to map
Contextrelevance{{query}} to the trace input, {{context}} to the output of the retrieve observation
Hallucination{{query}} to the trace input, {{generation}} to the trace output
Helpfulness, RelevanceThe query and the answer

Other RAG templates (Contextcorrectness, Context Precision, Context Recall, Faithfulness, Answer Correctness) use variables such as {{ground_truth}}, {{answer}} and {{question}}. Open a template to see its variables before mapping them; templates that need a ground truth fit best on experiment runs, where the dataset item's expected output is available.

Use a filter on trace name rag-assistant and a sampling rate to control judge cost. Judge calls run on your own provider key.

Feedback scores from your app

Record user feedback on the trace while the agent observation is open:

python
with ants.start_as_current_observation( name="rag-assistant", as_type="agent", agent_name="rag-assistant" ): ants.score_current_trace(name="user-feedback", value=1, data_type="BOOLEAN")

If the feedback arrives later, call ants.create_score(name="user-feedback", value=0, trace_id=trace_id, data_type="BOOLEAN") with the trace ID you stored. See Scores.

Regression runs on a dataset

Keep a set of questions with expected answers in a dataset and rerun it after changing the retriever, chunking or prompt:

python
ants.create_dataset(name="rag-questions") ants.create_dataset_item( dataset_name="rag-questions", input={"question": "How long do refunds take?"}, expected_output={"answer": "Within 14 days, to the original payment method."}, ) dataset = ants.get_dataset("rag-questions") for item in dataset.items: with ants.start_as_current_span(name="rag-eval", agent_name="rag-assistant"): with item.run(run_name="top-k-2") as span: answer = rag_assistant(item.input["question"], "eval", "rag-eval") span.update_trace(input=item.input, output=answer) ants.flush()

item.run() must be opened inside an observation that has an agent_name, as shown. Evaluators that target Experiment runs score the new run items automatically, and you can compare runs side by side in the dataset view. See Experiments.

Framework integrations

With LangChain, pass CallbackHandler() in config={"callbacks": [handler]} and set "metadata": {"ants_platform_agent_name": "rag-assistant"}. Retrievers are recorded as retriever observations and LLM calls as generations. See LangChain.

Common issues

SymptomFix
ValueError: agent_name is requiredThe pipeline ran outside the rag-assistant agent observation, for example a retriever or OpenAI call made at the top level.
Embedding step has tokens but no costAdd a price for the embedding model under Settings > Models.
Retrieved documents missing on the traceWrite them to the retriever's output with retriever.update(output=...).
Trace output emptyCall update_trace(output=...) on the root before it closes.
No trace after the script endsCall ants.flush() before the process exits.

Next steps

© 2026 ANTS Platform, Inc.Docs v1.0 · Last updated June 2026