Most agentic AI demos are a single model call wearing a trench coat. This tutorial builds something closer to a production research-and-reasoning system: web sources ingested asynchronously, split into provenance-tracked chunks, retrieved through a hybrid of sparse TF-IDF search and dense OpenAI embeddings, and synthesized by multiple cooperating agents — a planner, a synthesizer, and a repairer — under strict guardrails that force every major claim to rest on retrieved evidence. Episodic memory rounds out the design, letting the system improve its strategy across runs. The full code is linked from the original tutorial.
Step 1: Environment, Keys, and Core Utilities
!pip -q install openai openai-agents pydantic httpx beautifulsoup4 lxml scikit-learn numpy
import os, re, json, time, getpass, asyncio, sqlite3, hashlib
from typing import List, Dict, Tuple, Optional, Any
import numpy as np
import httpx
from bs4 import BeautifulSoup
from pydantic import BaseModel, Field
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
from openai import AsyncOpenAI
from agents import Agent, Runner, SQLiteSession
if not os.environ.get("OPENAI_API_KEY"):
os.environ("OPENAI_API_KEY") = getpass.getpass("Enter your OpenAI API key: ")
if not os.environ.get("OPENAI_API_KEY"):
raise RuntimeError("OPENAI_API_KEY not provided.")
print("✅ OpenAI API key loaded securely.")
oa = AsyncOpenAI(api_key=os.environ("OPENAI_API_KEY"))
def sha1(s: str) -> str:
return hashlib.sha1(s.encode("utf-8", errors="ignore")).hexdigest()
def normalize_url(u: str) -> str:
u = (u or "").strip()
return u.rstrip(").,)"'")
def clean_html_to_text(html: str) -> str:
soup = BeautifulSoup(html, "lxml")
for tag in soup(("script", "style", "noscript")):
tag.decompose()
txt = soup.get_text("n")
txt = re.sub(r"n{3,}", "nn", txt).strip()
txt = re.sub(r"( t)+", " ", txt)
return txt
def chunk_text(text: str, chunk_chars: int = 1600, overlap_chars: int = 320) -> List(str):
if not text:
return ()
text = re.sub(r"s+", " ", text).strip()
n = len(text)
step = max(1, chunk_chars - overlap_chars)
chunks = ()
i = 0
while i < n:
chunks.append(text(i:i + chunk_chars))
i += step
return chunks
def canonical_chunk_id(s: str) -> str:
if s is None:
return ""
s = str(s).strip()
s = s.strip("<>"'()(){}")
s = s.rstrip(".,;:")
return s
def inject_exec_summary_citations(exec_summary: str, citations: List(str), allowed_chunk_ids: List(str)) -> str:
exec_summary = exec_summary or ""
cset = ()
for c in citations:
c = canonical_chunk_id(c)
if c and c in allowed_chunk_ids and c not in cset:
cset.append(c)
if len(cset) >= 2:
break
if len(cset) < 2:
for c in allowed_chunk_ids:
if c not in cset:
cset.append(c)
if len(cset) >= 2:
break
if len(cset) >= 2:
needed = (c for c in cset if c not in exec_summary)
if needed:
exec_summary = exec_summary.strip()
if exec_summary and not exec_summary.endswith("."):
exec_summary += "."
exec_summary += f" (cite: {cset(0)}) (cite: {cset(1)})"
return exec_summaryThe first block sets up the environment, loads the OpenAI API key securely, and initializes the utilities everything else depends on: hashing, URL normalization, HTML cleaning, and chunking. Defining these once keeps every downstream stage consistent — the same text always hashes to the same identifier, which is what makes deduplication and citation tracking reliable later.
Step 2: Parallel Ingestion With Provenance
async def fetch_many(urls: List(str), timeout_s: float = 25.0, per_url_char_limit: int = 60000) -> Dict(str, str):
headers = {"User-Agent": "Mozilla/5.0 (AgenticAI/4.2)"}
urls = (normalize_url(u) for u in urls)
urls = (u for u in urls if u.startswith("http"))
urls = list(dict.fromkeys(urls))
out: Dict(str, str) = {}
async with httpx.AsyncClient(timeout=timeout_s, follow_redirects=True, headers=headers) as client:
async def _one(url: str):
try:
r = await client.get(url)
r.raise_for_status()
out(url) = clean_html_to_text(r.text)(:per_url_char_limit)
except Exception as e:
out(url) = f"__FETCH_ERROR__ {type(e).__name__}: {e}"
await asyncio.gather(*(_one(u) for u in urls))
return out
def dedupe_texts(sources: Dict(str, str)) -> Dict(str, str):
seen = set()
out = {}
for url, txt in sources.items():
if not isinstance(txt, str) or txt.startswith("__FETCH_ERROR__"):
continue
h = sha1(txt(:25000))
if h in seen:
continue
seen.add(h)
out(url) = txt
return out
class ChunkRecord(BaseModel):
chunk_id: str
url: str
chunk_index: int
text: str
class RetrievalHit(BaseModel):
chunk_id: str
url: str
chunk_index: int
score_sparse: float = 0.0
score_dense: float = 0.0
score_fused: float = 0.0
text: str
class EvidencePack(BaseModel):
query: str
hits: List(RetrievalHit)Multiple web sources are fetched in parallel and aggressively deduplicated to avoid redundant evidence. Raw pages become structured text, and core data models represent chunks and retrieval hits. The critical property: every piece of text can be traced back to a specific source and section index. Provenance is not a nice-to-have here — it is what makes the guardrails in later steps enforceable.
Step 3: Episodic Memory in SQLite
EPISODE_DB = "agentic_episode_memory.db"
def episode_db_init():
con = sqlite3.connect(EPISODE_DB)
cur = con.cursor()
cur.execute("""
CREATE TABLE IF NOT EXISTS episodes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts INTEGER NOT NULL,
question TEXT NOT NULL,
urls_json TEXT NOT NULL,
retrieval_queries_json TEXT NOT NULL,
useful_sources_json TEXT NOT NULL
)
""")
con.commit()
con.close()
def episode_store(question: str, urls: List(str), retrieval_queries: List(str), useful_sources: List(str)):
con = sqlite3.connect(EPISODE_DB)
cur = con.cursor()
cur.execute(
"INSERT INTO episodes(ts, question, urls_json, retrieval_queries_json, useful_sources_json) VALUES(?,?,?,?,?)",
(int(time.time()), question, json.dumps(urls), json.dumps(retrieval_queries), json.dumps(useful_sources)),
)
con.commit()
con.close()
def episode_recall(question: str, top_k: int = 2) -> List(Dict(str, Any)):
con = sqlite3.connect(EPISODE_DB)
cur = con.cursor()
cur.execute("SELECT ts, question, urls_json, retrieval_queries_json, useful_sources_json FROM episodes ORDER BY ts DESC LIMIT 200")
rows = cur.fetchall()
con.close()
q_tokens = set(re.findall(r"(A-Za-z){3,}", (question or "").lower()))
scored = ()
for ts, q2, u, rq, us in rows:
t2 = set(re.findall(r"(A-Za-z){3,}", (q2 or "").lower()))
if not t2:
continue
score = len(q_tokens & t2) / max(1, len(q_tokens))
if score > 0:
scored.append((score, {
"ts": ts,
"question": q2,
"urls": json.loads(u),
"retrieval_queries": json.loads(rq),
"useful_sources": json.loads(us),
}))
scored.sort(key=lambda x: x(0), reverse=True)
return (x(1) for x in scored(:top_k))
episode_db_init()An episodic memory layer, backed by SQLite, lets the system remember previous runs: the questions asked, the retrieval strategies used, and which sources proved useful. Lightweight similarity-based recall biases future planning toward historically successful patterns — a modest mechanism that compounds, because the system stops rediscovering the same strategy from scratch on every run.
Step 4: Hybrid Retrieval With Rank Fusion
class HybridIndex:
def __init__(self):
self.records: List(ChunkRecord) = ()
self.tfidf: Optional(TfidfVectorizer) = None
self.tfidf_mat = None
self.emb_mat: Optional(np.ndarray) = None
def build_sparse(self):
corpus = (r.text for r in self.records) if self.records else ("")
self.tfidf = TfidfVectorizer(stop_words="english", ngram_range=(1, 2), max_features=80000)
self.tfidf_mat = self.tfidf.fit_transform(corpus)
def search_sparse(self, query: str, k: int) -> List(Tuple(int, float)):
if not self.records or self.tfidf is None or self.tfidf_mat is None:
return ()
qv = self.tfidf.transform((query))
sims = cosine_similarity(qv, self.tfidf_mat).flatten()
top = np.argsort(-sims)(:k)
return ((int(i), float(sims(i))) for i in top)
def set_dense(self, mat: np.ndarray):
self.emb_mat = mat.astype(np.float32)
def search_dense(self, q_emb: np.ndarray, k: int) -> List(Tuple(int, float)):
if self.emb_mat is None or not self.records:
return ()
M = self.emb_mat
q = q_emb.astype(np.float32).reshape(1, -1)
M_norm = M / (np.linalg.norm(M, axis=1, keepdims=True) + 1e-9)
q_norm = q / (np.linalg.norm(q) + 1e-9)
sims = (M_norm @ q_norm.T).flatten()
top = np.argsort(-sims)(:k)
return ((int(i), float(sims(i))) for i in top)
def rrf_fuse(rankings: List(List(int)), k: int = 60) -> Dict(int, float):
scores: Dict(int, float) = {}
for r in rankings:
for pos, idx in enumerate(r, start=1):
scores(idx) = scores.get(idx, 0.0) + 1.0 / (k + pos)
return scores
HYBRID = HybridIndex()
ALLOWED_URLS: List(str) = ()
EMBED_MODEL = "text-embedding-3-small"
async def embed_batch(texts: List(str)) -> np.ndarray:
resp = await oa.embeddings.create(model=EMBED_MODEL, input=texts, encoding_format="float")
vecs = (np.array(item.embedding, dtype=np.float32) for item in resp.data)
return np.vstack(vecs) if vecs else np.zeros((0, 0), dtype=np.float32)
async def embed_texts(texts: List(str), batch_size: int = 96, max_concurrency: int = 3) -> np.ndarray:
sem = asyncio.Semaphore(max_concurrency)
mats: List(Tuple(int, np.ndarray)) = ()
async def _one(start: int, batch: List(str)):
async with sem:
m = await embed_batch(batch)
mats.append((start, m))
tasks = ()
for start in range(0, len(texts), batch_size):
batch = (t(:7000) for t in texts(start:start + batch_size))
tasks.append(_one(start, batch))
await asyncio.gather(*tasks)
mats.sort(key=lambda x: x(0))
emb = np.vstack((m for _, m in mats)) if mats else np.zeros((len(texts), 0), dtype=np.float32)
if emb.shape(0) != len(texts):
raise RuntimeError(f"Embedding rows mismatch: got {emb.shape(0)} expected {len(texts)}")
return emb
async def embed_query(query: str) -> np.ndarray:
m = await embed_batch((query(:7000)))
return m(0) if m.shape(0) else np.zeros((0,), dtype=np.float32)
async def build_index(urls: List(str), max_chunks_per_url: int = 60):
global ALLOWED_URLS
fetched = await fetch_many(urls)
fetched = dedupe_texts(fetched)
records: List(ChunkRecord) = ()
allowed: List(str) = ()
for url, txt in fetched.items():
if not isinstance(txt, str) or txt.startswith("__FETCH_ERROR__"):
continue
allowed.append(url)
chunks = chunk_text(txt)(:max_chunks_per_url)
for i, ch in enumerate(chunks):
cid = f"{sha1(url)}:{i}"
records.append(ChunkRecord(chunk_id=cid, url=url, chunk_index=i, text=ch))
if not records:
err_view = {normalize_url(u): fetched.get(normalize_url(u), "") for u in urls}
raise RuntimeError("No sources fetched successfully.n" + json.dumps(err_view, indent=2)(:4000))
ALLOWED_URLS = allowed
HYBRID.records = records
HYBRID.build_sparse()
texts = (r.text for r in HYBRID.records)
emb = await embed_texts(texts, batch_size=96, max_concurrency=3)
HYBRID.set_dense(emb)The retrieval index combines sparse TF-IDF search with dense embeddings, merged through reciprocal rank fusion so the two signals complement rather than compete. Sparse retrieval excels at exact terms — names, error codes, rare keywords — while dense retrieval catches paraphrases and conceptual matches; fusing them buys high recall without giving up precision. The index is built once per run and reused across all queries for efficiency.
Step 5: Evidence Packs and Strict Schemas
def build_evidence_pack(query: str, sparse: List(Tuple(int,float)), dense: List(Tuple(int,float)), k: int = 10) -> EvidencePack:
sparse_rank = (i for i,_ in sparse)
dense_rank = (i for i,_ in dense)
sparse_scores = {i:s for i,s in sparse}
dense_scores = {i:s for i,s in dense}
fused = rrf_fuse((sparse_rank, dense_rank), k=60) if dense_rank else rrf_fuse((sparse_rank), k=60)
top = sorted(fused.keys(), key=lambda i: fused(i), reverse=True)(:k)
hits: List(RetrievalHit) = ()
for idx in top:
r = HYBRID.records(idx)
hits.append(RetrievalHit(
chunk_id=r.chunk_id, url=r.url, chunk_index=r.chunk_index,
score_sparse=float(sparse_scores.get(idx, 0.0)),
score_dense=float(dense_scores.get(idx, 0.0)),
score_fused=float(fused.get(idx, 0.0)),
text=r.text
))
return EvidencePack(query=query, hits=hits)
async def gather_evidence(queries: List(str), per_query_k: int = 10, sparse_k: int = 60, dense_k: int = 60):
evidence: List(EvidencePack) = ()
useful_sources_count: Dict(str, int) = {}
all_chunk_ids: List(str) = ()
for q in queries:
sparse = HYBRID.search_sparse(q, k=sparse_k)
q_emb = await embed_query(q)
dense = HYBRID.search_dense(q_emb, k=dense_k)
pack = build_evidence_pack(q, sparse, dense, k=per_query_k)
evidence.append(pack)
for h in pack.hits(:6):
useful_sources_count(h.url) = useful_sources_count.get(h.url, 0) + 1
for h in pack.hits:
all_chunk_ids.append(h.chunk_id)
useful_sources = sorted(useful_sources_count.keys(), key=lambda u: useful_sources_count(u), reverse=True)
all_chunk_ids = sorted(list(dict.fromkeys(all_chunk_ids)))
return evidence, useful_sources(:8), all_chunk_ids
class Plan(BaseModel):
objective: str
subtasks: List(str)
retrieval_queries: List(str)
acceptance_checks: List(str)
class UltraAnswer(BaseModel):
title: str
executive_summary: str
architecture: List(str)
retrieval_strategy: List(str)
agent_graph: List(str)
implementation_notes: List(str)
risks_and_limits: List(str)
citations: List(str)
sources: List(str)
def normalize_answer(ans: UltraAnswer, allowed_chunk_ids: List(str)) -> UltraAnswer:
data = ans.model_dump()
data("citations") = (canonical_chunk_id(x) for x in (data.get("citations") or ()))
data("citations") = (x for x in data("citations") if x in allowed_chunk_ids)
data("executive_summary") = inject_exec_summary_citations(data.get("executive_summary",""), data("citations"), allowed_chunk_ids)
return UltraAnswer(**data)
def validate_ultra(ans: UltraAnswer, allowed_chunk_ids: List(str)) -> None:
extras = (u for u in ans.sources if u not in ALLOWED_URLS)
if extras:
raise ValueError(f"Non-allowed sources in output: {extras}")
cset = set(ans.citations or ())
missing = (cid for cid in cset if cid not in set(allowed_chunk_ids))
if missing:
raise ValueError(f"Citations reference unknown chunk_ids (not retrieved): {missing}")
if len(cset) < 6:
raise ValueError("Need at least 6 distinct chunk_id citations in ultra mode.")
es_text = ans.executive_summary or ""
es_count = sum(1 for cid in cset if cid in es_text)
if es_count < 2:
raise ValueError("Executive summary must include at least 2 chunk_id citations verbatim.")
PLANNER = Agent(
name="Planner",
model="gpt-4o-mini",
instructions=(
"Return a technical Plan schema.n"
"Make 10-16 retrieval_queries.n"
"Acceptance must include: at least 6 citations and exec_summary contains at least 2 citations verbatim."
),
output_type=Plan,
)
SYNTHESIZER = Agent(
name="Synthesizer",
model="gpt-4o-mini",
instructions=(
"Return UltraAnswer schema.n"
"Hard constraints:n"
"- executive_summary MUST include at least TWO citations verbatim as: (cite: ).n"
"- citations must be chosen ONLY from ALLOWED_CHUNK_IDS list.n"
"- citations list must include at least 6 unique chunk_ids.n"
"- sources must be subset of allowed URLs.n"
),
output_type=UltraAnswer,
)
FIXER = Agent(
name="Fixer",
model="gpt-4o-mini",
instructions=(
"Repair to satisfy guardrails.n"
"Ensure executive_summary includes at least TWO citations verbatim.n"
"Choose citations ONLY from ALLOWED_CHUNK_IDS list.n"
"Return UltraAnswer schema."
),
output_type=UltraAnswer,
)
session = SQLiteSession("ultra_agentic_user", "ultra_agentic_session.db") Evidence gathering runs multiple targeted queries, combines sparse and dense results, and assembles evidence packs carrying scores and provenance. Plans and final answers must conform to strict schemas, and citations are normalized and validated against the actual retrieved segment IDs — the guardrail that keeps answers grounded and auditable rather than confidently decorative.
Step 6: The Full Agentic Loop — Plan, Synthesize, Verify, Repair
async def run_ultra_agentic(question: str, urls: List(str), max_repairs: int = 2) -> UltraAnswer:
await build_index(urls)
recall_hint = json.dumps(episode_recall(question, top_k=2), indent=2)(:2000)
plan_res = await Runner.run(
PLANNER,
f"Question:n{question}nnAllowed URLs:n{json.dumps(ALLOWED_URLS, indent=2)}nnRecall:n{recall_hint}n",
session=session
)
plan: Plan = plan_res.final_output
queries = (plan.retrieval_queries or ())(:16)
evidence_packs, useful_sources, allowed_chunk_ids = await gather_evidence(queries)
evidence_json = json.dumps((p.model_dump() for p in evidence_packs), indent=2)(:16000)
allowed_chunk_ids_json = json.dumps(allowed_chunk_ids(:200), indent=2)
draft_res = await Runner.run(
SYNTHESIZER,
f"Question:n{question}nnAllowed URLs:n{json.dumps(ALLOWED_URLS, indent=2)}nn"
f"ALLOWED_CHUNK_IDS:n{allowed_chunk_ids_json}nn"
f"Evidence packs:n{evidence_json}nn"
"Return UltraAnswer.",
session=session
)
draft = normalize_answer(draft_res.final_output, allowed_chunk_ids)
last_err = None
for i in range(max_repairs + 1):
try:
validate_ultra(draft, allowed_chunk_ids)
episode_store(question, ALLOWED_URLS, plan.retrieval_queries, useful_sources)
return draft
except Exception as e:
last_err = str(e)
if i >= max_repairs:
draft = normalize_answer(draft, allowed_chunk_ids)
validate_ultra(draft, allowed_chunk_ids)
return draft
fixer_res = await Runner.run(
FIXER,
f"Question:n{question}nnAllowed URLs:n{json.dumps(ALLOWED_URLS, indent=2)}nn"
f"ALLOWED_CHUNK_IDS:n{allowed_chunk_ids_json}nn"
f"Guardrail error:n{last_err}nn"
f"Draft:n{json.dumps(draft.model_dump(), indent=2)(:12000)}nn"
f"Evidence packs:n{evidence_json}nn"
"Return corrected UltraAnswer that passes guardrails.",
session=session
)
draft = normalize_answer(fixer_res.final_output, allowed_chunk_ids)
raise RuntimeError(f"Unexpected failure: {last_err}")
question = (
"Design a production-lean but advanced agentic AI workflow in Python with hybrid retrieval, "
"provenance-first citations, critique-and-repair loops, and episodic memory. "
"Explain why each layer matters, failure modes, and evaluation."
)
urls = (
"https://openai.github.io/openai-agents-python/",
"https://openai.github.io/openai-agents-python/agents/",
"https://openai.github.io/openai-agents-python/running_agents/",
"https://github.com/openai/openai-agents-python",
)
ans = await run_ultra_agentic(question, urls, max_repairs=2)
print("nTITLE:n", ans.title)
print("nEXECUTIVE SUMMARY:n", ans.executive_summary)
print("nARCHITECTURE:")
for x in ans.architecture:
print("-", x)
print("nRETRIEVAL STRATEGY:")
for x in ans.retrieval_strategy:
print("-", x)
print("nAGENT GRAPH:")
for x in ans.agent_graph:
print("-", x)
print("nIMPLEMENTATION NOTES:")
for x in ans.implementation_notes:
print("-", x)
print("nRISKS & LIMITS:")
for x in ans.risks_and_limits:
print("-", x)
print("nCITATIONS (chunk_ids):")
for c in ans.citations:
print("-", c)
print("nSOURCES:")
for s in ans.sources:
print("-", s)The final stage chains planning, synthesis, verification, and repair into an async-safe pipeline. Outputs that fail validation are automatically retried and repaired until they pass every check, without human intervention. The run ends by printing a fully grounded response with verifiable citations.
Why the Design Choices Matter
The pipeline is deliberately robust to the failure modes that sink agentic RAG systems in practice: unstable embedding dimensions, citation drift, and executive summaries that quietly detach from their evidence. Output is validated against an allow-list of sources and retrieved section IDs, citations are normalized automatically, and deterministic citations are injected when necessary — guaranteeing compliance without sacrificing correctness. Hybrid retrieval, critique-and-repair loops, and episodic memory together form a reusable foundation.
Limitations and What to Watch
A grounded pipeline is only as good as its sources; guardrails guarantee that claims trace to retrieved text, not that the retrieved text is true, so source selection still deserves human judgment. Critique-and-repair loops multiply model calls, and costs scale with retry budgets — worth metering before pointing the system at large corpora. TF-IDF also needs re-indexing as corpora change, and SQLite memory is single-machine by design; production deployments at scale would swap in a vector database and shared store. The original tutorial suggests the right next investments: claim-to-evidence coverage scoring, adversarial red-teaming, and regression testing. For the coordination side of multi-agent design, see this related walkthrough of agent-to-agent collaboration patterns, and for testing discipline, this guide to applying CI and unit testing to data solutions.
