← Lesson
Agentic RAG
Joeven
Run
Reset
Python loads on first run
import json CHUNKS = { "runbook": "Runner OOM: raise worker memory, then replay the DLQ.", "billing": "Refunds take 5-7 days. Never cash.", "dlq": "DLQ replay: CLI cmd replay-dlq --since 24h. Needs worker healthy.", } def retrieve(query, k=1): q = set(query.lower().split()) scored = [] for cid, text in CHUNKS.items(): score = len(q & set(text.lower().replace(":", " ").replace(",", " ").split())) scored.append((score, cid, text)) scored.sort(reverse=True) return [{"id": cid, "text": text, "score": sc} for sc, cid, text in scored[:k] if sc > 0] def enough(hits): blob = " ".join(h["text"].lower() for h in hits) return "replay-dlq" in blob and "memory" in blob def agentic(question, max_hops=3): queries = [question] evidence = [] seen = set() for hop in range(max_hops): hits = retrieve(queries[-1], k=1) ids = tuple(h["id"] for h in hits) if ids in seen: break seen.add(ids) evidence.extend(hits) print(json.dumps({"hop": hop, "query": queries[-1], "hits": [h["id"] for h in hits]})) if enough(evidence): return { "answer": "Raise worker memory, then replay-dlq --since 24h.", "hops": hop + 1, "citations": [h["id"] for h in evidence], } blob = " ".join(h["text"] for h in hits) if "DLQ" in blob and "replay-dlq" not in blob: queries.append("DLQ replay CLI") else: queries.append(question + " runbook memory") return {"answer": "give up", "hops": max_hops, "citations": []} print("FINAL", json.dumps(agentic("worker OOM then what about the DLQ")))
Run to execute this in your browser. Nothing is sent to a server.