RAG System Monitoring
Observe retrieval-augmented generation systems end-to-end with comprehensive monitoring and optimization.
Overview
Retrieval-Augmented Generation (RAG) systems combine information retrieval with text generation. AgenticAnts provides complete observability for:
- Retrieval Quality - Monitor vector search performance
- Generation Performance - Track LLM response quality
- End-to-End Latency - Measure total system performance
- Cost Optimization - Track retrieval and generation costs
- Quality Metrics - Monitor answer relevance and accuracy
RAG Architecture Components
Typical RAG System Flow
Model the RAG pipeline as a nested agent observation: an embedding generation for the query vector, a retriever observation that captures the query as input and the retrieved documents as output, then a generation that consumes the assembled context. Open the agent with start_as_current_observation(..., as_type="agent", agent_name=...) (the agent_name argument is required for agents), set host="https://api.agenticants.ai", and supply your public/secret key pair (or rely on the ANTS_PLATFORM_* environment variables).
python
from ants_platform import AntsPlatform, get_client
# Configure the singleton once at startup.
AntsPlatform(
public_key=os.environ["ANTS_PLATFORM_PUBLIC_KEY"],
secret_key=os.environ["ANTS_PLATFORM_SECRET_KEY"],
host="https://api.agenticants.ai",
)
class RAGSystem:
def __init__(self):
self.vector_store = VectorStore()
self.retriever = Retriever()
self.generator = Generator()
def query(self, question: str, user_id: str = None):
client = get_client()
with client.start_as_current_observation(
name="rag-system",
as_type="agent",
agent_name="rag-system",
):
client.update_current_trace(
input=question,
user_id=user_id,
metadata={"system_type": "rag", "vector_db": "pinecone"},
)
# Step 1: Embed the query (record token usage for cost attribution).
with client.start_as_current_observation(
name="embed-query",
as_type="embedding",
model="text-embedding-3-small",
input=question,
) as emb:
query_vector = self.vector_store.embed(question)
emb.update(
output={"dimensions": len(query_vector)},
usage_details={"input": self.count_tokens(question), "total": self.count_tokens(question)},
)
# Step 2: Retrieve. The retriever captures the QUERY as input and the
# RETRIEVED DOCS as output -- this is the core RAG observation.
with client.start_as_current_observation(
name="vector-retrieval",
as_type="retriever",
input={"query": question, "top_k": 5},
) as retriever:
retrieved_docs = self.retriever.retrieve(query_vector, top_k=5)
retriever.update(
output=[
{"id": d.id, "score": d.score, "text": d.content[:500]}
for d in retrieved_docs
],
metadata={
"retrieved_count": len(retrieved_docs),
"top_score": max(d.score for d in retrieved_docs),
},
)
# Step 3: Assemble context from the retrieved documents.
context = "\n\n".join(d.content for d in retrieved_docs)
# Step 4: Generation consumes the retrieved context.
prompt = f"Context:\n{context}\n\nQuestion: {question}"
with client.start_as_current_observation(
name="generate-answer",
as_type="generation",
model="gpt-4o",
input=prompt,
model_parameters={"temperature": 0.2},
) as gen:
answer = self.generator.generate(prompt)
gen.update(
output=answer,
usage_details={
"input": self.count_tokens(prompt),
"output": self.count_tokens(answer),
"total": self.count_tokens(prompt) + self.count_tokens(answer),
},
)
client.update_current_trace(output=answer, metadata={"success": True})
return answer
Java
Java mirrors the same nested shape. Open the agent with Ants.observe(name, ObservationType.AGENT, ...) (no separate agent-name argument -- the SDK derives it from the observation name), then nest an EMBEDDING generation for the query vector, a RETRIEVER observation that captures the query as input and the retrieved documents as output, and a final GENERATION that consumes the assembled context. Report model + usage(...) on the generations so the backend computes cost server-side. Configure once at startup; the host defaults to https://api.agenticants.ai. Requires Java 17+.
java
public class RagSystem {
public static void main(String[] args) {
// Configure the singleton once at startup.
AntsPlatformOtel.configure(AntsPlatformOtel.options()
.publicKey(System.getenv("ANTS_PLATFORM_PUBLIC_KEY"))
.secretKey(System.getenv("ANTS_PLATFORM_SECRET_KEY"))
.environment("production"));
String question = "What is retrieval-augmented generation?";
String userId = "demo-user";
String answer = Ants.observe("rag-system", ObservationType.AGENT, agent -> {
agent.updateTrace(TraceAttributes.builder()
.name("rag-system").userId(userId).input(question)
.metadata(Map.of("system_type", "rag", "vector_db", "pinecone")).build());
// Step 1: Embed the query (report token usage for cost attribution).
AntsGeneration emb = Ants.startGeneration("embed-query")
.model("text-embedding-3-small").input(question);
emb.output(Map.of("dimensions", 1536)).usage(8, 0).end();
// Step 2: Retrieve -- the retriever captures the QUERY as input and the
// RETRIEVED DOCS as output. This is the core RAG observation.
List<Map<String, Object>> docs = List.of(
Map.of("id", "doc-1", "score", 0.88, "text", "RAG combines retrieval with generation."),
Map.of("id", "doc-2", "score", 0.81, "text", "It grounds answers in retrieved context."));
AntsObservation retriever = Ants.startObservation("vector-retrieval", ObservationType.RETRIEVER);
retriever.input(Map.of("query", question, "topK", 3))
.output(Map.of("documents", docs, "avgScore", 0.82))
.metadata(Map.of("retrieved_count", docs.size())).end();
// Step 3: Generation consumes the retrieved context.
AntsGeneration gen = Ants.startGeneration("generate-answer")
.model("gpt-4o-mini").modelParameters(Map.of("temperature", 0.2))
.input(Map.of("question", question, "context", docs));
String result = "RAG retrieves relevant documents, then generates a grounded answer.";
gen.output(result).usage(220, 60).end();
agent.updateTrace(TraceAttributes.builder().output(result)
.metadata(Map.of("success", true)).build());
return result;
});
System.out.println(answer);
AntsPlatformOtel.flush(); // deliver buffered spans before exit
}
}
Retrieval Quality Monitoring
Vector Search Performance
A standalone retriever observation makes the search auditable: the query goes in as input, every retrieved document (id, score, snippet) goes out as output. Recording the documents verbatim is what lets you later inspect why a generation answered the way it did.
python
from ants_platform import AntsPlatform, get_client
AntsPlatform(
public_key=os.environ["ANTS_PLATFORM_PUBLIC_KEY"],
secret_key=os.environ["ANTS_PLATFORM_SECRET_KEY"],
host="https://api.agenticants.ai",
)
class RetrievalMonitor:
def monitor_retrieval(self, query: str, top_k: int = 5):
client = get_client()
with client.start_as_current_observation(
name="vector-retrieval",
as_type="retriever",
input={"query": query, "top_k": top_k, "retrieval_type": "semantic_search"},
) as retriever:
start_time = time.time()
# Perform vector search.
results = self.vector_store.similarity_search(query=query, k=top_k)
retrieval_time = time.time() - start_time
# Calculate relevance scores for each retrieved document.
relevance_scores = [self.calculate_relevance(query, r.content) for r in results]
# Capture the retrieved documents verbatim as the observation output.
retriever.update(
output=[
{"id": r.id, "score": score, "text": r.content[:500]}
for r, score in zip(results, relevance_scores)
],
metadata={
"retrieval_time": retrieval_time,
"results_count": len(results),
"avg_relevance": sum(relevance_scores) / len(relevance_scores),
"min_relevance": min(relevance_scores),
"max_relevance": max(relevance_scores),
},
)
return results
def calculate_relevance(self, query: str, content: str) -> float:
# Implement relevance scoring logic
# This could use semantic similarity, keyword matching, etc.
...
Retrieval Quality Metrics
Open a retriever observation with startObservation(name, body, { asType: "retriever" }), record the query as input and the retrieved documents as output, then .end() it. Tracing requires Node.js 20+ and a tracer provider configured once at startup (see the end-to-end example below).
typescript
class RetrievalQualityMonitor {
async monitorRetrievalQuality(query: string, retrievedDocs: any[]) {
const retriever = startObservation(
'retrieval-quality',
{
input: { query, queryType: this.classifyQuery(query) },
},
{ asType: 'retriever' }
)
// Calculate quality metrics over the retrieved set.
const metrics = {
relevance: await this.calculateRelevance(query, retrievedDocs),
diversity: this.calculateDiversity(retrievedDocs),
coverage: this.calculateCoverage(query, retrievedDocs),
freshness: this.calculateFreshness(retrievedDocs),
}
// The retriever output captures the documents AND the quality scorecard.
retriever.update({
output: retrievedDocs.map((d) => ({ id: d.id, score: d.score, text: d.text })),
metadata: {
retrievedCount: retrievedDocs.length,
avgRelevance: metrics.relevance.avg,
diversityScore: metrics.diversity,
coverageScore: metrics.coverage,
freshnessScore: metrics.freshness,
},
})
retriever.end()
return metrics
}
private classifyQuery(query: string): string {
// Classify query type (factual, analytical, creative, etc.)
return 'factual' // Simplified
}
private calculateDiversity(docs: any[]): number {
// Calculate diversity of retrieved documents
return 0.8 // Simplified
}
private calculateCoverage(query: string, docs: any[]): number {
// Calculate how well documents cover the query
return 0.9 // Simplified
}
private calculateFreshness(docs: any[]): number {
// Calculate freshness of retrieved documents
return 0.7 // Simplified
}
private async calculateRelevance(query: string, docs: any[]): Promise<any> {
return { avg: 0.85 }
}
}
In Java, open a standalone RETRIEVER observation with Ants.startObservation(name, ObservationType.RETRIEVER), record the query as input and the retrieved documents (id, score, snippet) verbatim as output, then .end() it. Capturing the documents is what lets you later inspect why a generation answered the way it did.
java
public class RetrievalQualityMonitor {
public List<Map<String, Object>> monitorRetrievalQuality(String query) {
AntsObservation retriever = Ants.startObservation("retrieval-quality", ObservationType.RETRIEVER);
retriever.input(Map.of("query", query, "queryType", classifyQuery(query)));
long start = System.nanoTime();
List<Map<String, Object>> docs = vectorStore.similaritySearch(query, 5);
double retrievalTimeMs = (System.nanoTime() - start) / 1_000_000.0;
double avgScore = docs.stream()
.mapToDouble(d -> (double) d.get("score")).average().orElse(0.0);
// The retriever output captures the documents AND the quality scorecard.
retriever.output(docs).metadata(Map.of(
"retrievedCount", docs.size(),
"avgRelevance", avgScore,
"retrievalTimeMs", retrievalTimeMs)).end();
return docs;
}
}
Generation Performance Monitoring
LLM Generation Tracking
For OpenAI calls, the simplest path is the drop-in wrapper: replace import openai with from ants_platform.openai import openai and every call is auto-traced as a generation. The example below shows manual generation spans for non-OpenAI providers.
python
from ants_platform import AntsPlatform, get_client
AntsPlatform(
public_key=os.environ["ANTS_PLATFORM_PUBLIC_KEY"],
secret_key=os.environ["ANTS_PLATFORM_SECRET_KEY"],
host="https://api.agenticants.ai",
)
class GenerationMonitor:
def monitor_generation(self, question: str, context: str, model: str = "gpt-4o"):
client = get_client()
prompt = f"Context:\n{context}\n\nQuestion: {question}"
# The generation consumes the retrieved context. Pass model + usage_details
# so the backend can compute cost server-side.
with client.start_as_current_observation(
name="generate",
as_type="generation",
model=model,
input=prompt,
model_parameters={"temperature": 0.2},
) as gen:
start_time = time.time()
response = self.generate_response(question, context, model)
generation_time = time.time() - start_time
quality_metrics = self.evaluate_response_quality(question, response)
input_tokens = self.count_tokens(prompt)
output_tokens = self.count_tokens(response)
gen.update(
output=response,
usage_details={
"input": input_tokens,
"output": output_tokens,
"total": input_tokens + output_tokens,
},
metadata={
"generation_time": generation_time,
"context_length": len(context),
"quality_score": quality_metrics["overall_score"],
"relevance_score": quality_metrics["relevance"],
"coherence_score": quality_metrics["coherence"],
"factual_accuracy": quality_metrics["factual_accuracy"],
},
)
return response
def evaluate_response_quality(self, question: str, response: str) -> dict:
# Implement quality evaluation logic
return {
"overall_score": 0.85,
"relevance": 0.9,
"coherence": 0.8,
"factual_accuracy": 0.85,
}
Java has no framework drop-ins yet, so instrument LLM calls manually with Ants.startGeneration(...). Pass model + usage(inputTokens, outputTokens) so the backend computes cost server-side; attach quality scores as metadata.
java
public class GenerationMonitor {
public String monitorGeneration(String question, String context, String model) {
String prompt = "Context:\n" + context + "\n\nQuestion: " + question;
// The generation consumes the retrieved context. Report model + usage so the
// backend can compute cost server-side.
AntsGeneration gen = Ants.startGeneration("generate")
.model(model).modelParameters(Map.of("temperature", 0.2)).input(prompt);
long start = System.nanoTime();
String response = generateResponse(question, context, model);
double generationTimeMs = (System.nanoTime() - start) / 1_000_000.0;
Map<String, Object> quality = evaluateResponseQuality(question, response);
int inputTokens = countTokens(prompt);
int outputTokens = countTokens(response);
gen.output(response)
.usage(inputTokens, outputTokens)
.metadata(Map.of(
"generation_time_ms", generationTimeMs,
"context_length", context.length(),
"quality_score", quality.get("overall_score"),
"relevance_score", quality.get("relevance")))
.end();
return response;
}
}
Response Quality Evaluation
typescript
class ResponseQualityEvaluator {
async evaluateResponse(question: string, response: string, context: string) {
const evaluator = startObservation(
'response-evaluation',
{
input: { question, response, context },
},
{ asType: 'evaluator' }
)
// Evaluate multiple quality dimensions against the retrieved context.
const evaluations = await Promise.all([
this.evaluateRelevance(question, response),
this.evaluateCoherence(response),
this.evaluateFactualAccuracy(response, context),
this.evaluateCompleteness(question, response),
])
const overallScore =
evaluations.reduce((sum, score) => sum + score, 0) / evaluations.length
evaluator.update({
output: { overallScore, evaluations },
metadata: {
relevance: evaluations[0],
coherence: evaluations[1],
factualAccuracy: evaluations[2],
completeness: evaluations[3],
},
})
evaluator.end()
return { overallScore, evaluations }
}
private async evaluateRelevance(question: string, response: string): Promise<number> {
// Implement relevance evaluation
return 0.9
}
private async evaluateCoherence(response: string): Promise<number> {
// Implement coherence evaluation
return 0.8
}
private async evaluateFactualAccuracy(response: string, context: string): Promise<number> {
// Implement factual accuracy evaluation
return 0.85
}
private async evaluateCompleteness(question: string, response: string): Promise<number> {
// Implement completeness evaluation
return 0.9
}
}
End-to-End RAG Monitoring
Complete RAG Pipeline Tracking
python
from ants_platform import AntsPlatform, get_client
AntsPlatform(
public_key=os.environ["ANTS_PLATFORM_PUBLIC_KEY"],
secret_key=os.environ["ANTS_PLATFORM_SECRET_KEY"],
host="https://api.agenticants.ai",
)
class CompleteRAGMonitor:
def monitor_rag_pipeline(self, question: str, user_id: str = None):
client = get_client()
with client.start_as_current_observation(
name="complete-rag-pipeline",
as_type="agent",
agent_name="complete-rag-pipeline",
):
client.update_current_trace(
input=question,
user_id=user_id,
metadata={"pipeline_version": "1.0"},
)
# Step 1: Query analysis.
with client.start_as_current_observation(
name="query-analysis", as_type="span", input=question
) as span:
query_analysis = self.analyze_query(question)
span.update(
output=query_analysis,
metadata={
"query_type": query_analysis["type"],
"complexity": query_analysis["complexity"],
},
)
# Step 2: Embed the query.
with client.start_as_current_observation(
name="embed-query",
as_type="embedding",
model="text-embedding-3-small",
input=question,
) as emb:
query_vector = self.embed(question)
emb.update(
output={"dimensions": len(query_vector)},
usage_details={"input": self.count_tokens(question), "total": self.count_tokens(question)},
)
# Step 3: Retrieval -- query in, retrieved docs out.
with client.start_as_current_observation(
name="retrieval", as_type="retriever", input={"query": question, "top_k": 5}
) as retriever:
retrieved_docs = self.retrieve_documents(query_vector)
retriever.update(
output=[
{"id": d.id, "score": d.score, "text": d.content[:500]}
for d in retrieved_docs
],
metadata={
"retrieved_count": len(retrieved_docs),
"avg_relevance": self.calculate_avg_relevance(retrieved_docs),
},
)
# Step 4: Context preparation.
context = self.prepare_context(retrieved_docs)
# Step 5: Generation consumes the retrieved context.
prompt = f"Context:\n{context}\n\nQuestion: {question}"
with client.start_as_current_observation(
name="generation", as_type="generation", model="gpt-4o", input=prompt
) as gen:
response = self.generate_response(question, context)
input_tokens = self.count_tokens(prompt)
output_tokens = self.count_tokens(response)
gen.update(
output=response,
usage_details={
"input": input_tokens,
"output": output_tokens,
"total": input_tokens + output_tokens,
},
)
# Step 6: Quality evaluation.
with client.start_as_current_observation(
name="quality-evaluation",
as_type="evaluator",
input={"question": question, "response": response},
) as span:
quality_metrics = self.evaluate_response_quality(question, response, context)
span.update(
output=quality_metrics,
metadata={"quality_score": quality_metrics["overall_score"]},
)
client.update_current_trace(
output=response,
metadata={
"success": True,
"quality_score": quality_metrics["overall_score"],
},
)
return response
The TypeScript equivalent configures tracing once at process startup, then instruments each stage with spans. Put the provider setup in an instrumentation.ts that is imported before any traced code.
typescript
// instrumentation.ts -- imported first, before any traced code.
const provider = new NodeTracerProvider({
spanProcessors: [
new AntsPlatformSpanProcessor({
publicKey: process.env.ANTS_PLATFORM_PUBLIC_KEY,
secretKey: process.env.ANTS_PLATFORM_SECRET_KEY,
baseUrl: 'https://api.agenticants.ai',
}),
],
})
provider.register()
setAntsPlatformTracerProvider(provider)
export { provider }
typescript
startActiveObservation,
startObservation,
updateActiveTrace,
} from 'ants-platform'
async function monitorRagPipeline(question: string, userId?: string) {
return startActiveObservation(
'complete-rag-pipeline',
async () => {
updateActiveTrace({
input: question,
userId,
metadata: { pipelineVersion: '1.0' },
})
// 1. Embed the query.
const embedding = startObservation(
'embed-query',
{ model: 'text-embedding-3-small', input: question },
{ asType: 'embedding' }
)
const queryVector = await embedQuery(question)
embedding.update({
output: { dimensions: queryVector.length },
usageDetails: { input: countTokens(question), total: countTokens(question) },
})
embedding.end()
// 2. Retrieve -- query in, retrieved docs out.
const retriever = startObservation(
'retrieval',
{ input: { query: question, topK: 5 } },
{ asType: 'retriever' }
)
const retrievedDocs = await retrieveDocuments(queryVector)
retriever.update({
output: retrievedDocs.map((d) => ({ id: d.id, score: d.score, text: d.text })),
metadata: { retrievedCount: retrievedDocs.length },
})
retriever.end()
// 3. Generation consumes the retrieved context.
const context = retrievedDocs.map((d) => d.text).join('\n\n')
const prompt = `Context:\n${context}\n\nQuestion: ${question}`
const gen = startObservation(
'generation',
{ model: 'gpt-4o', input: prompt, modelParameters: { temperature: 0.2 } },
{ asType: 'generation' }
)
const response = await generateResponse(prompt)
gen.update({
output: response,
usageDetails: {
input: countTokens(prompt),
output: countTokens(response),
total: countTokens(prompt) + countTokens(response),
},
})
gen.end()
updateActiveTrace({ output: response, metadata: { success: true } })
return response
},
{ asType: 'agent' }
)
}
Cost Optimization
RAG Cost Tracking
Token usage and cost are derived automatically from generation observations when you record model, input, and output. The example below records per-stage cost metadata on the trace.
python
from ants_platform import AntsPlatform, get_client
AntsPlatform(
public_key=os.environ["ANTS_PLATFORM_PUBLIC_KEY"],
secret_key=os.environ["ANTS_PLATFORM_SECRET_KEY"],
host="https://api.agenticants.ai",
)
class RAGCostMonitor:
def track_rag_costs(self, question: str, retrieved_docs: list, response: str):
client = get_client()
with client.start_as_current_observation(
name="rag-cost-tracking",
as_type="agent",
agent_name="rag-cost-tracking",
):
client.update_current_trace(
input=question,
metadata={"cost_components": ["retrieval", "generation", "evaluation"]},
)
# Calculate retrieval costs.
retrieval_cost = self.calculate_retrieval_cost(retrieved_docs)
# Calculate generation costs.
generation_cost = self.calculate_generation_cost(response)
# Calculate evaluation costs.
evaluation_cost = self.calculate_evaluation_cost(question, response)
total_cost = retrieval_cost + generation_cost + evaluation_cost
client.update_current_trace(
output={"total_cost": total_cost},
metadata={
"retrieval_cost": retrieval_cost,
"generation_cost": generation_cost,
"evaluation_cost": evaluation_cost,
"total_cost": total_cost,
"cost_per_token": total_cost / self.count_tokens(response),
},
)
return total_cost
def calculate_retrieval_cost(self, docs: list) -> float:
# Calculate cost based on vector search operations
return len(docs) * 0.001 # Simplified
def calculate_generation_cost(self, response: str) -> float:
# Calculate cost based on token usage
tokens = self.count_tokens(response)
return tokens * 0.00003 # Simplified
def calculate_evaluation_cost(self, question: str, response: str) -> float:
# Calculate cost for quality evaluation
return 0.001 # Simplified
Cost Optimization Strategies
Aggregate RAG cost analysis and optimization recommendations are not exposed through the SDK. Review cost breakdowns, model comparisons, and FinOps recommendations in the dashboard at https://app.agenticants.ai. Use the cost metadata you record on traces (above) to drive your own optimization logic.
Performance Monitoring
RAG Performance Metrics
There is no metrics or trends query API in the SDK. Average query/retrieval/generation latency, response-quality trends, success rate, and error rate are available as charts and alerts in the dashboard at https://app.agenticants.ai. Configure alert thresholds (for example, high query time or low response quality) there as well.
Performance Optimization
Performance bottleneck analysis is a dashboard feature. Inspect per-stage latency breakdowns and identify bottlenecks (retrieval, generation, context preparation) in the trace explorer at https://app.agenticants.ai, then apply optimizations in your application code.
Quality Monitoring
Response Quality Tracking
Attach evaluation results to a trace with client.update_current_trace(...), or record numeric quality scores against a trace using the scores API on the REST client (AntsPlatformClient). The example below records quality dimensions as trace metadata.
python
from ants_platform import AntsPlatform, get_client
AntsPlatform(
public_key=os.environ["ANTS_PLATFORM_PUBLIC_KEY"],
secret_key=os.environ["ANTS_PLATFORM_SECRET_KEY"],
host="https://api.agenticants.ai",
)
class RAGQualityMonitor:
def monitor_response_quality(self, question: str, response: str, context: str):
client = get_client()
with client.start_as_current_observation(
name="rag-quality-monitoring",
as_type="evaluator",
input={"question": question, "response": response, "context": context},
) as evaluator:
# Evaluate response quality against the retrieved context.
quality_metrics = self.evaluate_response_quality(question, response, context)
evaluator.update(
output=quality_metrics,
metadata={
"overall_quality": quality_metrics["overall_score"],
"relevance": quality_metrics["relevance"],
"accuracy": quality_metrics["accuracy"],
"completeness": quality_metrics["completeness"],
"coherence": quality_metrics["coherence"],
},
)
return quality_metrics
def evaluate_response_quality(self, question: str, response: str, context: str) -> dict:
# Implement comprehensive quality evaluation
return {
"overall_score": 0.85,
"relevance": 0.9,
"accuracy": 0.8,
"completeness": 0.85,
"coherence": 0.8,
}
Best Practices
1. Comprehensive Monitoring
python
from ants_platform import AntsPlatform, get_client
AntsPlatform(
public_key=os.environ["ANTS_PLATFORM_PUBLIC_KEY"],
secret_key=os.environ["ANTS_PLATFORM_SECRET_KEY"],
host="https://api.agenticants.ai",
)
class BestPracticeRAGSystem:
def query_with_monitoring(self, question: str, user_id: str = None):
client = get_client()
with client.start_as_current_observation(
name="best-practice-rag",
as_type="agent",
agent_name="best-practice-rag",
):
client.update_current_trace(
input=question,
user_id=user_id,
metadata={"monitoring_level": "comprehensive"},
)
# Retriever: capture the query as input and the documents as output.
with client.start_as_current_observation(
name="retrieval", as_type="retriever", input={"query": question, "top_k": 5}
) as retriever:
docs = self.retrieve_documents(question)
retriever.update(
output=[{"id": d.id, "score": d.score, "text": d.content[:500]} for d in docs],
metadata={"retrieved_count": len(docs)},
)
# Generation consuming the retrieved context.
context = "\n\n".join(d.content for d in docs)
prompt = f"Context:\n{context}\n\nQuestion: {question}"
with client.start_as_current_observation(
name="generation", as_type="generation", model="gpt-4o", input=prompt
) as gen:
response = self.generate_response(question, docs)
input_tokens = self.count_tokens(prompt)
output_tokens = self.count_tokens(response)
gen.update(
output=response,
usage_details={
"input": input_tokens,
"output": output_tokens,
"total": input_tokens + output_tokens,
},
)
with client.start_as_current_observation(
name="quality-evaluation", as_type="evaluator", input={"question": question}
) as evaluator:
quality = self.evaluate_quality(question, response, docs)
evaluator.update(
output=quality, metadata={"quality_score": quality["overall_score"]}
)
client.update_current_trace(
output=response,
metadata={"success": True, "quality_score": quality["overall_score"]},
)
# Flush before a short-lived process exits.
client.flush()
return response
2. Error Handling and Recovery
typescript
startActiveObservation,
startObservation,
updateActiveTrace,
} from 'ants-platform'
class ResilientRAGSystem {
async queryWithRecovery(question: string, userId?: string) {
return startActiveObservation(
'resilient-rag',
async () => {
updateActiveTrace({
input: question,
userId,
metadata: { retryEnabled: true, fallbackEnabled: true },
})
// Wrap each attempt in a tool observation so failures are captured.
const attempt = startObservation(
'primary-rag',
{ input: question },
{ asType: 'tool' }
)
try {
const response = await this.primaryRAGSystem.query(question)
attempt.update({ output: response, metadata: { system: 'primary' } })
attempt.end()
updateActiveTrace({ output: response, metadata: { success: true, system: 'primary' } })
return response
} catch (error) {
attempt.update({ level: 'ERROR', statusMessage: String(error) })
attempt.end()
// Try fallback system.
const fallback = startObservation(
'fallback-rag',
{ input: question },
{ asType: 'tool' }
)
try {
const fallbackResponse = await this.fallbackRAGSystem.query(question)
fallback.update({ output: fallbackResponse, metadata: { system: 'fallback' } })
fallback.end()
updateActiveTrace({
output: fallbackResponse,
metadata: { success: true, system: 'fallback' },
})
return fallbackResponse
} catch (fallbackError) {
fallback.update({ level: 'ERROR', statusMessage: String(fallbackError) })
fallback.end()
updateActiveTrace({
metadata: { success: false, error: (fallbackError as Error).message },
})
throw fallbackError
}
}
},
{ asType: 'agent' }
)
}
}
Troubleshooting
Common RAG Issues
Issue: Low retrieval quality
Filter your vector-retrieval traces in the trace explorer at https://app.agenticants.ai and sort by the avg_relevance metadata you recorded. Traces with an average relevance below your threshold (for example, 0.7) surface the queries whose retrieval needs tuning. The SDK records the data; querying and filtering happen in the dashboard.
Issue: High generation costs
Open the trace explorer at https://app.agenticants.ai, filter to generation observations, and sort by cost or token usage to find the most expensive generations. Per-model cost breakdowns and FinOps recommendations are available in the dashboard.
Next Steps
- Production Deployment - Learn production best practices
- Cost Optimization - Reduce costs with our optimization guide
- Debugging - Troubleshoot issues with our debugging guide
Example Projects
- Document Q&A System - GitHub Repository
- Knowledge Base Chatbot - GitHub Repository
- Research Assistant - GitHub Repository
Congratulations! You now have comprehensive monitoring for your RAG systems with complete visibility into retrieval quality, generation performance, and end-to-end metrics.