如果您能看到 RAG 为什么会失败？用 Project_Golem 和 Milvus 在 3D 中调试 RAG

当 RAG 检索出错时，你通常知道它出了问题--相关的文档没有显示出来，或者不相关的文档显示出来了。但要找出原因就另当别论了。你所能利用的只有相似性得分和结果的平面列表。你根本无法看到文档在向量空间中的实际位置，也不知道文档块之间的关系，更不知道你的查询相对于它应该匹配的内容落在了哪里。在实践中，这意味着 RAG 的调试主要靠反复试验：调整分块策略、更换 Embeddings 模型、调整 top-k，然后希望结果有所改善。

Project_Golem是一款开源工具，能让向量空间变得可见。它使用 UMAP 将高维嵌入投射到 3D 中，并使用 Three.js 在浏览器中交互式地渲染它们。你无需猜测检索失败的原因，只需在一个可视化界面中就能看到数据块的语义聚类、查询的落脚点以及检索到的文档。

这真是太神奇了。然而，最初的 Project_Golem 是为小型演示而设计的，而不是真实世界的系统。它依赖于平面文件、暴力搜索和全数据集重建，这意味着当数据增长超过几千个文档时，它很快就会崩溃。

为了弥补这一缺陷，我们将 Project_Golem 与Milvus（特别是 2.6.8 版）集成，作为其向量骨干。Milvus 是一个开源的高性能向量数据库，可处理实时摄取、可扩展索引和毫秒级检索，而 Project_Golem 则专注于其最擅长的领域：使向量检索行为可见。它们共同将三维可视化从玩具演示变成了生产型 RAG 系统的实用调试工具。

在这篇文章中，我们将介绍 Project_Golem，并展示我们如何将其与 Milvus 集成，使向量搜索行为可观察、可扩展，并为生产做好准备。

什么是 Project_Golem？

RAG 调试很难，原因很简单：向量空间是高维的，人类无法看到它们。

Project_Golem是一种基于浏览器的工具，可以让你看到 RAG 系统操作符的向量空间。它将驱动检索的高维 Embeddings（通常为 768 或 1536 维）投射到一个交互式三维场景中，你可以直接进行探索。

下面是它的工作原理：

• 使用 UMAP 降低维度。Project_Golem 使用 UMAP 将高维向量压缩到三维，同时保留它们之间的相对距离。在原始空间中语义相似的数据块在三维投影中会靠得很近；而不相关的数据块则会相距甚远。
• 使用 Three.js 进行三维渲染。在浏览器中渲染的 3D 场景中，每个文档块都显示为一个节点。你可以旋转、缩放和探索空间，查看文档的聚类情况--哪些主题紧密地组合在一起，哪些主题相互重叠，以及边界在哪里。
• 查询时突出显示。当您运行查询时，检索仍在原始高维空间中使用余弦相似性进行。但一旦结果返回，检索到的信息块就会在三维视图中亮起。您可以立即看到您的查询相对于结果的位置--同样重要的是，相对于没有检索到的文档的位置。

这就是 Project_Golem 在调试方面的作用。你不用盯着结果的排序列表猜测为什么某个相关文档会被漏掉，而是可以看到它是否位于一个遥远的集群中（嵌入问题），是否与不相关的内容重叠（分块问题），或者只是勉强在检索阈值之外（配置问题）。三维视图将抽象的相似性得分转化为可以推理的空间关系。

为什么 Project_Golem 无法投入生产

Project_Golem 是作为可视化原型设计的，在这方面运行良好。但它的架构所做的假设在大规模使用时很快就会被打破--如果你想用它来进行实际的 RAG 调试，这些假设就很重要。

每次更新都需要全面重建

这是最基本的限制。在最初的设计中，添加新文档会触发整个管道的重建：重新生成 Embeddings 并将其写入 .npy 文件，在整个数据集上重新运行 UMAP，并以 JSON 格式重新导出 3D 坐标。

即使是 100,000 个文档，单核 UMAP 运行也需要 5-10 分钟。到了百万文档级别，就完全不可行了。对于任何持续变化的数据集（如新闻源、文档、用户对话），都不能使用这种方法，因为每次更新都意味着要等待一个完整的重新处理周期。

粗暴搜索无法扩展

检索方面也有自己的上限。最初的实现使用 NumPy 进行粗暴的余弦相似性搜索--线性时间复杂度，没有索引。在百万文档数据集上，单次查询可能需要超过一秒的时间。这对于任何交互式或在线系统来说都是无法使用的。

内存压力使问题更加复杂。每个 768 维 float32 向量大约需要 3 KB，因此一个百万向量数据集需要超过 3 GB 的内存--所有数据都加载到一个平面 NumPy 数组中，没有索引结构来提高搜索效率。

无元数据过滤，无多租户

在实际的 RAG 系统中，向量相似性很少是唯一的检索标准。你几乎总是需要通过元数据进行过滤，如文档类型、时间戳、用户权限或应用级边界。例如，客户支持 RAG 系统需要将检索范围限定为特定租户的文档，而不是在所有人的数据中进行搜索。

Project_Golem 不支持这些。没有 ANN 索引（如 HNSW 或 IVF），没有标量过滤，没有租户隔离，也没有混合搜索。这是一个可视化层，下面没有生产检索引擎。

Milvus 如何为 Project_Golem 的检索层提供动力

上一节指出了三个差距：每次更新都要完全重建、强制搜索以及没有元数据感知检索。所有这三点都源于同一个根本原因--Project_Golem 没有数据库层。检索、存储和可视化纠结在一个管道中，因此更改任何部分都会强制重建所有内容。

解决办法不是优化管道，而是将其拆分。而是将其拆分开来。

通过集成 Milvus 2.6.8 作为向量骨干，检索成为一个独立于可视化操作的专用生产级层。Milvus 负责向量存储、索引和搜索。Project_Golem 则纯粹专注于渲染--从 Milvus 中获取文档 ID 并在 3D 视图中突出显示。

这种分离产生了两个简洁、独立的流程：

检索流程（在线，毫秒级）

• 使用 OpenAI Embeddings 将您的查询转换为向量。
• 查询向量被发送到 Milvus Collections。
• Milvus AUTOINDEX 会选择并优化适当的索引。
• 实时余弦相似性搜索会返回相关文档 ID。

可视化流程（离线，演示规模）

• UMAP 在接收数据时生成三维坐标（n_neighbors=30，min_dist=0.1）。
• 坐标存储在 golem_cortex.json 中。
• 前端使用 Milvus 返回的文档 ID 高亮显示相应的 3D 节点。

关键点：检索不再等待可视化。您可以摄入新文档并立即进行搜索，三维视图会按照自己的计划进行搜索。

流节点的变化

这种实时摄取由 Milvus 2.6.8 中的一项新功能提供支持：流节点。在早期版本中，实时摄取需要一个外部消息队列，如 Kafka 或 Pulsar。流节点将这种协调转移到了 Milvus 本身--新向量被持续摄取，索引被增量更新，新添加的文档可立即被搜索，无需完全重建，也没有外部依赖性。

对于 Project_Golem，这就是架构的实用之处。您可以不断向 RAG 系统添加文档--新文章、更新文档、用户生成的内容--并且检索保持最新，而不会触发昂贵的 UMAP → JSON → 重新加载循环。

将可视化扩展到百万级别（未来之路）

在 Milvus 的支持下，Project_Golem 目前可支持约 10,000 个文档的交互式演示。检索规模远不止于此，Milvus 可以处理数百万个文档，但可视化管道仍然依赖于 UMAP 的批量运行。为了缩小这一差距，该架构可通过增量可视化管道进行扩展：

• 更新触发器：系统会监听 Milvus Collections 上的插入事件。一旦新添加的文档达到定义的阈值（例如 1,000 条），就会触发增量更新。

• 增量预测：新向量不是在整个数据集上重新运行 UMAP，而是使用 UMAP 的 transform() 方法投射到现有的三维空间中。这样既保留了全局结构，又大大降低了计算成本。

• 前端同步：更新的坐标片段通过 WebSocket 流式传输到前端，允许新节点动态出现，而无需重新加载整个场景。

除了可扩展性，Milvus 2.6.8 还通过将向量相似性与全文搜索和标量过滤相结合，实现了混合搜索。这为更丰富的三维交互（如关键字高亮显示、类别过滤和基于时间的切片）打开了大门，为开发人员探索、调试和推理 RAG 行为提供了更强大的方法。

如何使用 Milvus 部署和探索 Project_Golem

升级后的 Project_Golem 现已在GitHub 上开源。使用 Milvus 官方文档作为我们的数据集，我们将以三维方式介绍可视化 RAG 检索的全过程。设置使用 Docker 和 Python，即使从头开始也很容易。

前提条件

• Docker ≥ 20.10
• Docker Compose ≥ 2.0
• Python ≥ 3.11
• OpenAI API 密钥
• 数据集（Milvus 文件 Markdown 格式）

1.部署 Milvus

Download docker-compose.yml
wget https://github.com/milvus-io/milvus/releases/download/v2.6.8/milvus-standalone-docker-compose.yml -O docker-compose.yml
Start Milvus（verify port mapping：19530:19530）
docker-compose up -d
Verify that the services are running
docker ps | grep milvus
You should see three containers：milvus-standalone, milvus-etcd, milvus-minio

2.核心实现

Milvus 集成（ingest.py）

注：该实现最多支持八个文档类别。如果类别数量超过此限制，颜色将以循环方式重复使用。

from pymilvus import MilvusClient
from pymilvus.milvus_client.index import IndexParams
from openai import OpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter
import umap
from sklearn.neighbors import NearestNeighbors
import json
import numpy as np
import os
import glob
--- CONFIG ---
MILVUS_URI = "http://localhost:19530"
COLLECTION_NAME = "golem_memories"
JSON_OUTPUT_PATH = "./golem_cortex.json"
Data directory (users place .md files in this folder)
DATA_DIR = "./data"
OpenAI Embedding Config
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
OPENAI_BASE_URL = "https://api.openai.com/v1"  #
OPENAI_EMBEDDING_MODEL = "text-embedding-3-small"
1536 dimensions
EMBEDDING_DIM = 1536
Color mapping: colors are assigned automatically and reused in a round-robin manner
COLORS = [
[0.29, 0.87, 0.50],
Green
[0.22, 0.74, 0.97],
Blue
[0.60, 0.20, 0.80],
Purple
[0.94, 0.94, 0.20],
Gold
[0.98, 0.55, 0.00],
Orange
[0.90, 0.30, 0.40],
Red
[0.40, 0.90, 0.90],
Cyan
[0.95, 0.50, 0.90],
Magenta
]
def get_embeddings(texts):
"""Batch embedding using OpenAI API"""
client = OpenAI(api_key=OPENAI_API_KEY, base_url=OPENAI_BASE_URL)
embeddings = []
batch_size = 100
OpenAI allows multiple texts per request
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
response = client.embeddings.create(
model=OPENAI_EMBEDDING_MODEL,
input=batch
)
embeddings.extend([item.embedding for item in response.data])
print(f"   ↳ Embedded {min(i + batch_size, len(texts))}/{len(texts)}...")
return np.array(embeddings)
def load_markdown_files(data_dir):
"""Load all markdown files from the data directory"""
md_files = glob.glob(os.path.join(data_dir, "**/*.md"), recursive=True)
if not md_files:
print(f"   ❌ ERROR: No .md files found in '{data_dir}'")
print(f"   👉 Create a '{data_dir}' folder and put your markdown files there.")
print(f"   👉 Example: {data_dir}/doc1.md, {data_dir}/docs/doc2.md")
return None
docs = []
print(f"\n📚 FOUND {len(md_files)} MARKDOWN FILES:")
for i, file_path in enumerate(md_files):
filename = os.path.basename(file_path)
Categories are derived from the file’s path relative to data_dir
rel_path = os.path.relpath(file_path, data_dir)
category = os.path.dirname(rel_path) if os.path.dirname(rel_path) else "default"
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
docs.append({
"title": filename,
"text": content,
"cat": category,
"path": file_path
})
print(f"   {i+1}. [{category}] {filename}")
return docs
def ingest_dense():
print(f"🧠 PROJECT GOLEM - NEURAL MEMORY BUILDER")
print(f"=" * 50)
if not OPENAI_API_KEY:
print("   ❌ ERROR: OPENAI_API_KEY environment variable not set!")
print("   👉 Run: export OPENAI_API_KEY='your-key-here'")
return
print(f"   ↳ Using OpenAI Embedding: {OPENAI_EMBEDDING_MODEL}")
print(f"   ↳ Embedding Dimension: {EMBEDDING_DIM}")
print(f"   ↳ Data Directory: {DATA_DIR}")
1. Load local markdown files
docs = load_markdown_files(DATA_DIR)
if docs is None:
return
2. Split documents into chunks
print(f"\n📦 PROCESSING DOCUMENTS...")
splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=50)
chunks = []
raw_texts = []
colors = []
chunk_titles = []
categories = []
for doc in docs:
doc_chunks = splitter.create_documents([doc['text']])
cat_index = hash(doc['cat']) % len(COLORS)
for i, chunk in enumerate(doc_chunks):
chunks.append({
"text": chunk.page_content,
"title": doc['title'],
"cat": doc['cat']
})
raw_texts.append(chunk.page_content)
colors.append(COLORS[cat_index])
chunk_titles.append(f"{doc['title']} (chunk {i+1})")
categories.append(doc['cat'])
print(f"   ↳ Created {len(chunks)} text chunks from {len(docs)} documents")
3. Generate embeddings
print(f"\n🔮 GENERATING EMBEDDINGS...")
vectors = get_embeddings(raw_texts)
4. 3D Projection (UMAP)
print("\n🎨 CALCULATING 3D MANIFOLD...")
reducer = umap.UMAP(n_components=3, n_neighbors=30, min_dist=0.1, metric='cosine')
embeddings_3d = reducer.fit_transform(vectors)
5. Wiring (KNN)
print("   ↳ Wiring Synapses (finding connections)...")
nbrs = NearestNeighbors(n_neighbors=8, metric='cosine').fit(vectors)
distances, indices = nbrs.kneighbors(vectors)
6. Prepare output data
cortex_data = []
milvus_data = []
for i in range(len(chunks)):
cortex_data.append({
"id": i,
"title": chunk_titles[i],
"cat": categories[i],
"pos": embeddings_3d[i].tolist(),
"col": colors[i],
"nbs": indices[i][1:].tolist()
})
milvus_data.append({
"id": i,
"text": chunks[i]['text'],
"title": chunk_titles[i],
"category": categories[i],
"vector": vectors[i].tolist()
})
with open(JSON_OUTPUT_PATH, 'w') as f:
json.dump(cortex_data, f)
7. Store vectors in Milvus
print("\n💾 STORING IN MILVUS...")
client = MilvusClient(uri=MILVUS_URI)
Drop existing collection if it exists
if client.has_collection(COLLECTION_NAME):
print(f"   ↳ Dropping existing collection '{COLLECTION_NAME}'...")
client.drop_collection(COLLECTION_NAME)
Create new collection
print(f"   ↳ Creating collection '{COLLECTION_NAME}' (dim={EMBEDDING_DIM})...")
client.create_collection(
collection_name=COLLECTION_NAME,
dimension=EMBEDDING_DIM
)
Insert data
print(f"   ↳ Inserting {len(milvus_data)} vectors...")
client.insert(
collection_name=COLLECTION_NAME,
data=milvus_data
)
Create index for faster search
print("   ↳ Creating index...")
index_params = IndexParams()
index_params.add_index(
field_name="vector",
index_type="AUTOINDEX",
metric_type="COSINE"
)
client.create_index(
collection_name=COLLECTION_NAME,
index_params=index_params
)
print(f"\n✅ CORTEX GENERATED SUCCESSFULLY!")
print(f"   📊 {len(chunks)} memory nodes stored in Milvus")
print(f"   📁 Cortex data saved to: {JSON_OUTPUT_PATH}")
print(f"   🚀 Run 'python GolemServer.py' to start the server")
if __name__ == "__main__":
ingest_dense()

前端可视化（GolemServer.py）

from flask import Flask, request, jsonify, send_from_directory
from openai import OpenAI
from pymilvus import MilvusClient
import json
import os
import sys
--- CONFIG ---
Explicitly set the folder to where this script is located
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
OpenAI Embedding Config
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
OPENAI_BASE_URL = "https://api.openai.com/v1"
OPENAI_EMBEDDING_MODEL = "text-embedding-3-small"
Milvus Config
MILVUS_URI = "http://localhost:19530"
COLLECTION_NAME = "golem_memories"
These match the files generated by ingest.py
JSON_FILE = "golem_cortex.json"
UPDATED: Matches your new repo filename
HTML_FILE = "index.html"
app = Flask(__name__, static_folder=BASE_DIR)
print(f"\n🧠 PROJECT GOLEM SERVER")
print(f"   📂 Serving from: {BASE_DIR}")
--- DIAGNOSTICS ---
Check if files exist before starting
missing_files = []
if not os.path.exists(os.path.join(BASE_DIR, JSON_FILE)):
missing_files.append(JSON_FILE)
if not os.path.exists(os.path.join(BASE_DIR, HTML_FILE)):
missing_files.append(HTML_FILE)
if missing_files:
print(f"   ❌ CRITICAL ERROR: Missing files in this folder:")
for f in missing_files:
print(f"      - {f}")
print("   👉 Did you run 'python ingest.py' successfully?")
sys.exit(1)
else:
print(f"   ✅ Files Verified: Cortex Map found.")
Check API Key
if not OPENAI_API_KEY:
print(f"   ❌ CRITICAL ERROR: OPENAI_API_KEY environment variable not set!")
print("   👉 Run: export OPENAI_API_KEY='your-key-here'")
sys.exit(1)
print(f"   ↳ Using OpenAI Embedding: {OPENAI_EMBEDDING_MODEL}")
print("   ↳ Connecting to Milvus...")
milvus_client = MilvusClient(uri=MILVUS_URI)
Verify collection exists
if not milvus_client.has_collection(COLLECTION_NAME):
print(f"   ❌ CRITICAL ERROR: Collection '{COLLECTION_NAME}' not found in Milvus.")
print("   👉 Did you run 'python ingest.py' successfully?")
sys.exit(1)
Initialize OpenAI client
openai_client = OpenAI(api_key=OPENAI_API_KEY, base_url=OPENAI_BASE_URL)
--- ROUTES ---
@app.route('/')
def root():
Force serve the specific HTML file
return send_from_directory(BASE_DIR, HTML_FILE)
@app.route('/')
def serve_static(filename):
return send_from_directory(BASE_DIR, filename)
@app.route('/query', methods=['POST'])
def query_brain():
data = request.json
text = data.get('query', '')
if not text: return jsonify({"indices": []})
print(f"🔎 Query: {text}")
Get query embedding from OpenAI
response = openai_client.embeddings.create(
model=OPENAI_EMBEDDING_MODEL,
input=text
)
query_vec = response.data[0].embedding
Search in Milvus
results = milvus_client.search(
collection_name=COLLECTION_NAME,
data=[query_vec],
limit=50,
output_fields=["id"]
)
Extract indices and scores
indices = [r['id'] for r in results[0]]
scores = [r['distance'] for r in results[0]]
return jsonify({
"indices": indices,
"scores": scores
})
if __name__ == '__main__':
print("   ✅ SYSTEM ONLINE: http://localhost:8000")
app.run(port=8000)

下载数据集并将其放入指定目录

https://github.com/milvus-io/milvus-docs/tree/v2.6.x/site/en

3.启动项目

将文本 Embeddings 转换为三维空间

python ingest.py

图像

启动前端服务

python GolemServer.py

4.可视化和交互

前台收到检索结果后，会根据余弦相似度得分对节点亮度进行缩放，同时保留节点的原始颜色，以保持清晰的类别聚类。从查询点到每个匹配节点之间会画出半透明的线条，摄像头会平滑地平移和缩放，以聚焦于激活的簇。

示例 1：域内匹配

查询"Milvus 支持哪些索引类型？"

可视化行为：

• 在三维空间中，标有 INDEXES 的红色集群中约有 15 个节点的亮度明显增加（约 2-3倍）。

• 匹配的节点包括 index_types.md、hnsw_index.md 和 ivf_index.md 等文档中的大块内容。

• 从查询向量到每个匹配节点的半透明线被渲染出来，摄像头平滑地聚焦在红色集群上。

示例 2：域外查询拒绝

查询"肯德基超值套餐多少钱？"

可视化行为：

• 所有节点都保留了原来的颜色，只是大小略有变化（小于 1.1×）。

• 匹配的节点分散在不同颜色的多个群组中，没有显示出明显的语义集中。

• 由于未达到相似性阈值（0.5），摄像机不会触发聚焦操作。

结论

Project_Golem 与 Milvus 的搭配不会取代现有的 RAG 评估管道，但它增加了大多数管道完全缺乏的功能：查看向量空间内部发生了什么。

有了这种设置，你就能分辨出不同的检索失败情况，前者是由于糟糕的 Embeddings 导致的，后者是由于糟糕的分块导致的，还有一种情况是由于阈值略微过紧导致的。这种诊断过去需要猜测和反复试验。现在你可以看到它。

当前的集成支持演示规模（约 10,000 个文档）的交互式调试，Milvus 向量数据库在幕后处理生产级检索。通往百万规模可视化的道路已经规划好，但尚未构建，因此现在是参与其中的好时机。

在 GitHub 上查看Project_Golem，用你自己的数据集试试看，看看你的向量空间到底是什么样子的。

如果您有问题或想分享您的发现，请加入我们的Slack 频道，或预约Milvus Office Hours课程，以获得有关设置的实践指导。