什么是RAG？朴素模式、混合模式、图模型以及智能体模式
检索增强生成的实际应用指南：经典向量RAG、混合检索、GraphRAG以及代理模式——它们各自适用的场景以及在实际应用中可能出现的问题。
RAG（检索增强生成）在生成答案时借助外部知识源，而不仅仅依赖模型自身的记忆。系统会首先检索出相关段落并将其作为上下文提供给大语言模型，使其能够有依据地给出答案。这就好比开卷考试：模型是学生，RAG则提供了参考资料。如此一来，无需重新训练即可持续更新答案。
RAG的工作原理
分为三个阶段：
- 索引——将文档分割成可搜索的存储单元（向量数据库、图结构、倒排索引等）
- 检索——为查询找到最相关的文档片段
- 生成——将这些片段交给大语言模型以合成答案
检索质量至关重要。即便模型性能很强，但如果检索能力不足，生成的答案依然会很差；真正的设计难点在于如何高效地找到合适的上下文。
1. 简单版RAG
最简单的模式：嵌入文档，存储向量，在查询时通过相似度进行检索。
该模式
将文档拆分成多个块，对它们进行嵌入并存储向量。在查询时，对问题进行嵌入，找出最近的邻居，然后让模型根据这些片段生成答案。
适用场景
- 原型与演示
- 规模较小且结构清晰的语料库（块数约在5万以下）
- 语义相似度能够匹配用户意图的领域（散文式问答）
不适用场景
- 嵌入模型无法识别的精确标记（名称、代码、ID等）
- 没有元数据过滤功能（日期、来源、类别等）
- 未进行重排的简单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 ──┘
第一阶段——元数据过滤在嵌入处理之前，通过结构化字段缩小候选范围。
第二阶段——混合搜索同时运行向量相似度计算和BM25检索，从而使语义信息和精确术语都能获得评分。
第三阶段——RRF融合通过互反排名融合方法（score = 1 / (k + rank)）将排序后的列表合并，而不对不兼容的分数进行标准化处理。
第四阶段——交叉编码器重排序会对(查询词, 文档)对进行联合评分，从而得出更精确的最终排序结果——速度较慢，但精度更高。
适用场景
- 需要同时兼顾召回率和精度的生产环境
- 包含有用元数据的语料库
- 混合了语义查询和关键词查询的场景
- 简单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？”、“微软收购了什么？”、“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")
适用场景
- 需要多次检索才能得到答案的情况
- 初始阶段就无法确定处理方案的模糊查询
- 将检索与计算或其他操作结合使用的场景
- 需要探索性跟进对话的情况
选择与问题类型最匹配的轻量级模式：简单模式适用于简洁的演示，混合模式适合生产环境中的文本搜索，图结构模式用于关系型数据的处理，而当检索计划本身需要动态调整时则选用智能代理模式。