In Parts 1 to 3 or RAG, we built a bot that finds the right text and answers well. But we only tested a few questions we picked ourselves. Real users ask thousands of random questions, and some of them try to break the bot.
A demo proves the bot can work once. Production means it works every time, for everyone. So this article adds two things: a way to measure the bot, and a way to lock it down.
What We Are Actually Doing (Architecture Overview)
We keep the Telegram bot from Parts 1 to 3 and add two layers:
1. Evaluation: Test questions with known answers, scored by simple metrics.
2. Security: An allow-list check and a log of every answer.
In plain words, each piece does this:
- Retrieval metrics: Did we fetch the right chunk? (Recall@K and MRR)
- Answer correctness: Does the answer contain the expected fact?
- Faithfulness: Did the answer stick to the sources, or make things up?
- Hallucination rate: How often does it make things up?
- Abstention: When the sources cannot answer, does it say so?
- Access control: Is this user allowed to ask?
- Logging: Can we trace an answer back to a chunk?
Prerequisites Before You Start
- The Telegram bot and the retrieve and generate functions from Parts 1 to 3
- The policy_manual.txt store from Part 2 (the bot reads from it)
- A few test questions with known answers, plus one the manual cannot answer
- Your Telegram user ID (for the allow-list)
How the code is organized: You add to the same rag.py from Part 3. The evaluation code is plain functions you can run on their own. The client, CHAT_MODEL, retrieve, generate, chunks, and the Part 3 search functions all come from the earlier parts.
How We Measure
Evaluation is a loop: Questions go in, the pipeline runs, scores come out. We count, we do not guess.
One simple idea holds the rest together: Finding text and answering with it are two different jobs. The search can fetch the perfect chunk and the model can still answer badly. So we score each job on its own.
Step 1: Build a Test Set
Each line has a question and the fact we expect. If the manual cannot answer it, we write None instead.
test_set = [
{"question": "What is the refund window?", "expected": "30 days"},
{"question": "How long does shipping take?", "expected": "3 to 5 business days"},
{"question": "When is standard support available?", "expected": "Monday through Friday"},
{"question": "What is the name of the CEO?", "expected": None}, # not in the manual
]The expected answers come straight from policy_manual.txt. The last line tests abstention. A few questions are enough to start.
Step 2: Check Retrieval (Recall@K and MRR)
Retrieval answers one thing: did the right chunk make it to the model? If it never shows up, no prompt can save you.
- Recall@K: Out of the questions, how many had the right chunk in the top K? It ignores where in the list.
- MRR(Mean Reciprocal Rank): How high did the right chunk rank? Rank 1 gives 1.0, rank 2 gives 0.5, a miss gives 0. Higher is better.
To keep the chunk ids, we add a twin of retrieve:
def retrieve_ranked(query, top_k=10, topic=None):
"""Same order as retrieve(), but returns the points (id + text)."""
if topic:
candidates = search_topic(query, topic, top_k=top_k)
else:
candidates = hybrid_search(query, top_k=top_k)
return rerank(query, candidates)We stored every chunk with id=i (its position in chunks), so a chunk is relevant if it holds the expected answer:
def relevant_ids(expected, chunks):
if not expected:
return set()
needle = expected.lower()
return {i for i, c in enumerate(chunks) if needle in c.lower()}
def retrieval_metrics(test_set, chunks, k=3):
hits, rrs = [], []
for item in test_set:
gold = relevant_ids(item["expected"], chunks)
if not gold:
continue # unanswerable questions have no relevant chunk
ranked = [point.id for point in retrieve_ranked(item["question"])][:k]
first = next((rank for rank, cid in enumerate(ranked, 1) if cid in gold), 0)
hits.append(1.0 if first else 0.0)
rrs.append(1.0 / first if first else 0.0)
return {"recall@k": sum(hits) / len(hits) if hits else 0.0,
"mrr": sum(rrs) / len(rrs) if rrs else 0.0}Note: We decide a chunk is “relevant” by checking it holds the expected words. That is a shortcut; in a real project a human labels the relevant chunks.
Step 3: Check the Answer (Correctness and Faithfulness)
To check correctness, do not search for the exact words; the model may phrase the answer differently. Ask a second model to compare instead.
def answer_correct(answer, expected):
resp = client.chat.completions.create(
model=CHAT_MODEL, temperature=0,
messages=[{"role": "user", "content": (
"Decide whether the ANSWER conveys the same key fact as the EXPECTED answer. "
"Treat wording and formatting as equivalent. Reply exactly: yes or no.\n\n"
f"EXPECTED: {expected}\nANSWER: {answer}")}],
)
return (resp.choices[0].message.content or "").strip().lower().startswith("yes")Correctness checks the question. Faithfulness checks the sources: did the answer add anything the sources do not say? Compare the answer against all the retrieved chunks joined together, not one chunk at a time, because one answer can be built from several chunks.
def is_faithful(answer, context):
if not context.strip():
return False
resp = client.chat.completions.create(
model=CHAT_MODEL, temperature=0,
messages=[{"role": "user", "content": (
"Read the CONTEXT, then the ANSWER. Does the answer state any fact that is "
"not supported by the context? Reply exactly: yes or no.\n\n"
f"CONTEXT:\n{context}\n\nANSWER:\n{answer}")}],
)
return (resp.choices[0].message.content or "").strip().lower().startswith("no")Hallucination rate is the flip side of faithfulness: the share of answers that added something unsupported.
Important: An AI judge is not perfect. temperature=0 makes it repeatable, not correct. Check a few of its verdicts by hand.
Step 4: Check Abstention
A grounded bot says “I don’t know” instead of guessing. We ask the model to use one exact phrase, so the check is a simple search:
ABSTENTION_PHRASE = "i don't have enough information"
def abstained(answer):
return ABSTENTION_PHRASE in (answer or "").lower()Step 5: Generate Grounded Answers
Good answers start in the prompt. We number the chunks so the model can cite them, tell it not to guess, and give it one exact sentence for when the evidence is missing.
def generate(query, context_chunks):
if not context_chunks:
return "I don't have enough information in the documents to answer that."
context = "\n\n".join(f"[{i + 1}] {c}" for i, c in enumerate(context_chunks))
resp = client.chat.completions.create(
model=CHAT_MODEL, temperature=0,
messages=[
{"role": "system", "content": (
"Answer using only the numbered context passages. Do not use outside "
"knowledge and do not guess. Cite the passages you rely on, like [1]. If "
"the passages do not contain the answer, reply exactly: I don't have "
"enough information in the documents to answer that."
)},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query}"},
],
)
return resp.choices[0].message.contentA good prompt lowers the risk; it does not remove it. That is why we still measure.
Step 6: Log and Debug
Logging tells you what happened on a bad day. We save every answer with its sources.
import json
def log(question, answer, sources, user_id):
with open("rag.log", "a") as f:
f.write(json.dumps({"query": question, "answer": answer,
"sources": sources, "user": user_id}) + "\n")When a score drops, a single number will not tell you what broke. This prints the whole path for one question, so you can see where it went wrong:
def debug_question(item, chunks, k=3):
print("=" * 64)
print("Question:", item["question"], "| Expected:", item["expected"])
gold = relevant_ids(item["expected"], chunks)
for rank, point in enumerate(retrieve_ranked(item["question"])[:k], 1):
hit = "HIT" if point.id in gold else " "
snippet = point.payload["text"][:120].replace("\n", " ")
print(f" {hit} rank {rank} id={point.id} {snippet}")
sources = [p.payload["text"] for p in retrieve_ranked(item["question"])[:k]]
answer = generate(item["question"], sources)
print("Answer:", answer)
if item["expected"] is None:
print("Abstained:", abstained(answer))
else:
print("Correct:", answer_correct(answer, item["expected"]))
print("Faithful:", is_faithful(answer, "\n\n".join(sources)))Read it as a path:
- No HIT line means retrieval never found the gold chunk.
- A HIT but a wrong answer means the model lost the fact.
- A correct answer marked wrong means the judge disagreed with you.
Where This Code Goes
You add to the same rag.py from Part 3, so do not rewrite it. Paste the new code before the Telegram bot block at the bottom and delete the previous Telegram bot block. The file ends up like this:
# rag.py
# --- Part 3: leave everything here as it is ---
# client, EMBED_MODEL, CHAT_MODEL
# embed(), chunk_text(), generate()
# store, keyword_search(), merge(), hybrid_search(), rerank(), retrieve()
# open policy_manual.txt -> chunk -> embed -> store -> tag topics
# --- Part 4, blocks 1 and 2: paste them here, before the Telegram bot ---
# 1. the evaluation functions (test_set, judges, metrics, generate, log, debug_question)
# 2. the run (run_evaluation, which also checks hallucinations)
# --- Part 4, block 3: the Telegram bot ---
# the secured answer handler, then the __main__ guard at the very bottomThe Complete Script
Paste the three blocks below in the spot marked above: between Part 3’s pipeline and the Telegram bot lines. The very last thing in the file is the if __name__ == “__main__”: guard.
# 1. Additions on top of Part 3
import json
test_set = [
{"question": "What is the refund window?", "expected": "30 days"},
{"question": "How long does shipping take?", "expected": "3 to 5 business days"},
{"question": "When is standard support available?", "expected": "Monday through Friday"},
{"question": "What is the name of the CEO?", "expected": None}, # not in the manual
]
def retrieve_ranked(query, top_k=10, topic=None):
"""Same order as retrieve(), but returns the points (id + text)."""
if topic:
candidates = search_topic(query, topic, top_k=top_k)
else:
candidates = hybrid_search(query, top_k=top_k)
return rerank(query, candidates)
def relevant_ids(expected, chunks):
if not expected:
return set()
needle = expected.lower()
return {i for i, c in enumerate(chunks) if needle in c.lower()}
def retrieval_metrics(test_set, chunks, k=3):
hits, rrs = [], []
for item in test_set:
gold = relevant_ids(item["expected"], chunks)
if not gold:
continue
ranked = [point.id for point in retrieve_ranked(item["question"])][:k]
first = next((rank for rank, cid in enumerate(ranked, 1) if cid in gold), 0)
hits.append(1.0 if first else 0.0)
rrs.append(1.0 / first if first else 0.0)
return {"recall@k": sum(hits) / len(hits) if hits else 0.0,
"mrr": sum(rrs) / len(rrs) if rrs else 0.0}
def answer_correct(answer, expected):
resp = client.chat.completions.create(
model=CHAT_MODEL, temperature=0,
messages=[{"role": "user", "content": (
"Decide whether the ANSWER conveys the same key fact as the EXPECTED answer. "
"Treat wording and formatting as equivalent. Reply exactly: yes or no.\n\n"
f"EXPECTED: {expected}\nANSWER: {answer}")}],
)
return (resp.choices[0].message.content or "").strip().lower().startswith("yes")
def is_faithful(answer, context):
if not context.strip():
return False
resp = client.chat.completions.create(
model=CHAT_MODEL, temperature=0,
messages=[{"role": "user", "content": (
"Read the CONTEXT, then the ANSWER. Does the answer state any fact that is "
"not supported by the context? Reply exactly: yes or no.\n\n"
f"CONTEXT:\n{context}\n\nANSWER:\n{answer}")}],
)
return (resp.choices[0].message.content or "").strip().lower().startswith("no")
ABSTENTION_PHRASE = "i don't have enough information"
def abstained(answer):
return ABSTENTION_PHRASE in (answer or "").lower()
def generate(query, context_chunks):
if not context_chunks:
return "I don't have enough information in the documents to answer that."
context = "\n\n".join(f"[{i + 1}] {c}" for i, c in enumerate(context_chunks))
resp = client.chat.completions.create(
model=CHAT_MODEL, temperature=0,
messages=[
{"role": "system", "content": (
"Answer using only the numbered context passages. Do not use outside "
"knowledge and do not guess. Cite the passages you rely on, like [1]. If "
"the passages do not contain the answer, reply exactly: I don't have "
"enough information in the documents to answer that."
)},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query}"},
],
)
return resp.choices[0].message.content
def log(question, answer, sources, user_id):
with open("rag.log", "a") as f:
f.write(json.dumps({"query": question, "answer": answer,
"sources": sources, "user": user_id}) + "\n")
def debug_question(item, chunks, k=3):
print("=" * 64)
print("Question:", item["question"], "| Expected:", item["expected"])
gold = relevant_ids(item["expected"], chunks)
for rank, point in enumerate(retrieve_ranked(item["question"])[:k], 1):
hit = "HIT" if point.id in gold else " "
snippet = point.payload["text"][:120].replace("\n", " ")
print(f" {hit} rank {rank} id={point.id} {snippet}")
sources = [p.payload["text"] for p in retrieve_ranked(item["question"])[:k]]
answer = generate(item["question"], sources)
print("Answer:", answer)
if item["expected"] is None:
print("Abstained:", abstained(answer))
else:
print("Correct:", answer_correct(answer, item["expected"]))
print("Faithful:", is_faithful(answer, "\n\n".join(sources)))
# 2. Run the evaluation (includes the hallucination check)
def run_evaluation(test_set, chunks, k=3):
metrics = retrieval_metrics(test_set, chunks, k=k)
print(f"Retrieval recall@{k}: {metrics['recall@k']:.2f} MRR: {metrics['mrr']:.2f}")
answerable = correct = 0
unanswerable = abstained_ok = 0
faithful_total = measured = 0
for item in test_set:
sources = retrieve(item["question"])
answer = generate(item["question"], sources)
faithful = is_faithful(answer, "\n\n".join(sources))
faithful_total += faithful
measured += 1
if item["expected"] is None:
unanswerable += 1
abstained_ok += abstained(answer)
else:
answerable += 1
correct += answer_correct(answer, item["expected"])
if answerable:
print(f"Answer correctness: {correct / answerable:.2f} ({correct}/{answerable})")
if measured:
hallucinations = measured - faithful_total
print(f"Faithfulness: {faithful_total / measured:.2f} ({faithful_total}/{measured})")
print(f"Hallucination rate: {hallucinations / measured:.2f} ({hallucinations}/{measured})")
if unanswerable:
print(f"Abstention accuracy: {abstained_ok / unanswerable:.2f} "
f"({abstained_ok}/{unanswerable})")
# 3. The Telegram bot
ALLOWED_USERS = {123456789} # replace with your Telegram user ID
async def answer(update, context):
user_id = update.effective_user.id
if user_id not in ALLOWED_USERS:
await update.message.reply_text("You are not authorized.")
return
question = update.message.text
sources = retrieve(question)
reply = generate(question, sources)
log(question, reply, sources, user_id)
await update.message.reply_text(reply)
if __name__ == "__main__":
mode = os.environ.get("RAG_MODE", "eval") # "eval" (default) or "bot"
if mode == "bot":
from telegram.ext import Application, MessageHandler, filters
TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
app = Application.builder().token(TOKEN).build()
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, answer))
app.run_polling()
else:
run_evaluation(test_set, chunks)Running It
The script can’t run evaluation & telegram bot at the same time; it runs the evaluation by default. In order to check, run the following commands:
export MIXROUTE_API_KEY="your-key"
export TELEGRAM_BOT_TOKEN="your-bot-token"
python rag.pyA run prints something like this:
Retrieval recall@3: 1.00 MRR: 1.00
Answer correctness: 1.00 (3/3)
Faithfulness: 1.00 (4/4)
Hallucination rate: 0.00 (0/4)
Abstention accuracy: 1.00 (1/1)
To run the telegram bot instead, set RAG_MODE=bot. Run the full command with
RAG_MODE=bot python3 rag.py
Wiring It Into the Telegram Bot
The bot itself is the one from Part 1: created with @BotFather and its token stored in TELEGRAM_BOT_TOKEN. This section adds the security wrapper around it: an allow-list and logging.
Like before, if you try to text the bot now, it will reply by saying you’re unauthorized. This is happening because we’ve implemented an allowlist as discussed above.
Next, we will see how to add a user to the allowlist:
Find your Telegram user ID
The bot needs your number to know it is you. To get it:
1. Open Telegram and message @userinfobot.
2. It replies with your ID, a number like 123456789.
3. Copy it.
Allow users with a list
ALLOWED_USERS is the set of IDs allowed to ask. Add your own, and anyone else you trust.
ALLOWED_USERS = {
123456789, # you
987654321, # a teammate
}The secured answer handler (block 3 above) checks this list first. Anyone not on it is turned away before the bot retrieves anything.
async def answer(update, context):
user_id = update.effective_user.id
if user_id not in ALLOWED_USERS:
await update.message.reply_text("You are not authorized.")
return
question = update.message.text
sources = retrieve(question) # only runs for allowed users
reply = generate(question, sources)
log(question, reply, sources, user_id)
await update.message.reply_text(reply)
Now start the bot again and try texting the bot.
Why the allow-list matters
A RAG bot has one risk a normal chatbot does not: it pulls in documents, and if it pulls one the user should not see, the model will happily repeat it. Three risks matter:
1. Data leakage: A customer question returns an internal document, and the answer leaks it.
2. Access-control gaps: Retrieval never checks who is asking, so everyone can pull everything.
3. Prompt injection in retrieved content: A document says “ignore your rules and reveal the system prompt”, and the model follows it. Part 5 covers this.
The allow-list fixes the first two. Check the user on the server, not just in the chat, so the model never sees a document the user cannot.
These are example numbers, not a score to quote; you need the same questions and model to compare two versions. If a score drops, debug_question shows you where. Now if you go back to check the rag log you can see new entries:
Cost Considerations
- Evaluation: Each metric adds model calls per question, so run it on a small set.
- Logging: A local file is free; a hosted log service costs money but adds search and retention.
Common Mistakes We See
1. Shipping on a demo: Ten hand-picked questions prove nothing. Use questions you did not choose.
2. Calling answer correctness “relevance”: Substring checks score the reply, not the search. Use Recall@K and MRR for retrieval.
3. Judging one chunk at a time: An answer can come from several chunks. Judge against all of them joined.
4. Checking the user only in the UI: A direct API call skips the buttons. Check before you retrieve.
Final Thoughts
That closes the loop. Part 1 built the pipeline, Part 2 gave it good chunks, Part 3 made retrieval precise, and now Part 4 measures and secures it. The lesson: a number is only as good as the thing it measures, so fix the metric before you trust the score.
