Was ist RAG? Naive, hybride, grafische und agentbasierte Muster
Ein praktischer Überblick über Retrieval-augmented Generation: klassisches Vektor-RAG, hybrides Retrieving, GraphRAG sowie agentebasierte Ansätze – wann welche Methode geeignet ist und was in der Praxis schiefgehen kann.
RAG (Retrieval-Augmented Generation) gibt Antworten mit Hilfe einer externen Wissensquelle statt allein aus dem Modellgedächtnis. Zunächst werden relevante Abschnitte abgerufen und als Kontext übergeben, damit das LLM seine Antwort begründen kann. Stellen Sie sich eine Prüfung mit offenen Büchern vor: Das Modell ist der Schüler, RAG stellt die Seiten bereit. Die Antworten bleiben aktualisierbar, ohne neu trainiert werden zu müssen.
Wie RAG funktioniert
Drei Phasen:
- Indexieren – Dokumente in eine suchbare Datenbank (Vektor-Datenbank, Graph, umgekehrter Index usw.) aufteilen
- Absuchen – die besten Abschnitte für eine Anfrage finden
- Erstellen – diese Abschnitte an das LLM weitergeben, um eine Antwort zu synthetisieren
Die Qualität der Informationsabrufung ist entscheidend. Ein starkes Modell mit schwacher Informationsabruffunktion liefert weiterhin schwache Antworten; die eigentliche Herausforderung liegt darin, den passenden Kontext zu finden.
1. Naives RAG
Das einfachste Muster: Dokumente einbetten, Vektoren speichern und zur Abfragezeit nach Ähnlichkeit abrufen.
Das Muster
Dokumente in Blöcke aufteilen, sie einbetten und die Vektoren speichern. Zur Abfragezeit die Frage einbetten, die nächstgelegenen Nachbarn auswählen und das Modell dazu bringen, eine Antwort aus diesen Abschnitten zu schreiben.
Wann es verwendet werden sollte
- Prototypen und Demos
- Kleine, übersichtliche Korpora (etwa unter ~50.000 Blöcken)
- Bereiche, in denen semantische Ähnlichkeit die Absicht widerspiegelt (Prosa-Fragen & Antworten)
Wann es versagt
- Genaue Token (Namen, Codes, IDs), die von den Embeddings übersehen werden
- Keine Metadatenfilter (Datum, Quelle, Kategorie)
- Störende Top-k-Listen ohne Neubewertung
- Vollständige Scans, die bei Millionen von Blöcken teuer werden
Was im Code beachtet werden sollte
Indizierung in ChromaDB, ein einfacher retrieve()-Aufruf auf Basis der Kosinusähnlichkeit sowie ein Systemprompt, der das Modell auf den bereitgestellten Kontext beschränkt:
"""
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. Hybrid RAG
In der Produktivumgebung werden bei der Informationsabrufung in der Regel Vektorabfragen, BM25-Schlüsselwörter, Metadatenfilter sowie Cross-Encoder zur Neubewertung der Ergebnisse verwendet.
Das Problem mit Naiver RAG
Pure Vektoren können bedeutungsbasierte Fragen bewältigen, stoßen aber auf exakte Übereinstimmungen, gefilterte Datensätze wie „nur Rechtsdokumente aus 2023“ sowie störende ähnliche Einträge. Schlüsselwörter, Metadaten und Neubewertungsmethoden schließen diese Lücken.
Muster
Query
│
├─→ Metadata filter ──→ narrow candidate pool
│
├─→ Vector search ──┐
│ ├─→ RRF fusion ──→ Cross-encoder rerank ──→ Top-k ──→ LLM
└─→ BM25 search ──┘
Phase 1 – Metadatenfilterung reduziert die Kandidaten durch strukturierte Felder, bevor mit der Embedding-Verarbeitung begonnen wird.
Phase 2 – Hybridsuche führt Vektorähnlichkeitsprüfungen und BM25-Abfragen parallel durch, sodass sowohl semantische Aspekte als auch exakte Begriffe bewertet werden können.
Phase 3 – RRF-Fusion verschmilzt gerankte Listen mithilfe der Reciprocal Rank Fusion (score = 1 / (k + rank)) ohne Normalisierung inkompatibler Scores.
Phase 4 – Cross-Encoder-Reranking bewertet (query, document)-Paare gemeinsam, um eine präzise Endreihenfolge zu ermitteln – langsamer, aber deutlich genauer.
Wann ist es geeignet?
- Produktionsanforderungen an Recall und Präzision
- Korpora mit nützlichen Metadaten
- Mischungen aus semantischen und schlagwortbasierten Abfragen
- Fälle, in denen naive RAG-Methoden den Kontextfenster verfehlen oder verschmutzen
"""
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
Hier dient ein Wissensgraph als Rückgrat der Informationsabrufung (allein oder zusätzlich zu Vektoren).
Das Problem, das GraphRAG löst
Ähnlichkeit fragt „Welcher Text sieht wie meine Abfrage aus?“ Beziehungsfragen – „Wer hat in OpenAI investiert?“, „Was hat Microsoft erworben?“, „Wie sind X und Y miteinander verbunden?“ – sind Graphen-Überquerungen. Embeddings glätten die Struktur; Graphen halten typisierte Kanten explizit bereit.
Muster
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")
Graphenstruktur
- Knoten – Entitäten (Unternehmen, Personen, Produkte) mit Beschreibungen
- Kanten – typisierte, gerichtete Beziehungen (
invested_in,acquired,CEO_of, …)
Auswertung durch Überquerung
- Entitäten extrahieren, die in der Abfrage erwähnt werden
- Aus den Attributen dieser Knoten sowie direkten Kanten ausgehen
- Gegebenenfalls mehrstufig erweitern, um einen reichhaltigeren Kontext zu erhalten
Wann verwenden?
- Klare Entitäten und benannte Beziehungen
4. Aggressives RAG
Das Modell plant die Informationsbeschaffung: welche Tools, in welcher Reihenfolge und wann ausreichende Beweise vorliegen.
Das Problem mit festen Pipelines
Naive, hybride und grafbasierte Pipelines legen für jede Frage einen festen Ablauf fest. Mehrschrittige oder unklare Anfragen erfordern eine Verzweigung.
Das Muster – ReAct-Schleife
ReAct = Reasonen + Handeln: Der Agent denkt nach, ruft ein Tool zur Informationsbeschaffung oder Berechnung auf, beobachtet die Ergebnisse und wiederholt den Vorgang, bis er eine Antwort geben kann.
"""
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")
Wann sollte man es verwenden
- Antworten, die mehrere Schritte zur Informationsbeschaffung erfordern
- Ungenaue Anfragen, bei denen der Ablauf im Voraus unbekannt ist
- Kombination von Informationsbeschaffung mit Berechnungen oder anderen Aktionen
- Erforschende, fortlaufende Gespräche
Wählen Sie das leichteste Muster, das zur Form der Frage passt: „naive“ für saubere Demonstrationen, „hybrid“ für die Textsuche in der Produktion, „Graphs“ für relationale Verknüpfungen und „Agents“, wenn auch der Suchplan selbst dynamisch sein muss.