Что такое RAG? Наивные, гибридные, графовые и агентные паттерны
Практическая карта технологий генерации с усилением через поиск: классический векторный RAG, гибридные методы поиска, GraphRAG и агентные подходы — когда они подходят и что может сломаться в производственной среде.
RAG (Retrieval-Augmented Generation) использует внешние источники знаний для формирования ответов, а не только память модели. Сначала извлекаются релевантные фрагменты, которые затем передаются в качестве контекста, позволяя большой языковой модели обосновать свой ответ. Можно представить себе экзамен с открытыми учебниками: модель — это студент, а RAG предоставляет необходимые страницы. Ответы можно обновлять без переобучения модели.
Как работает RAG
Три этапа:
- Индексация — разбиение документов на части для поиска в хранилище (векторная БД, граф, обратный индекс и т. д.)
- Извлечение — поиск наиболее подходящих фрагментов для запроса
- Генерация — передача этих фрагментов большой языковой модели для формирования ответа
Качество извлечения данных имеет решающее значение. Даже мощная модель с слабым механизмом извлечения будет давать слабые ответы; ключ к эффективной работе заключается в способе нахождения контекста.
1. Примитивный RAG
Самый простой подход: встраивать документы, хранить векторы и находить соответствия по схожести в момент запроса.
Подход
Разбиваем документы на фрагменты, встраиваем их и храним полученные векторы. При запросе встраиваем вопрос, находим ближайших соседей и позволяем модели сформировать ответ на основе этих фрагментов.
Когда его использовать
- Прототипы и демо-версии
- Небольшие, структурированные корпуса (примерно менее 50 тыс. фрагментов)
- Области, где семантическое сходство соответствует намерению пользователя (вопросы и ответы в прозе)
Когда он теряет эффективность
- Точные токены (имена, коды, идентификаторы), которые не учитываются алгоритмами встраивания
- Отсутствие фильтров метаданных (дата, источник, категория)
- Использование метода top-k без повторной сортировки
- Полный поиск, который становится дорогостоящим при миллионах фрагментов
На что обратить внимание в коде
Индексация в ChromaDB, простая функция retrieve(), основанная на косинусной схожести, и системное указание, ограничивающее модель предоставленным контекстом:
"""
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
В реальных условиях поиск обычно сочетает векторный поиск, ключевые слова BM25, фильтры метаданных и переранжирование с помощью кросс-кодера.
Проблемы примитивного RAG
Чистые векторы хорошо справляются с вопросами, основанными на значении, но терпят неудачи при поиске точных совпадений, фильтрованных данных, таких как «только юридические документы 2023 года», и «шумных» соседних элементов. Ключевые слова, метаданные и инструменты переранжирования помогают преодолеть эти недостатки.
Паттерн
Query
│
├─→ Metadata filter ──→ narrow candidate pool
│
├─→ Vector search ──┐
│ ├─→ RRF fusion ──→ Cross-encoder rerank ──→ Top-k ──→ LLM
└─→ BM25 search ──┘
Этап 1 — Фильтрация метаданных сужает список кандидатов с помощью структурированных полей перед началом процесса включения данных в векторное представление.
Этап 2 — Гибридный поиск выполняет расчёт косинусной схожести и анализ по ключевым словам BM25 одновременно, чтобы как семантика, так и точные термины учитывались при оценке результатов.
Этап 3 — слияние RRF объединяет ранжированные списки с методом взаимного слияния рангов (score = 1 / (k + rank)) без нормализации несовместимых оценок.
Этап 4 — переранжирование с использованием кросс-кодера одновременно оценивает пары (запрос, документ) для получения точного окончательного порядка — это медленнее, но результат более точный.
Когда использовать
- Потребности производства в одновременном учете воспроизводимости и точности
- Корпуса с полезными метаданными
- Смешанные запросы на основе семантики и ключевых слов
- Ситуации, когда примитивный RAG упускает информацию или искажает контекстное окно
"""
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. Графовый RAG
Здесь основой поиска выступает граф знаний (один или вместе с векторами).
Проблема, которую решает GraphRAG
Подход схожести заключается в поиске «какой текст похож на мой запрос?» Вопросы, связанные с отношениями — «Кто инвестировал в OpenAI?», «Что приобрела Microsoft?», «Как связаны X и Y?» — представляют собой просмотр графа. Эмбеддинги упрощают структуру, тогда как графы сохраняют типизированные рёбра в явном виде.
Шаблон
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")
Структура графа
- Узлы — сущности (компании, люди, продукты) с описаниями
- Рёбра — типизированные, ориентированные отношения (
invested_in,acquired,CEO_of, …)
Поиск через просмотр графа
- Извлекаем сущности, упомянутые в запросе
- Начинаем с атрибутов этих узлов и прямых рёбер
- По желанию расширяем поиск на несколько шагов для получения более полного контекста
Когда использовать этот подход
- При наличии чётко определённых сущностей и названных отношений
4. Агентный RAG
Модель планирует поиск: какие инструменты использовать, в каком порядке и когда будет достаточно доказательств.
Проблема фиксированных конвейеров
Наивные, гибридные и графовые конвейеры жестко задают один путь для каждого вопроса. Двусмысленные или многоэтапные запросы требуют разветвления.
Паттерн — цикл ReAct
ReAct = Размышление + Действие: агент размышляет, вызывает инструмент поиска или вычислений, наблюдает за результатами и повторяет процесс до тех пор, пока не сможет дать ответ.
"""
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")
Когда его использовать
- Ответы, требующие нескольких этапов поиска
- Двусмысленные запросы, план которых неизвестен изначально
- Сочетание поиска с вычислениями или другими действиями
- Исследовательские последующие диалоги
Выберите самый простой вариант шаблона, соответствующий формату вопроса: «naive» для простых демонстраций, «hybrid» для поиска текста в производственных системах, «graphs» для работы с реляционными данными, а «agents» — когда сам план поиска должен быть динамическим.