On this tutorial, we construct a sophisticated, end-to-end studying pipeline round Atomic-Brokers by wiring collectively typed agent interfaces, structured prompting, and a compact retrieval layer that grounds outputs in actual venture documentation. Additionally, we reveal how you can plan retrieval, retrieve related context, inject it dynamically into an answering agent, and run an interactive loop that turns the setup right into a reusable analysis assistant for any new Atomic Brokers query. Try the FULL CODES right here.
import os, sys, textwrap, time, json, re
from typing import Record, Optionally available, Dict, Tuple
from dataclasses import dataclass
import subprocess
subprocess.check_call([sys.executable, "-m", "pip", "install", "-q",
"atomic-agents", "instructor", "openai", "pydantic",
"requests", "beautifulsoup4", "scikit-learn"])
from getpass import getpass
if not os.environ.get("OPENAI_API_KEY"):
os.environ["OPENAI_API_KEY"] = getpass("Enter OPENAI_API_KEY (input hidden): ").strip()
MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")
from pydantic import Area
from openai import OpenAI
import teacher
from atomic_agents import AtomicAgent, AgentConfig, BaseIOSchema
from atomic_agents.context import SystemPromptGenerator, ChatHistory, BaseDynamicContextProvider
import requests
from bs4 import BeautifulSoupWe set up all required packages, import the core Atomic-Brokers primitives, and arrange Colab-compatible dependencies in a single place. We securely seize the OpenAI API key from the keyboard and retailer it within the atmosphere so downstream code by no means hardcodes secrets and techniques. We additionally lock in a default mannequin title whereas protecting it configurable through an atmosphere variable.
def fetch_url_text(url: str, timeout: int = 20) -> str:
r = requests.get(url, timeout=timeout, headers={"User-Agent": "Mozilla/5.0"})
r.raise_for_status()
soup = BeautifulSoup(r.textual content, "html.parser")
for tag in soup(["script", "style", "nav", "header", "footer", "noscript"]):
tag.decompose()
textual content = soup.get_text("n")
textual content = re.sub(r"[ t]+", " ", textual content)
textual content = re.sub(r"n{3,}", "nn", textual content).strip()
return textual content
def chunk_text(textual content: str, max_chars: int = 1400, overlap: int = 200) -> Record[str]:
if not textual content:
return []
chunks = []
i = 0
whereas i < len(textual content):
chunk = textual content[i:i+max_chars].strip()
if chunk:
chunks.append(chunk)
i += max_chars - overlap
return chunks
def clamp(s: str, n: int = 800) -> str:
s = (s or "").strip()
return s if len(s) <= n else s[:n].rstrip() + "…"We fetch internet pages from the Atomic Brokers repo and docs, then clear them into plain textual content so retrieval turns into dependable. We chunk lengthy paperwork into overlapping segments, preserving context whereas protecting every chunk sufficiently small for rating and quotation. We additionally add a small helper to clamp lengthy snippets so our injected context stays readable.
from sklearn.feature_extraction.textual content import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
@dataclass
class Snippet:
doc_id: str
url: str
chunk_id: int
textual content: str
rating: float
class MiniCorpusRetriever:
def __init__(self, docs: Dict[str, Tuple[str, str]]):
self.objects: Record[Tuple[str, str, int, str]] = []
for doc_id, (url, uncooked) in docs.objects():
for idx, ch in enumerate(chunk_text(uncooked)):
self.objects.append((doc_id, url, idx, ch))
if not self.objects:
increase RuntimeError("No documents were fetched; cannot build TF-IDF index.")
self.vectorizer = TfidfVectorizer(stop_words="english", max_features=50000)
self.matrix = self.vectorizer.fit_transform([it[3] for it in self.objects])
def search(self, question: str, okay: int = 6) -> Record[Snippet]:
qv = self.vectorizer.remodel([query])
sims = cosine_similarity(qv, self.matrix).ravel()
prime = sims.argsort()[::-1][:k]
out = []
for j in prime:
doc_id, url, chunk_id, txt = self.objects[j]
out.append(Snippet(doc_id=doc_id, url=url, chunk_id=chunk_id, textual content=txt, rating=float(sims[j])))
return out
class RetrievedContextProvider(BaseDynamicContextProvider):
def __init__(self, title: str, snippets: Record[Snippet]):
tremendous().__init__(title=title)
self.snippets = snippets
def get_info(self) -> str:
blocks = []
for s in self.snippets:
blocks.append(
f"[{s.doc_id}#{s.chunk_id}] (score={s.score:.3f}) {s.url}n{clamp(s.text, 900)}"
)
return "nn".be a part of(blocks)We construct a mini retrieval system utilizing TF-IDF and cosine similarity over the chunked documentation corpus. We wrap every retrieved chunk in a structured Snippet object to trace doc IDs, chunk IDs, and quotation scores. We then inject top-ranked chunks into the agent’s runtime through a dynamic context supplier, protecting the answering agent grounded. Try the FULL CODES right here.
class PlanInput(BaseIOSchema):
"""Input schema for the planner agent: describes the user's task and how many retrieval queries to draft."""
activity: str = Area(...)
num_queries: int = Area(4)
class PlanOutput(BaseIOSchema):
"""Output schema from the planner agent: retrieval queries, coverage checklist, and safety checks."""
queries: Record[str]
must_cover: Record[str]
safety_checks: Record[str]
class AnswerInput(BaseIOSchema):
"""Input schema for the answering agent: user question plus style constraints."""
query: str
type: str = "concise but advanced"
class AnswerOutput(BaseIOSchema):
"""Output schema for the answering agent: grounded answer, next steps, and which citations were used."""
reply: str
next_steps: Record[str]
used_citations: Record[str]
consumer = teacher.from_openai(OpenAI(api_key=os.environ["OPENAI_API_KEY"]))
planner_prompt = SystemPromptGenerator(
background=[
"You are a rigorous research planner for a small RAG system.",
"You propose retrieval queries that are diverse (lexical + semantic) and designed to find authoritative info.",
"You do NOT answer the task; you only plan retrieval."
],
steps=[
"Read the task.",
"Propose diverse retrieval queries (not too long).",
"List must-cover aspects and safety checks."
],
output_instructions=[
"Return strictly the PlanOutput schema.",
"Queries must be directly usable as search strings.",
"Must-cover should be 4–8 bullets."
]
)
planner = AtomicAgent[PlanInput, PlanOutput](
config=AgentConfig(
consumer=consumer,
mannequin=MODEL,
system_prompt_generator=planner_prompt,
historical past=ChatHistory(),
)
)
answerer_prompt = SystemPromptGenerator(
background=[
"You are an expert technical tutor for Atomic Agents (atomic-agents).",
"You are given retrieved context snippets with IDs like [doc#chunk].",
"You must ground claims in the provided snippets and cite them inline."
],
steps=[
"Read the question and the provided context.",
"Synthesize an accurate answer using only supported facts.",
"Cite claims inline using the provided snippet IDs."
],
output_instructions=[
"Use inline citations like [readme#12] or [docs_home#3].",
"If the context does not support something, say so briefly and suggest what to retrieve next.",
"Return strictly the AnswerOutput schema."
]
)
answerer = AtomicAgent[AnswerInput, AnswerOutput](
config=AgentConfig(
consumer=consumer,
mannequin=MODEL,
system_prompt_generator=answerer_prompt,
historical past=ChatHistory(),
)
)
We outline strict-typed schemas for planner and answerer inputs and outputs, and embrace docstrings to fulfill Atomic Brokers’ schema necessities. We create an Teacher-wrapped OpenAI consumer and configure two Atomic Brokers with specific system prompts and chat historical past. We implement structured outputs so the planner produces queries and the answerer produces a cited response with clear subsequent steps.
SOURCES = {
"readme": "
"docs_home": "
"examples_index": "
}
raw_docs: Dict[str, Tuple[str, str]] = {}
for doc_id, url in SOURCES.objects():
strive:
raw_docs[doc_id] = (url, fetch_url_text(url))
besides Exception:
raw_docs[doc_id] = (url, "")
non_empty = [d for d in raw_docs.values() if d[1].strip()]
if not non_empty:
increase RuntimeError("All source fetches failed or were empty. Check network access in Colab and retry.")
retriever = MiniCorpusRetriever(raw_docs)
def run_atomic_rag(query: str, okay: int = 7, verbose: bool = True) -> AnswerOutput:
t0 = time.time()
plan = planner.run(PlanInput(activity=query, num_queries=4))
all_snips: Record[Snippet] = []
for q in plan.queries:
all_snips.lengthen(retriever.search(q, okay=max(2, okay // 2)))
finest: Dict[Tuple[str, int], Snippet] = {}
for s in all_snips:
key = (s.doc_id, s.chunk_id)
if (key not in finest) or (s.rating > finest[key].rating):
finest[key] = s
snips = sorted(finest.values(), key=lambda x: x.rating, reverse=True)[:k]
ctx = RetrievedContextProvider(title="Retrieved Atomic Agents Context", snippets=snips)
answerer.register_context_provider("retrieved_context", ctx)
out = answerer.run(AnswerInput(query=query, type="concise, advanced, practical"))
if verbose:
print(out.reply)
return out
demo_q = "Teach me Atomic Agents at an advanced level: explain the core building blocks and show how to chain agents with typed schemas and dynamic context."
run_atomic_rag(demo_q, okay=7, verbose=True)
whereas True:
user_q = enter("nYour question> ").strip()
if not user_q or user_q.decrease() in {"exit", "quit"}:
break
run_atomic_rag(user_q, okay=7, verbose=True)We fetch a small set of authoritative Atomic Brokers sources and construct an area retrieval index from them. We implement a full pipeline operate that plans queries, retrieves related context, injects it, and produces a grounded last reply. We end by working a demo question and launching an interactive loop so we are able to maintain asking questions and getting cited solutions.
In conclusion, we accomplished the Atomic-Brokers workflow in Colab, cleanly separating planning, retrieval, answering, and guaranteeing robust typing. We stored the system grounded by injecting solely the highest-signal documentation chunks as dynamic context, and we enforced a quotation self-discipline that makes outputs auditable. From right here, we are able to scale this sample by including extra sources, swapping in stronger retrievers or rerankers, introducing tool-use brokers, and turning the pipeline right into a production-grade analysis assistant that is still each quick and reliable.
Try the FULL CODES right here. Additionally, be at liberty to comply with us on Twitter and don’t overlook to affix our 100k+ ML SubReddit and Subscribe to our E-newsletter. Wait! are you on telegram? now you may be a part of us on telegram as effectively.
