Qu’est-ce que le RAG ? Modèles naïfs, hybrides, graphiques et agents
Une carte pratique de la génération augmentée par la récupération d’informations : RAG vectoriel classique, récupération hybride, GraphRAG et modèles agents — quand chacun convient et ce qui échoue en environnement de production.
RAG (Retrieval-Augmented Generation) fournit des réponses en s’appuyant sur une source de connaissances externe plutôt que uniquement sur la mémoire du modèle. Les passages pertinents sont d’abord récupérés et transmis en tant que contexte afin que le LLM puisse fonder sa réponse. Imaginez un examen avec des livres ouverts : le modèle est l’étudiant, tandis que RAG fournit les pages. Les réponses restent actualisables sans avoir à reformerater le modèle.
Fonctionnement de RAG
Trois étapes :
- Indexation — segmentation des documents en éléments recherchables (base de données vectorielle, graphe, index inversé, …)
- Récupération — identification des meilleurs éléments pour une requête
- Génération — transmission de ces éléments au LLM pour synthétiser une réponse
La qualité de la récupération est déterminante. Un modèle puissant mais doté d’une capacité de récupération faible produit encore des réponses médiocres ; l’essentiel réside dans la manière de trouver le contexte adéquat.
1. RAG naïf
Le schéma le plus simple : intégrer des documents, stocker des vecteurs, et récupérer les résultats en fonction de la similarité au moment de la requête.
Le schéma
Découper les documents en fragments, les intégrer et stocker les vecteurs correspondants. Lors de la requête, intégrer la question, identifier les voisins les plus proches, puis permettre au modèle d’écrire une réponse à partir de ces passages.
Quand l’utiliser
- Prototypes et démos
- Corpus petits et bien structurés (environ moins de ~50 000 fragments)
- Domaines où la similarité sémantique correspond à l’intention (questions-réponses en prose)
Lorsque cela ne fonctionne plus
- Tokens exacts (noms, codes, IDs) que les embeddings ne captent pas
- Absence de filtres métadonnées (date, source, catégorie)
- Méthode top-k bruitée sans réclassification
- Scans complets devenant coûteux avec des millions de fragments
Sur quoi se concentrer dans le code
Indexation dans ChromaDB, une simple opération retrieve() basée sur la similarité cosinus, ainsi qu’un prompt système qui contraint le modèle au contexte fourni :
"""
Naive RAG — the simplest form of Retrieval-Augmented Generation.
Pattern:
1. Chunk documents into small pieces
2. Embed each chunk into a vector
3. Store vectors in a vector database
4. At query time: embed the query, find the closest chunk vectors
5. Pass those chunks as context to the LLM
This is the baseline. It's quick to build and works well for small,
clean corpora where semantic similarity reliably maps to relevance.
"""
import os
from openai import OpenAI
import chromadb
from sentence_transformers import SentenceTransformer
from dotenv import load_dotenv
load_dotenv()
# ─────────────────────────────────────────────────────────────────
# LLM abstraction — same code works with OpenAI, Groq, or Ollama
# ─────────────────────────────────────────────────────────────────
def get_llm():
"""
Returns (client, model_name) for the configured provider.
All three providers expose an OpenAI-compatible API, so the
rest of the code doesn't need to change per provider.
"""
provider = os.getenv("LLM_PROVIDER", "openai")
if provider == "groq":
client = OpenAI(
api_key=os.getenv("GROQ_API_KEY"),
base_url="https://api.groq.com/openai/v1"
)
return client, "llama-3.1-8b-instant"
if provider == "ollama":
# Ollama runs locally — no API key needed
client = OpenAI(api_key="ollama", base_url="http://localhost:11434/v1")
return client, os.getenv("OLLAMA_MODEL", "llama3.2")
# Default: OpenAI
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
return client, "gpt-4o-mini"
# ─────────────────────────────────────────────────────────────────
# Knowledge base — the documents we want to search over
# ─────────────────────────────────────────────────────────────────
# In a real system, these would come from PDFs, databases, or APIs.
# Here we use a small set of space exploration facts so you can run
# this without any external data source.
DOCUMENTS = [
"The Apollo 11 mission landed the first humans on the Moon on July 20, 1969. "
"Neil Armstrong and Buzz Aldrin walked on the surface while Michael Collins orbited above.",
"The James Webb Space Telescope (JWST) launched on December 25, 2021. "
"It observes in infrared light, letting it see through dust clouds and capture images of the earliest galaxies.",
"Mars has two small moons called Phobos and Deimos. "
"Scientists believe they are captured asteroids, not moons that formed alongside the planet.",
"Voyager 1 is the most distant human-made object ever launched. "
"It entered interstellar space in 2012 and continues to transmit data back to Earth from over 23 billion km away.",
"SpaceX's Falcon 9 is a partially reusable rocket. "
"The first stage booster autonomously lands back on Earth or a drone ship and is refurbished for future flights.",
"The International Space Station (ISS) orbits Earth at ~400 km altitude and travels at roughly 28,000 km/h. "
"It has been continuously inhabited since November 2000.",
"Saturn's rings are made mostly of water ice and rocky debris, ranging from microscopic grains to chunks the size of a house. "
"Despite spanning hundreds of thousands of kilometers, the rings are only about 10 meters thick in some places.",
"The Hubble Space Telescope has been operating since 1990. "
"Its deep-field images revealed thousands of galaxies in a patch of sky that appeared completely empty to the naked eye.",
]
# ─────────────────────────────────────────────────────────────────
# Step 1 — Embed and index all documents
# ─────────────────────────────────────────────────────────────────
# sentence-transformers runs locally — no embedding API key required.
# all-MiniLM-L6-v2 is small (~80MB) but good enough for most tasks.
embedding_model = SentenceTransformer("all-MiniLM-L6-v2")
# ChromaDB in-memory client — no disk, no server, no setup.
# For production, swap this for chromadb.PersistentClient() or a hosted DB.
chroma_client = chromadb.Client()
collection = chroma_client.create_collection("space_facts")
print("📥 Indexing documents...")
doc_embeddings = embedding_model.encode(DOCUMENTS).tolist()
collection.add(
documents=DOCUMENTS,
embeddings=doc_embeddings,
ids=[f"doc_{i}" for i in range(len(DOCUMENTS))]
)
print(f"✅ Indexed {len(DOCUMENTS)} documents.\n")
# ─────────────────────────────────────────────────────────────────
# Step 2 — Retrieve relevant chunks for a query
# ─────────────────────────────────────────────────────────────────
def retrieve(query: str, top_k: int = 3) -> list[str]:
"""
Embeds the query and finds the top-k most semantically similar chunks.
This is pure cosine similarity in vector space — there's no keyword
matching, no filters, no reranking. Simple, but fragile at scale.
"""
query_embedding = embedding_model.encode([query]).tolist()
results = collection.query(
query_embeddings=query_embedding,
n_results=top_k
)
# results["documents"] is a list-of-lists (one per query),
# so we take index [0] for our single query
return results["documents"][0]
# ─────────────────────────────────────────────────────────────────
# Step 3 — Generate an answer using the retrieved context
# ─────────────────────────────────────────────────────────────────
def generate(query: str, context_chunks: list[str]) -> str:
"""
Passes the query + retrieved chunks to the LLM.
The system prompt instructs the LLM to answer ONLY from the provided
context. Without this constraint the model might blend retrieved facts
with its own training data, which defeats the purpose of RAG.
"""
client, model = get_llm()
# Join chunks into a numbered list for readability
context = "\n".join(f"{i+1}. {chunk}" for i, chunk in enumerate(context_chunks))
messages = [
{
"role": "system",
"content": (
"You are a factual assistant. Answer the user's question using ONLY "
"the context provided. If the answer is not in the context, say "
"'I don't have that information.'"
)
},
{
"role": "user",
"content": f"Context:\n{context}\n\nQuestion: {query}"
}
]
response = client.chat.completions.create(
model=model,
messages=messages,
temperature=0 # deterministic — we want factual answers, not creative ones
)
return response.choices[0].message.content
# ─────────────────────────────────────────────────────────────────
# Run the full RAG pipeline on a few example queries
# ─────────────────────────────────────────────────────────────────
queries = [
"When did the first humans land on the Moon?",
"How fast does the ISS travel?",
"What are Saturn's rings made of?",
"Who invented the telephone?", # not in our knowledge base — watch what happens
]
for query in queries:
print(f"❓ {query}")
chunks = retrieve(query)
print(f" Retrieved chunks:")
for chunk in chunks:
# Print just the first 90 chars so the output stays readable
print(f" • {chunk[:90]}...")
answer = generate(query, chunks)
print(f" 💬 {answer}\n")
2. RAG hybride
La récupération en environnement de production mélange généralement la recherche vectorielle, les mots-clés BM25, les filtres de métadonnées et le réclassement par cross-encoder.
Le problème du RAG naïf
Les vecteurs purs gèrent bien les questions axées sur le sens, mais ont du mal avec les correspondances exactes, les ensembles filtrés tels que « uniquement des documents juridiques de 2023 », et les voisins bruyants. Les mots-clés, les métadonnées et les outils de réclassement comblent ces lacunes.
Le schéma
Query
│
├─→ Metadata filter ──→ narrow candidate pool
│
├─→ Vector search ──┐
│ ├─→ RRF fusion ──→ Cross-encoder rerank ──→ Top-k ──→ LLM
└─→ BM25 search ──┘
Étape 1 — Filtrage des métadonnées : les candidats sont réduits grâce à des champs structurés avant l’incorporation.
Étape 2 — Recherche hybride : la similarité vectorielle et BM25 s’exécutent en parallèle afin que tant la sémantique que les termes exacts soient pris en compte dans l’évaluation.
Étape 3 — Fusion RRF fusionne les listes classées à l’aide de la fusion de rangs réciproques (score = 1 / (k + rank)) sans normaliser les scores incompatibles.
Étape 4 — Réclassement par encodeur croisé évalue conjointement les paires (query, document) afin d’obtenir un ordre final précis — plus lent, mais bien plus fiable.
Quand l’utiliser
- Besoins en production tant pour la retenue que pour la précision
- Corpus dotés de métadonnées utiles
- Recherches mixtes sémantiques et basées sur des mots-clés
- Cas où une approche RAG naïve manque d’informations ou pollue la fenêtre de contexte
"""
Hybrid RAG — vector search + BM25 keyword search + metadata filtering + reranking.
The Problem with Naive RAG:
Pure vector similarity misses exact keyword matches (names, codes, product IDs)
and has no way to filter by structured metadata (date, source, category).
The Fix — three upgrades layered on top of each other:
1. Metadata filtering → narrow the search space before any ranking
2. Hybrid search → run vector search AND BM25 in parallel, fuse scores
3. Reranking → use a cross-encoder to re-score the fused candidates
This combination is what most production RAG systems use.
"""
import os
import math
from openai import OpenAI
from rank_bm25 import BM25Okapi
import chromadb
from sentence_transformers import SentenceTransformer, CrossEncoder
from dotenv import load_dotenv
load_dotenv()
# ─────────────────────────────────────────────────────────────────
# LLM abstraction — same pattern as 01-naive-rag
# ─────────────────────────────────────────────────────────────────
def get_llm():
provider = os.getenv("LLM_PROVIDER", "openai")
if provider == "groq":
return OpenAI(
api_key=os.getenv("GROQ_API_KEY"),
base_url="https://api.groq.com/openai/v1"
), "llama-3.1-8b-instant"
if provider == "ollama":
return OpenAI(api_key="ollama", base_url="http://localhost:11434/v1"), \
os.getenv("OLLAMA_MODEL", "llama3.2")
return OpenAI(api_key=os.getenv("OPENAI_API_KEY")), "gpt-4o-mini"
# ─────────────────────────────────────────────────────────────────
# Knowledge base — documents WITH metadata
# ─────────────────────────────────────────────────────────────────
# Each document now carries structured metadata alongside the text.
# This is the foundation for metadata filtering — you can pre-filter
# before running any expensive embedding or BM25 computation.
DOCUMENTS = [
{
"id": "doc_0",
"text": "The Apollo 11 mission landed the first humans on the Moon on July 20, 1969. "
"Neil Armstrong and Buzz Aldrin walked on the lunar surface.",
"metadata": {"category": "missions", "year": 1969, "source": "nasa.gov"}
},
{
"id": "doc_1",
"text": "The James Webb Space Telescope (JWST) launched on December 25, 2021. "
"It uses infrared imaging to observe early universe galaxies and exoplanet atmospheres.",
"metadata": {"category": "telescopes", "year": 2021, "source": "nasa.gov"}
},
{
"id": "doc_2",
"text": "SpaceX's Falcon 9 is a partially reusable rocket. The first stage booster lands "
"autonomously after launch and is refurbished for reuse.",
"metadata": {"category": "rockets", "year": 2015, "source": "spacex.com"}
},
{
"id": "doc_3",
"text": "NASA's Artemis program aims to return humans to the Moon by 2026 and establish "
"a sustainable lunar presence as a stepping stone to Mars.",
"metadata": {"category": "missions", "year": 2022, "source": "nasa.gov"}
},
{
"id": "doc_4",
"text": "The Hubble Space Telescope has been operating since 1990, producing iconic images "
"of nebulae and distant galaxies in visible and ultraviolet light.",
"metadata": {"category": "telescopes", "year": 1990, "source": "nasa.gov"}
},
{
"id": "doc_5",
"text": "SpaceX's Starship is designed for full reusability — both the Super Heavy booster "
"and the Starship upper stage return and land after flight.",
"metadata": {"category": "rockets", "year": 2023, "source": "spacex.com"}
},
{
"id": "doc_6",
"text": "The Mars Perseverance Rover landed in Jezero Crater in February 2021. "
"It is searching for signs of ancient microbial life and collecting rock samples.",
"metadata": {"category": "missions", "year": 2021, "source": "nasa.gov"}
},
{
"id": "doc_7",
"text": "The Nancy Grace Roman Space Telescope is NASA's next flagship telescope, "
"designed to survey wide fields of the sky for dark energy and exoplanets.",
"metadata": {"category": "telescopes", "year": 2026, "source": "nasa.gov"}
},
]
# ─────────────────────────────────────────────────────────────────
# Build the two search indexes
# ─────────────────────────────────────────────────────────────────
# --- Embedding model (vector search) ---
embedding_model = SentenceTransformer("all-MiniLM-L6-v2")
# --- Cross-encoder (reranking) ---
# A cross-encoder takes (query, document) pairs and outputs a relevance score.
# It's slower than bi-encoders but far more accurate — used as a final step.
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
# --- ChromaDB (vector store) ---
chroma_client = chromadb.Client()
collection = chroma_client.create_collection("space_hybrid")
texts = [d["text"] for d in DOCUMENTS]
doc_embeddings = embedding_model.encode(texts).tolist()
collection.add(
documents=texts,
embeddings=doc_embeddings,
ids=[d["id"] for d in DOCUMENTS],
metadatas=[d["metadata"] for d in DOCUMENTS]
)
# --- BM25 (keyword search) ---
# BM25 (Best Match 25) is a classic probabilistic ranking function.
# It scores documents based on term frequency and inverse document frequency.
# It excels at exact keyword matches that embeddings often miss.
tokenized_corpus = [doc.lower().split() for doc in texts]
bm25_index = BM25Okapi(tokenized_corpus)
print(f"✅ Indexed {len(DOCUMENTS)} documents (vector + BM25).\n")
# ─────────────────────────────────────────────────────────────────
# Retrieval — 3-stage pipeline
# ─────────────────────────────────────────────────────────────────
def filter_by_metadata(category: str | None = None, min_year: int | None = None) -> list[int]:
"""
Stage 1: Metadata filtering.
Returns the indices of documents that match the given filters.
Filtering BEFORE vector/keyword search narrows the corpus, making
retrieval faster and more precise — especially important at scale.
"""
indices = []
for i, doc in enumerate(DOCUMENTS):
meta = doc["metadata"]
if category and meta["category"] != category:
continue
if min_year and meta["year"] < min_year:
continue
indices.append(i)
return indices
def reciprocal_rank_fusion(
vector_ids: list[str],
bm25_indices: list[int],
k: int = 60
) -> list[str]:
"""
Stage 2b: Score fusion using Reciprocal Rank Fusion (RRF).
RRF combines rankings from multiple retrieval systems without needing
to normalize their raw scores. The formula is: score = 1 / (k + rank).
Results ranked high in EITHER system get boosted — catching what the
other system missed.
"""
scores: dict[str, float] = {}
# Score from vector ranking
for rank, doc_id in enumerate(vector_ids):
scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
# Score from BM25 ranking
for rank, doc_idx in enumerate(bm25_indices):
doc_id = DOCUMENTS[doc_idx]["id"]
scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
# Sort by combined score, highest first
return sorted(scores, key=scores.get, reverse=True)
def hybrid_retrieve(
query: str,
category: str | None = None,
min_year: int | None = None,
top_k: int = 3,
candidate_k: int = 6
) -> list[dict]:
"""
Full hybrid retrieval:
1. Filter by metadata → narrow the candidate pool
2. Vector search → semantic similarity
3. BM25 search → keyword matching
4. RRF fusion → combine the two rankings
5. Cross-encoder rerank → final high-precision scoring
candidate_k: how many candidates to gather before reranking.
More candidates = more recall, but slower reranking.
"""
# ── Stage 1: Metadata filter ──────────────────────────────────
allowed_indices = filter_by_metadata(category=category, min_year=min_year)
allowed_ids = {DOCUMENTS[i]["id"] for i in allowed_indices}
allowed_texts = [DOCUMENTS[i]["text"] for i in allowed_indices]
if not allowed_ids:
print(" ⚠️ No documents match the metadata filters.")
return []
# ── Stage 2a: Vector search (within filtered set) ─────────────
query_embedding = embedding_model.encode([query]).tolist()
vector_results = collection.query(
query_embeddings=query_embedding,
n_results=min(candidate_k, len(allowed_ids)),
where={"category": category} if category else None # ChromaDB metadata filter
)
vector_ids = vector_results["documents"][0] and vector_results["ids"][0] or []
# Keep only IDs that passed our metadata filter
vector_ids = [id_ for id_ in vector_ids if id_ in allowed_ids]
# ── Stage 2b: BM25 keyword search (within filtered set) ───────
tokenized_query = query.lower().split()
# Score all filtered documents with BM25
bm25_filtered = BM25Okapi([DOCUMENTS[i]["text"].lower().split() for i in allowed_indices])
bm25_scores = bm25_filtered.get_scores(tokenized_query)
# Get indices sorted by score (within the filtered subset)
top_bm25_local = sorted(range(len(allowed_indices)), key=lambda i: bm25_scores[i], reverse=True)[:candidate_k]
# Map back to original DOCUMENTS indices
top_bm25_global = [allowed_indices[i] for i in top_bm25_local]
# ── Stage 3: RRF fusion ────────────────────────────────────────
fused_ids = reciprocal_rank_fusion(vector_ids, top_bm25_global)[:candidate_k]
# ── Stage 4: Cross-encoder reranking ──────────────────────────
# Fetch the actual text for each candidate
candidates = [
next(d for d in DOCUMENTS if d["id"] == doc_id)
for doc_id in fused_ids
if any(d["id"] == doc_id for d in DOCUMENTS)
]
# Cross-encoder scores each (query, document) pair independently
# This is more accurate than embedding similarity but O(n) slower
pairs = [(query, c["text"]) for c in candidates]
rerank_scores = reranker.predict(pairs)
# Sort candidates by reranker score, take top_k
ranked = sorted(zip(candidates, rerank_scores), key=lambda x: x[1], reverse=True)
return [doc for doc, _ in ranked[:top_k]]
# ─────────────────────────────────────────────────────────────────
# Generate
# ─────────────────────────────────────────────────────────────────
def generate(query: str, context_docs: list[dict]) -> str:
client, model = get_llm()
context = "\n".join(
f"{i+1}. [{doc['metadata']['category']} | {doc['metadata']['year']}] {doc['text']}"
for i, doc in enumerate(context_docs)
)
messages = [
{
"role": "system",
"content": "You are a factual assistant. Answer using ONLY the provided context."
},
{
"role": "user",
"content": f"Context:\n{context}\n\nQuestion: {query}"
}
]
response = client.chat.completions.create(
model=model, messages=messages, temperature=0
)
return response.choices[0].message.content
# ─────────────────────────────────────────────────────────────────
# Run example queries
# ─────────────────────────────────────────────────────────────────
examples = [
{
"query": "Tell me about reusable rockets",
"filters": {"category": "rockets"}, # only look at rocket documents
"description": "Metadata filter: rockets only"
},
{
"query": "What telescope launched most recently?",
"filters": {"category": "telescopes", "min_year": 2020}, # telescopes after 2020
"description": "Metadata filter: telescopes after 2020"
},
{
"query": "NASA Mars mission 2021",
"filters": {"category": "missions"},
"description": "BM25 shines: exact keywords (NASA, Mars, 2021)"
},
]
for ex in examples:
print(f"❓ {ex['query']}")
print(f" Filter: {ex['description']}")
docs = hybrid_retrieve(
query=ex["query"],
category=ex["filters"].get("category"),
min_year=ex["filters"].get("min_year"),
)
print(f" Retrieved {len(docs)} docs after reranking:")
for doc in docs:
print(f" • [{doc['metadata']['category']} | {doc['metadata']['year']}] {doc['text'][:80]}...")
answer = generate(ex["query"], docs)
print(f" 💬 {answer}\n")
3. Graph RAG
Ici, un graphe de connaissances constitue le socle de la récupération (seul ou en complément des vecteurs).
Le problème résolu par GraphRAG
La similarité se demande « quel texte ressemble à ma requête ? » Les questions de relation — « Qui a investi dans OpenAI ? », « Qu’a acquis Microsoft ? », « Comment X et Y sont-ils liés ? » — constituent des parcours dans un graphe. Les embeddings aplatisent la structure ; les graphes conservent explicitement les relations de type.
Le schéma
Documents / structured data
│
└─→ Entity + Relationship extraction ──→ Knowledge Graph (nodes + edges)
Query
│
├─→ Entity extraction ──→ Find matching nodes
│
└─→ Graph traversal (N hops) ──→ Subgraph context ──→ LLM ──→ Answer
"""
GraphRAG — retrieval using a knowledge graph instead of (or alongside) a vector DB.
Why a graph?
Documents store text. Graphs store *relationships* — who owns what, what caused what,
which thing is part of which system. When your questions are about connections
("Who invested in OpenAI?", "What products did Microsoft acquire?"), traversing a
graph gives you exactly the right context without keyword or semantic guesswork.
Pattern:
1. Build a knowledge graph from entities and their relationships
2. Extract entities from the query (simple keyword matching here; NER in production)
3. Find matching nodes in the graph
4. Traverse 1-2 hops to collect related context
5. Format the subgraph as text and pass to the LLM
"""
import os
import networkx as nx
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
# ─────────────────────────────────────────────────────────────────
# LLM abstraction
# ─────────────────────────────────────────────────────────────────
def get_llm():
provider = os.getenv("LLM_PROVIDER", "openai")
if provider == "groq":
return OpenAI(
api_key=os.getenv("GROQ_API_KEY"),
base_url="https://api.groq.com/openai/v1"
), "llama-3.1-8b-instant"
if provider == "ollama":
return OpenAI(api_key="ollama", base_url="http://localhost:11434/v1"), \
os.getenv("OLLAMA_MODEL", "llama3.2")
return OpenAI(api_key=os.getenv("OPENAI_API_KEY")), "gpt-4o-mini"
# ─────────────────────────────────────────────────────────────────
# Knowledge Graph — entities and relationships in the AI industry
# ─────────────────────────────────────────────────────────────────
# We use networkx's directed graph: nodes are entities, edges are relationships.
# In a production GraphRAG system, this graph would be extracted automatically
# from documents using an LLM or NER pipeline. Here we build it manually.
G = nx.DiGraph()
# ── Nodes — each entity has a type and a short description ───────
entities = [
("OpenAI", {"type": "company", "desc": "AI research company, creator of GPT-4, DALL-E, and Sora"}),
("Microsoft", {"type": "company", "desc": "Technology giant, owner of Azure, Office, and GitHub"}),
("Google", {"type": "company", "desc": "Technology company, owner of Search, YouTube, and DeepMind"}),
("DeepMind", {"type": "company", "desc": "AI research lab, creator of AlphaFold and Gemini"}),
("Anthropic", {"type": "company", "desc": "AI safety company, creator of the Claude model family"}),
("Sam Altman", {"type": "person", "desc": "CEO of OpenAI, former president of Y Combinator"}),
("Ilya Sutskever", {"type": "person", "desc": "Co-founder of OpenAI and SSI, key architect of GPT models"}),
("Demis Hassabis", {"type": "person", "desc": "CEO and co-founder of DeepMind"}),
("Dario Amodei", {"type": "person", "desc": "CEO and co-founder of Anthropic, former VP Research at OpenAI"}),
("GPT-4", {"type": "product", "desc": "Large language model by OpenAI, released in 2023"}),
("Claude", {"type": "product", "desc": "LLM family by Anthropic, focused on safety and helpfulness"}),
("Gemini", {"type": "product", "desc": "Multimodal LLM by Google DeepMind, released in 2023"}),
("AlphaFold", {"type": "product", "desc": "AI system by DeepMind that predicts protein structures"}),
("Azure", {"type": "product", "desc": "Microsoft's cloud platform, hosts OpenAI models via Azure OpenAI"}),
("GitHub", {"type": "product", "desc": "Code hosting platform owned by Microsoft"}),
("Y Combinator", {"type": "org", "desc": "Startup accelerator that funded companies like OpenAI, Dropbox, Airbnb"}),
]
G.add_nodes_from(entities)
# ── Edges — directional relationships between entities ────────────
# Format: (source, target, {"relation": "...", "detail": "..."})
relationships = [
("Sam Altman", "OpenAI", {"relation": "CEO_of", "detail": "CEO since 2019 (with a brief ouster in 2023)"}),
("Ilya Sutskever", "OpenAI", {"relation": "co_founded", "detail": "Co-founded OpenAI in 2015 alongside Sam Altman"}),
("Dario Amodei", "Anthropic", {"relation": "co_founded", "detail": "Founded Anthropic in 2021 after leaving OpenAI"}),
("Dario Amodei", "OpenAI", {"relation": "former_VP_at", "detail": "Was VP of Research at OpenAI before leaving"}),
("Demis Hassabis", "DeepMind", {"relation": "co_founded", "detail": "Co-founded DeepMind in 2010, acquired by Google in 2014"}),
("Microsoft", "OpenAI", {"relation": "invested_in", "detail": "Invested ~$13B in OpenAI across multiple rounds"}),
("Google", "Anthropic", {"relation": "invested_in", "detail": "Invested ~$300M in Anthropic in 2023"}),
("Google", "DeepMind", {"relation": "acquired", "detail": "Acquired DeepMind in 2014 for ~$500M"}),
("Microsoft", "GitHub", {"relation": "acquired", "detail": "Acquired GitHub in 2018 for $7.5B"}),
("OpenAI", "GPT-4", {"relation": "created", "detail": "Released GPT-4 in March 2023"}),
("Anthropic", "Claude", {"relation": "created", "detail": "Claude 3 family released in 2024"}),
("DeepMind", "Gemini", {"relation": "created", "detail": "Gemini 1.0 released December 2023 as GPT-4 competitor"}),
("DeepMind", "AlphaFold", {"relation": "created", "detail": "AlphaFold 2 solved protein structure prediction in 2020"}),
("Microsoft", "Azure", {"relation": "owns", "detail": "Azure hosts OpenAI's models via Azure OpenAI Service"}),
("OpenAI", "Azure", {"relation": "partners_with","detail": "OpenAI's API and models are available through Azure"}),
("Sam Altman", "Y Combinator", {"relation": "led", "detail": "Was president of Y Combinator from 2014 to 2019"}),
]
G.add_edges_from(relationships)
print(f"✅ Knowledge graph built: {G.number_of_nodes()} nodes, {G.number_of_edges()} edges.\n")
# ─────────────────────────────────────────────────────────────────
# Step 1 — Entity extraction from the query
# ─────────────────────────────────────────────────────────────────
def extract_entities(query: str) -> list[str]:
"""
Find which graph nodes are mentioned in the query.
This is simple case-insensitive substring matching — good enough for demos.
In production you'd use an LLM or a NER model to extract entities, which
handles synonyms, abbreviations, and entities not spelled out exactly.
"""
query_lower = query.lower()
return [node for node in G.nodes if node.lower() in query_lower]
# ─────────────────────────────────────────────────────────────────
# Step 2 — Graph traversal to collect context
# ─────────────────────────────────────────────────────────────────
def retrieve_from_graph(query: str, max_hops: int = 2) -> str:
"""
Retrieves context by traversing the graph from query entities.
For each entity found in the query:
- Include the entity's own description
- Walk outgoing edges (what this entity does / relates to)
- Walk incoming edges (what points to this entity)
- Optionally go 1 more hop for richer context
max_hops=1 gives direct neighbors; max_hops=2 also includes neighbors' neighbors.
More hops = richer context but also more noise.
"""
seed_entities = extract_entities(query)
if not seed_entities:
return "No matching entities found in the knowledge graph."
print(f" Entities found: {seed_entities}")
context_lines = []
visited = set()
def collect_node(node: str):
"""Collect a node's description and all its edges as text."""
if node in visited:
return
visited.add(node)
data = G.nodes[node]
context_lines.append(f"[{data['type'].upper()}] {node}: {data['desc']}")
# Outgoing edges: what this entity does/relates to
for _, target, edge_data in G.out_edges(node, data=True):
context_lines.append(
f" → {node} --[{edge_data['relation']}]--> {target}: {edge_data['detail']}"
)
# Incoming edges: what points to this entity
for source, _, edge_data in G.in_edges(node, data=True):
context_lines.append(
f" ← {source} --[{edge_data['relation']}]--> {node}: {edge_data['detail']}"
)
# Collect seed entities and their neighbors up to max_hops
for entity in seed_entities:
collect_node(entity)
if max_hops >= 2:
# Also collect 1-hop neighbors for richer context
for neighbor in list(G.successors(entity)) + list(G.predecessors(entity)):
collect_node(neighbor)
return "\n".join(context_lines)
# ─────────────────────────────────────────────────────────────────
# Step 3 — Generate answer from graph context
# ─────────────────────────────────────────────────────────────────
def generate(query: str, graph_context: str) -> str:
client, model = get_llm()
messages = [
{
"role": "system",
"content": (
"You are a helpful assistant with access to a knowledge graph about the AI industry. "
"Answer the question using ONLY the graph context provided. "
"Be concise and factual."
)
},
{
"role": "user",
"content": f"Knowledge Graph Context:\n{graph_context}\n\nQuestion: {query}"
}
]
response = client.chat.completions.create(
model=model, messages=messages, temperature=0
)
return response.choices[0].message.content
# ─────────────────────────────────────────────────────────────────
# Run example queries
# ─────────────────────────────────────────────────────────────────
queries = [
"Who co-founded OpenAI?",
"What has Microsoft invested in or acquired?",
"What is the relationship between Google and DeepMind?",
"What did Dario Amodei do before Anthropic?",
]
for query in queries:
print(f"❓ {query}")
context = retrieve_from_graph(query)
print(f" Graph context snippet:\n {context[:200]}...")
answer = generate(query, context)
print(f" 💬 {answer}\n")
Structure du graphe
- Nœuds — entités (entreprises, personnes, produits) avec des descriptions
- Arêtes — relations orientées de type spécifique (
invested_in,acquired,CEO_of, …)
Récupération par parcours
- Extraire les entités mentionnées dans la requête
- Partir des attributs de ces nœuds et de leurs arêtes directes
- Étendre éventuellement le parcours à plusieurs étapes pour obtenir un contexte de voisinage plus riche
Quand l’utiliser
- Lorsqu’il existe des entités claires et des relations nommées
4. Agentic RAG
Le modèle planifie la récupération des informations : quels outils utiliser, dans quel ordre, et lorsque suffisamment de preuves sont disponibles.
Le problème des pipelines fixes
Les pipelines naïfs, hybrides et graphiques définissent de manière fixe un parcours pour chaque question. Les demandes ambiguës ou nécessitant plusieurs étapes exigent une structure en branches.
Le modèle — Boucle ReAct
ReAct = Raisonnement + Action : l’agent réfléchit, invoque un outil de récupération ou de calcul, observe les résultats, et répète le processus jusqu’à pouvoir répondre.
"""
Agentic RAG — the LLM plans its own retrieval strategy.
The Problem with Fixed Pipelines:
In Naive/Hybrid/GraphRAG, the retrieval strategy is hardcoded. Every query
goes through the same pipeline regardless of what the question actually needs.
But questions vary:
"What is SpaceX?" → single lookup, done
"Compare SpaceX and NASA's approach to Mars" → needs two retrievals + synthesis
"What is the orbital speed of the ISS in miles per hour?" → needs retrieval + math
A static pipeline can't adapt. An agent can.
The Agentic Approach:
We implement a ReAct loop (Reason + Act) where the LLM:
1. Reads the question
2. Decides which tool to use and with what input
3. Sees the tool's output
4. Decides whether it has enough to answer, or needs another tool call
5. Repeats until it can give a final answer
Tools available to the agent:
- search_docs(query) → vector search over a document corpus
- get_entity(name) → exact lookup of a known entity (like a mini-graph)
- calculate(expression) → evaluate a math expression safely
"""
import os
import json
import ast
import operator
import chromadb
from openai import OpenAI
from sentence_transformers import SentenceTransformer
from dotenv import load_dotenv
load_dotenv()
# ─────────────────────────────────────────────────────────────────
# LLM abstraction
# ─────────────────────────────────────────────────────────────────
def get_llm():
provider = os.getenv("LLM_PROVIDER", "openai")
if provider == "groq":
return OpenAI(
api_key=os.getenv("GROQ_API_KEY"),
base_url="https://api.groq.com/openai/v1"
), "llama-3.1-8b-instant"
if provider == "ollama":
return OpenAI(api_key="ollama", base_url="http://localhost:11434/v1"), \
os.getenv("OLLAMA_MODEL", "llama3.2")
return OpenAI(api_key=os.getenv("OPENAI_API_KEY")), "gpt-4o-mini"
# ─────────────────────────────────────────────────────────────────
# Knowledge base — indexed for search_docs tool
# ─────────────────────────────────────────────────────────────────
DOCUMENTS = [
"SpaceX was founded in 2002 by Elon Musk with the goal of making space travel cheaper "
"and eventually colonizing Mars. Its Falcon 9 is the world's first orbital-class reusable rocket.",
"NASA was founded in 1958 and is a US government agency. Its Mars missions include "
"Curiosity (2012) and Perseverance (2021). NASA's SLS rocket is expendable, not reusable.",
"The International Space Station (ISS) orbits Earth at 408 km altitude and travels "
"at 7.66 km/s (27,576 km/h or 17,132 mph). It completes one orbit every 92 minutes.",
"SpaceX's Starship is a fully reusable spacecraft designed for Mars missions, lunar landings, "
"and point-to-point Earth travel. Its first successful orbital flight was in 2024.",
"The Artemis program is NASA's plan to return humans to the Moon. Artemis 1 (2022) was "
"uncrewed. Artemis 2 (2024) will be the first crewed test. Moon landing planned for Artemis 3.",
"Mars is approximately 225 million km from Earth on average. A one-way trip with current "
"propulsion takes 7-9 months. SpaceX aims to cut this with Starship's higher thrust-to-weight ratio.",
"Blue Origin is a space company founded by Jeff Bezos in 2000. Its New Shepard rocket does "
"suborbital tourism flights. Its New Glenn rocket is designed for orbital launches.",
"Orbital velocity is the minimum speed needed to stay in orbit. At 400 km altitude, "
"this is approximately 7.66 km/s. Below this, the spacecraft would fall back to Earth.",
]
# Build in-memory vector index for the search_docs tool
embedding_model = SentenceTransformer("all-MiniLM-L6-v2")
chroma_client = chromadb.Client()
collection = chroma_client.create_collection("agent_kb")
collection.add(
documents=DOCUMENTS,
embeddings=embedding_model.encode(DOCUMENTS).tolist(),
ids=[f"doc_{i}" for i in range(len(DOCUMENTS))]
)
# Entity lookup table for the get_entity tool
ENTITIES = {
"spacex": "SpaceX: Private space company by Elon Musk. Falcon 9 (reusable), Starship (Mars). Founded 2002.",
"nasa": "NASA: US government space agency. Artemis (Moon), Perseverance (Mars). Founded 1958.",
"iss": "ISS: Orbits at 408km, speed 7.66 km/s (17,132 mph), 92-min orbit. Inhabited since 2000.",
"blue origin": "Blue Origin: Jeff Bezos' space company. New Shepard (suborbital), New Glenn (orbital). Founded 2000.",
"starship": "Starship: SpaceX's fully reusable Mars rocket. First orbital flight 2024. Largest rocket ever built.",
}
print("✅ Agent knowledge base ready.\n")
# ─────────────────────────────────────────────────────────────────
# Tool definitions
# ─────────────────────────────────────────────────────────────────
# Tool specs in OpenAI function-calling format.
# The agent (LLM) reads these to know what tools exist and how to call them.
TOOLS = [
{
"type": "function",
"function": {
"name": "search_docs",
"description": "Search the knowledge base using semantic similarity. Use for general questions about space, rockets, missions, or orbit.",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "The search query"}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "get_entity",
"description": "Look up a specific named entity (SpaceX, NASA, ISS, Blue Origin, Starship). Use when you need precise facts about a known entity.",
"parameters": {
"type": "object",
"properties": {
"name": {"type": "string", "description": "Entity name to look up"}
},
"required": ["name"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "Evaluate a mathematical expression. Use for unit conversions or arithmetic. Example: '7.66 * 3600' to convert km/s to km/h.",
"parameters": {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "A safe arithmetic expression using +, -, *, /, (, ), and numbers"}
},
"required": ["expression"]
}
}
}
]
# ─────────────────────────────────────────────────────────────────
# Tool implementations
# ─────────────────────────────────────────────────────────────────
def search_docs(query: str) -> str:
"""Semantic search over the knowledge base."""
query_embedding = embedding_model.encode([query]).tolist()
results = collection.query(query_embeddings=query_embedding, n_results=2)
chunks = results["documents"][0]
return "\n---\n".join(chunks)
def get_entity(name: str) -> str:
"""Exact entity lookup."""
result = ENTITIES.get(name.lower().strip())
return result if result else f"No entity found for '{name}'. Try: {list(ENTITIES.keys())}"
def calculate(expression: str) -> str:
"""
Safe arithmetic evaluator — no eval(), no exec().
Only allows numbers and basic operators to prevent code injection.
"""
# Only allow digits, spaces, and basic math operators
allowed_chars = set("0123456789 +-*/().")
if not all(c in allowed_chars for c in expression):
return "Error: Only basic arithmetic is allowed (+, -, *, /, parentheses, numbers)."
try:
# Parse to AST and evaluate node-by-node — never calls eval()
tree = ast.parse(expression, mode="eval")
ops = {
ast.Add: operator.add, ast.Sub: operator.sub,
ast.Mult: operator.mul, ast.Div: operator.truediv,
ast.USub: operator.neg
}
def eval_node(node):
if isinstance(node, ast.Constant):
return node.value
if isinstance(node, ast.BinOp):
return ops[type(node.op)](eval_node(node.left), eval_node(node.right))
if isinstance(node, ast.UnaryOp):
return ops[type(node.op)](eval_node(node.operand))
raise ValueError(f"Unsupported operation: {type(node)}")
result = eval_node(tree.body)
return f"{expression} = {result:.4f}"
except Exception as e:
return f"Calculation error: {e}"
def dispatch_tool(name: str, args: dict) -> str:
"""Route a tool call from the agent to the right implementation."""
if name == "search_docs":
return search_docs(args["query"])
if name == "get_entity":
return get_entity(args["name"])
if name == "calculate":
return calculate(args["expression"])
return f"Unknown tool: {name}"
# ─────────────────────────────────────────────────────────────────
# The ReAct Agent Loop
# ─────────────────────────────────────────────────────────────────
def run_agent(question: str, max_iterations: int = 5) -> str:
"""
Runs the Reason + Act (ReAct) loop.
Each iteration:
1. Send the conversation history to the LLM
2. If the LLM calls a tool → execute it, append result, continue
3. If the LLM gives a text response → that's the final answer, stop
max_iterations prevents infinite loops if the agent gets stuck.
"""
client, model = get_llm()
# Conversation starts with a system prompt + the user's question
messages = [
{
"role": "system",
"content": (
"You are a research assistant with tools to look up space industry facts. "
"For each question, think about what information you need, use your tools to "
"retrieve it, and synthesize a clear, factual answer. "
"If a question requires multiple lookups or calculations, do them step by step."
)
},
{"role": "user", "content": question}
]
for iteration in range(max_iterations):
response = client.chat.completions.create(
model=model,
messages=messages,
tools=TOOLS,
tool_choice="auto" # let the LLM decide whether to call a tool
)
message = response.choices[0].message
finish_reason = response.choices[0].finish_reason
# ── Case 1: LLM wants to call a tool ─────────────────────
if finish_reason == "tool_calls" and message.tool_calls:
messages.append(message) # add assistant message with tool call
for tool_call in message.tool_calls:
tool_name = tool_call.function.name
tool_args = json.loads(tool_call.function.arguments)
print(f" 🔧 Tool call [{iteration+1}]: {tool_name}({tool_args})")
result = dispatch_tool(tool_name, tool_args)
print(f" Result: {result[:120]}...")
# Append the tool result so the LLM sees it in the next iteration
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
# ── Case 2: LLM has a final answer ────────────────────────
else:
return message.content
return "Agent reached maximum iterations without a final answer."
# ─────────────────────────────────────────────────────────────────
# Run example queries
# ─────────────────────────────────────────────────────────────────
questions = [
# Single-hop: one tool call should be enough
"What is SpaceX's approach to rocket reusability?",
# Multi-hop: needs two entity lookups then synthesis
"Compare SpaceX and NASA's plans for Mars exploration.",
# Tool-chaining: retrieve + calculate
"What is the ISS orbital speed in miles per hour?",
]
for question in questions:
print(f"❓ {question}")
answer = run_agent(question)
print(f" 💬 {answer}\n")
Quand l’utiliser
- Réponses nécessitant plusieurs étapes de récupération
- Consultations ambiguës dont le plan n’est pas connu à l’avance
- Mélange de récupération d’informations avec des calculs ou d’autres actions
- Conversations exploratoires ultérieures
Choisissez le modèle le plus léger qui correspond à la forme de la question : « naive » pour des démonstrations simples, « hybrid » pour une recherche de texte en production, les graphes pour un ancrage relationnel, et les agents lorsque le plan de récupération lui-même doit être dynamique.