在当今的 AI 应用开发领域，检索增强生成（Retrieval-Augmented Generation，RAG）已成为构建智能对话系统的核心技术架构。本文将从实战角度出发，详细讲解如何将 RAG 架构与向量数据库进行深度集成，并提供可直接运行的代码示例。作为演示平台，我将使用 HolySheep AI 作为后端推理服务。
核心概念解析
RAG 架构的核心思想是将大型语言模型（LLM）的生成能力与外部知识库的检索能力相结合。当用户提出问题时，系统首先从向量数据库中检索相关的文档片段，然后将这些片段作为上下文提供给 LLM，从而生成更加准确和基于事实的回答。
向量数据库在这一架构中扮演着至关重要的角色。它负责将文本、图像或其他形式的数据转换为高维向量表示（embeddings），并提供高效的相似性搜索能力。当用户查询时，系统会将查询转换为向量，然后在数据库中寻找与之最相似的向量，返回对应的原始数据。
服务提供商对比分析
| 特性 | HolySheep AI | OpenAI API | 其他中转服务 |
|---|---|---|---|
| Embeddings 定价 | ¥1=$1（约 $0.15/MTok） | $0.13/MTok | ¥8-15/百万 tokens |
| LLM 推理定价 | GPT-4.1 $8, Claude 4.5 $15, DeepSeek V3.2 $0.42/MTok | GPT-4o $5/MTok | 浮动定价 |
| 延迟表现 | <50ms | 100-300ms | 200-500ms |
| 支付方式 | WeChat/Alipay/银行卡 | 国际信用卡 | 微信/支付宝 |
| 免费额度 | 注册即送 Credits | $5 试用额度 | 有限或无 |
| API 稳定性 | 企业级保障 | 高可用 | 参差不齐 |
从上述对比可以看出，HolySheep AI 在价格方面具有明显优势。以 DeepSeek V3.2 为例，其价格仅为 $0.42/MTok，相比官方渠道可节省 85% 以上的成本。同时，其 API 响应延迟控制在 50 毫秒以内，能够满足大多数实时对话系统的需求。
系统架构设计
一个完整的 RAG 系统通常包含以下几个核心组件：文档处理模块负责将原始文档分割成适合处理的文本块；向量化模块使用 embedding 模型将文本块转换为向量表示；向量数据库负责存储这些向量并提供相似性搜索能力；最后，生成模块结合检索结果和用户问题，通过 LLM 生成最终回答。
# RAG 系统核心组件实现
import httpx
import json
from typing import List, Dict, Tuple
class HolySheepEmbedding:
"""HolySheep AI Embeddings API 封装"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.client = httpx.Client(timeout=60.0)
def get_embedding(self, text: str) -> List[float]:
"""获取单条文本的向量表示"""
response = self.client.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "text-embedding-3-small",
"input": text
}
)
response.raise_for_status()
return response.json()["data"][0]["embedding"]
def get_embeddings_batch(self, texts: List[str]) -> List[List[float]]:
"""批量获取文本向量（提升效率）"""
response = self.client.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "text-embedding-3-small",
"input": texts
}
)
response.raise_for_status()
return [item["embedding"] for item in response.json()["data"]]
class SimpleVectorStore:
"""简化的向量数据库实现（生产环境建议使用 Milvus/Pinecone）"""
def __init__(self, dimension: int = 1536):
self.dimension = dimension
self.vectors: List[List[float]] = []
self.metadata: List[Dict] = []
def add_vectors(self, vectors: List[List[float]], metadata: List[Dict]):
"""添加向量及其元数据"""
self.vectors.extend(vectors)
self.metadata.extend(metadata)
def cosine_similarity(self, a: List[float], b: List[float]) -> float:
"""计算余弦相似度"""
dot_product = sum(x * y for x, y in zip(a, b))
norm_a = sum(x * x for x in a) ** 0.5
norm_b = sum(x * x for x in b) ** 0.5
return dot_product / (norm_a * norm_b + 1e-8)
def search(self, query_vector: List[float], top_k: int = 5) -> List[Tuple[Dict, float]]:
"""搜索最相似的向量"""
similarities = [
(meta, self.cosine_similarity(query_vector, vec))
for vec, meta in zip(self.vectors, self.metadata)
]
return sorted(similarities, key=lambda x: x[1], reverse=True)[:top_k]
class RAGSystem:
"""完整的 RAG 检索生成系统"""
def __init__(self, api_key: str):
self.embedding_client = HolySheepEmbedding(api_key)
self.vector_store = SimpleVectorStore()
self.client = httpx.Client(timeout=60.0)
self.base_url = "https://api.holysheep.ai/v1"
def index_documents(self, documents: List[Dict]):
"""为文档建立索引"""
texts = [doc["content"] for doc in documents]
embeddings = self.embedding_client.get_embeddings_batch(texts)
self.vector_store.add_vectors(embeddings, documents)
print(f"已索引 {len(documents)} 条文档")
def retrieve(self, query: str, top_k: int = 3) -> List[Dict]:
"""检索相关文档"""
query_vector = self.embedding_client.get_embedding(query)
results = self.vector_store.search(query_vector, top_k)
return [{"content": r[0]["content"], "score": r[1]} for r in results]
def generate(self, query: str, context_docs: List[Dict]) -> str:
"""基于检索结果生成回答"""
context = "\n\n".join([
f"[文档 {i+1}] {doc['content']}"
for i, doc in enumerate(context_docs)
])
prompt = f"""基于以下参考资料回答用户问题。如果资料中没有相关信息，请说明"我无法从提供的资料中找到答案"。
参考资料：
{context}
用户问题：{query}
回答："""
response = self.client.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.7,
"max_tokens": 1000
}
)
response.raise_for_status()
return response.json()["choices"][0]["message"]["content"]
def query(self, question: str) -> str:
"""完整的问答流程"""
docs = self.retrieve(question)
return self.generate(question, docs)
使用示例
if __name__ == "__main__":
api_key = "YOUR_HOLYSHEEP_API_KEY"
rag = RAGSystem(api_key)
# 添加示例文档
documents = [
{"content": "RAG（检索增强生成）是一种结合检索和生成的 AI 技术架构。", "source": "tech_wiki"},
{"content": "向量数据库用于存储和搜索高维向量表示，如 Milvus、Pinecone。", "source": "tech_wiki"},
{"content": "Embeddings 是将文本转换为向量的技术，是 RAG 的核心组件。", "source": "tech_wiki"},
]
rag.index_documents(documents)
# 执行查询
answer = rag.query("什么是 RAG 技术？")
print(f"回答：{answer}")上述代码展示了一个完整的 RAG 系统实现。在实际部署时，我建议将向量存储部分替换为专业的向量数据库服务，如 Milvus、Chroma 或 Pinecone，以获得更好的扩展性和性能。以下是使用 Chroma 作为向量数据库的增强版本实现。
生产级 RAG 系统实现
在生产环境中，我们需要考虑更多的因素，包括文档的智能分割、向量索引的优化、以及检索结果的重新排序（Re-ranking）。以下是一个更完整的实现方案。
# 生产级 RAG 系统（使用 Chroma 向量数据库）
import httpx
import chromadb
from chromadb.utils import embedding_functions
from bs4 import BeautifulSoup
import re
from dataclasses import dataclass
from typing import Optional
@dataclass
class Document:
"""文档数据结构"""
content: str
metadata: dict
class ProductionRAGSystem:
"""生产级 RAG 系统 - 集成 Chroma 向量数据库"""
def __init__(self, api_key: str, collection_name: str = "knowledge_base"):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.client = httpx.Client(timeout=120.0)
# 初始化 Chroma 客户端（使用内存模式，生产环境建议持久化）
self.chroma_client = chromadb.Client()
# 配置 embedding 函数
self.embedding_fn = embedding_functions.HuggingFaceEmbeddingFunction(
api_key=api_key,
model_name="sentence-transformers/all-MiniLM-L6-v2"
)
# 获取或创建集合
try:
self.collection = self.chroma_client.get_collection(
name=collection_name,
embedding_function=self.embedding_fn
)
except:
self.collection = self.chroma_client.create_collection(
name=collection_name,
embedding_function=self.embedding_fn,
metadata={"description": "知识库向量集合"}
)
def chunk_text(self, text: str, chunk_size: int = 500, overlap: int = 50) -> List[str]:
"""智能文本分割 - 保持语义完整性"""
# 清理文本
text = re.sub(r'\s+', ' ', text).strip()
# 按句子分割
sentences = re.split(r'[。！？\n]', text)
chunks = []
current_chunk = ""
for sentence in sentences:
sentence = sentence.strip()
if not sentence:
continue
if len(current_chunk) + len(sentence) <= chunk_size:
current_chunk += sentence + "。"
else:
if current_chunk:
chunks.append(current_chunk.strip())
# 保留重叠部分
current_chunk = current_chunk[-overlap:] + sentence + "。"
if current_chunk.strip():
chunks.append(current_chunk.strip())
return chunks
def load_documents(self, file_path: str) -> List[Document]:
"""加载并解析文档（支持多种格式）"""
docs = []
if file_path.endswith('.txt'):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
chunks = self.chunk_text(content)
for i, chunk in enumerate(chunks):
docs.append(Document(
content=chunk,
metadata={"source": file_path, "chunk_id": i}
))
elif file_path.endswith('.html'):
with open(file_path, 'r', encoding='utf-8') as f:
soup = BeautifulSoup(f.read(), 'html.parser')
text = soup.get_text(separator=' ', strip=True)
chunks = self.chunk_text(text)
for i, chunk in enumerate(chunks):
docs.append(Document(
content=chunk,
metadata={"source": file_path, "chunk_id": i, "type": "html"}
))
return docs
def index_documents(self, documents: List[Document]):
"""批量索引文档到向量数据库"""
ids = [f"doc_{doc.metadata.get('chunk_id', i)}" for i, doc in enumerate(documents)]
contents = [doc.content for doc in documents]
metadatas = [doc.metadata for doc in documents]
self.collection.add(
ids=ids,
documents=contents,
metadatas=metadatas
)
print(f"成功索引 {len(documents)} 个文档块")
def query(self, question: str, top_k: int = 5, model: str = "gpt-4.1") -> dict:
"""执行完整的 RAG 查询"""
# 第一阶段：向量检索
results = self.collection.query(
query_texts=[question],
n_results=top_k
)
# 整理检索结果
retrieved_docs = []
for i in range(len(results['documents'][0])):
retrieved_docs.append({
"content": results['documents'][0][i],
"metadata": results['metadatas'][0][i],
"distance": results['distances'][0][i]
})
# 构建上下文
context = "\n\n".join([
f"[来源 {doc['metadata'].get('source', 'unknown')}] {doc['content']}"
for doc in retrieved_docs
])
# 构建提示词
prompt = f"""你是一个专业的知识库助手。请根据以下参考资料，准确回答用户的问题。
参考资料：
---
{context}
---
用户问题：{question}
请注意：
1. 只根据提供的参考资料回答问题
2. 如果资料中没有相关信息，诚实地说明"我无法从提供的资料中找到答案"
3. 如果资料中的信息不完整，可以基于现有信息给出合理的推测，但需说明这是推测
4. 回答要清晰、有条理，适当引用参考资料"""
# 调用 LLM 生成回答
try:
response = self.client.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": [
{"role": "system", "content": "你是一个有帮助的 AI 助手。"},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 1500
}
)
response.raise_for_status()
answer = response.json()["choices"][0]["message"]["content"]
except httpx.HTTPStatusError as e:
answer = f"API 请求失败：{e.response.status_code}"
return {
"question": question,
"answer": answer,
"sources": retrieved_docs
}
性能监控装饰器
import time
from functools import wraps
def monitor_performance(func):
"""监控函数执行性能"""
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
elapsed = (time.time() - start_time) * 1000
print(f"[性能] {func.__name__} 执行耗时: {elapsed:.2f}ms")
return result
return wrapper
使用示例
if __name__ == "__main__":
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
rag = ProductionRAGSystem(API_KEY)
# 加载文档并索引
docs = rag.load_documents("knowledge_base.txt")
rag.index_documents(docs)
# 执行查询
result = rag.query("RAG 系统的核心组件有哪些？")
print(f"\n问题：{result['question']}")
print(f"回答：{result['answer']}")
print(f"\n参考来源：{len(result['sources'])} 条")系统优化策略
在实际应用中，我总结了以下几种优化 RAG 系统性能的方法。首先是混合检索策略，结合稠密向量检索（Dense Retrieval）和稀疏向量检索（Sparse Retrieval），如 BM25 算法，能够显著提升检索的召回率和准确性。其次是查询改写（Query Rewrite），通过 LLM 对用户问题进行预处理，使其更适合检索。
第三种优化方法是引入重排序模型（Re-ranker），如 Cohere 的 Rerank API 或 BGE-Reranker，在初次检索后对结果进行二次排序，确保最相关的文档排在最前面。第四种优化是缓存机制，对频繁查询的向量和检索结果进行缓存，减少 API 调用次数和响应延迟。
# RAG 系统优化 - 混合检索与查询改写
import httpx
import hashlib
from collections import OrderedDict
from typing import List, Dict, Optional
import json
class LRUCache:
"""LRU 缓存实现"""
def __init__(self, capacity: int = 100):
self.cache = OrderedDict()
self.capacity = capacity
def get(self, key: str) -> Optional[any]:
if key in self.cache:
self.cache.move_to_end(key)
return self.cache[key]
return None
def put(self, key: str, value: any):
if key in self.cache:
self.cache.move_to_end(key)
self.cache[key] = value
if len(self.cache) > self.capacity:
self.cache.popitem(last=False)
class OptimizedRAG:
"""优化版 RAG 系统 - 包含混合检索与查询改写"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.client = httpx.Client(timeout=120.0)
# 初始化缓存
self.vector_cache = LRUCache(capacity=200)
self.search_cache = LRUCache(capacity=100)
def _cache_key(self, text: str) -> str:
"""生成缓存键"""
return hashlib.md5(text.encode()).hexdigest()
def rewrite_query(self, query: str, style: str = "detailed") -> str:
"""使用 LLM 改写查询，提升检索效果"""
prompts = {
"detailed": f"""将以下问题改写为一个更详细、更适合检索的查询。
要求：
1. 展开缩写和简称
2. 添加相关领域的术语
3. 保持原意不变
4. 只输出改写后的查询，不要其他解释
问题：{query}
改写后的查询：""",
"multiple": f"""为以下问题生成 3 个不同的表述版本，用于多路召回检索。
输出格式：每行一个版本
问题：{query}
版本："""
}
response = self.client.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": prompts[style]}],
"temperature": 0.7,
"max_tokens": 300
}
)
response.raise_for_status()
result = response.json()["choices"][0]["message"]["content"]
if style == "multiple":
return [line.strip() for line in result.split('\n') if line.strip()]
return result
def get_embedding_with_cache(self, text: str) -> List[float]:
"""带缓存的向量获取"""
cache_key = self._cache_key(text)
cached = self.vector_cache.get(cache_key)
if cached:
return cached
response = self.client.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "text-embedding-3-small",
"input": text
}
)
response.raise_for_status()
embedding = response.json()["data"][0]["embedding"]
self.vector_cache.put(cache_key, embedding)
return embedding
def hybrid_search(
self,
query: str,
vector_store,
bm25_index,
top_k: int = 10,
alpha: float = 0.7
) -> List[Dict]:
"""混合搜索 - 结合向量相似度和 BM25 分数"""
# 获取向量
query_vector = self.get_embedding_with_cache(query)
# 向量检索
vector_results = vector_store.search(query_vector, top_k * 2)
# BM25 检索
bm25_results = bm25_index.search(query, top_k * 2)
# 融合分数
combined_scores = {}
# 归一化向量分数
max_vector_score = max(r[1] for r in vector_results) if vector_results else 1
for doc, score in vector_results:
doc_id = doc.get('id', doc.get('content', ''))
combined_scores[doc_id] = alpha * (score / max_vector_score)
# 归一化 BM25 分数
max_bm25_score = max(r[1] for r in bm25_results) if bm25_results else 1
for doc, score in bm25_results:
doc_id = doc.get('id', doc.get('content', ''))
if doc_id in combined_scores:
combined_scores[doc_id] += (1 - alpha) * (score / max_bm25_score)
else:
combined_scores[doc_id] = (1 - alpha) * (score / max_bm25_score)
# 排序返回
sorted_results = sorted(
combined_scores.items(),
key=lambda x: x[1],
reverse=True
)[:top_k]
# 重建结果列表
doc_map = {r[0].get('id', r[0].get('content', '')): r[0] for r in vector_results}
doc_map.update({r[0].get('id', r[0].get('content', '')): r[0] for r in bm25_results})
return [
{"document": doc_map[doc_id], "score": score}
for doc_id, score in sorted_results
if doc_id in doc_map
]
def rerank_results(
self,
query: str,
documents: List[Dict],
top_n: int = 5
) -> List[Dict]:
"""使用交叉编码器对结果进行重排序"""
# 构建重排序提示
doc_contents = "\n---\n".join([
f"文档 {i+1}:\n{doc['content'] if isinstance(doc, dict) else doc}"
for i, doc in enumerate(documents)
])
rerank_prompt = f"""请根据问题评估每个文档的相关性，给出 1-10 的评分。
问题：{query}
{doc_contents}
输出格式（每行一个评分）：
文档 1: [分数]
文档 2: [分数]
..."""
try:
response = self.client.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": rerank_prompt}],
"temperature": 0,
"max_tokens": 200
}
)
response.raise_for_status()
scores_text = response.json()["choices"][0]["message"]["content"]
# 解析评分
scores = []
for line in scores_text.split('\n'):
if ':' in line:
score_str = line.split(':')[-1].strip()
try:
scores.append(float(score_str))
except ValueError:
scores.append(5.0) # 默认分数
# 更新文档分数并排序
for i, doc in enumerate(documents):
if i < len(scores):
doc['rerank_score'] = scores[i]
else:
doc['rerank_score'] = 5.0
return sorted(documents, key=lambda x: x.get('rerank_score', 0), reverse=True)[:top_n]
except Exception as e:
print(f"重排序失败：{e}")
return documents[:top_n]
BM25 实现（简化版）
class SimpleBM25:
"""简化的 BM25 搜索引擎"""
def __init__(self, k1: float = 1.5, b: float = 0.75):
self.k1 = k1
self.b = b
self.documents = []
self.avgdl = 0
self.doc_freqs = {}
self.idf = {}
self.doc_lengths = []
def index(self, documents: List[Dict]):
"""构建 BM25 索引"""
self.documents = documents
N = len(documents)
# 计算文档长度
self.doc_lengths = [len(doc.get('content', '').split()) for doc in documents]
self.avgdl = sum(self.doc_lengths) / N if N > 0 else 0
# 统计词频和文档频率
for doc in documents:
words = set(doc.get('content', '').lower().split())
for word in words:
self.doc_freqs[word] = self.doc_freqs.get(word, 0) + 1
# 计算 IDF
for word, df in self.doc_freqs.items():
self.idf[word] = (N - df + 0.5) / (df + 0.5)
def search(self, query: str, top_k: int = 10) -> List[tuple]:
"""搜索相关文档"""
query_words = query.lower().split()
scores = []
for i, doc in enumerate(self.documents):
score = 0
doc_words = doc.get('content', '').lower().split()
word_freq = {}
for word in doc_words:
word_freq[word] = word_freq.get(word, 0) + 1
dl = self.doc_lengths[i]
for word in query_words:
if word in word_freq:
tf = word_freq[word]
idf = self.idf.get(word, 0)
# BM25 公式
numerator = tf * (self.k1 + 1)
denominator = tf + self.k1 * (1 - self.b + self.b * dl / self.avgdl)
score += idf * numerator / denominator
scores.append((doc, score))
return sorted(scores, key=lambda x: x[1], reverse=True)[:top_k]
使用示例
if __name__ == "__main__":
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
rag = OptimizedRAG(API_KEY)
# 查询改写示例
rewritten = rag.rewrite_query("RAG 是什么", style="multiple")
print("查询改写结果：")
for version in rewritten:
print(f" - {version}")
# 构建 BM25 索引
bm25 = SimpleBM25()
bm25.index([
{"id": "1", "content": "RAG 是检索增强生成技术"},
{"id": "2", "content": "向量数据库用于存储向量"},
{"id": "3", "content": "Embeddings 将文本转为向量"},
])常见错误与解决方案
在开发 RAG 系统时，开发者经常会遇到一些典型问题。以下是我总结的三个最常见的错误及其解决方案。
错误一：API 密钥未正确配置
# ❌ 错误示例：API 密钥硬编码在代码中
client = httpx.Client()
response = client.post(
"https://api.holysheep.ai/v1/embeddings",
headers={
"Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY", # 直接暴露密钥
"Content-Type": "application/json"
},
json={"model": "text-embedding-3-small", "input": "text"}
)
✅ 正确示例：从环境变量读取密钥
import os
from dotenv import load_dotenv
load_dotenv() # 加载 .env 文件
API_KEY = os.getenv("HOLYSHEEP_API_KEY")
if not API_KEY:
raise ValueError("未设置 HOLYSHEEP_API_KEY 环境变量")
client = httpx.Client()
response = client.post(
"https://api.holysheep.ai/v1/embeddings",
headers={
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
},
json={"model": "text-embedding-3-small", "input": "text"}
)
.env 文件内容：
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
错误二：向量维度不匹配
# ❌ 错误示例：未指定 embedding 模型，导致维度不一致
class VectorStore:
def __init__(self):
self.vectors = []
def add(self, vector, metadata):
self.vectors.append({"vector": vector, "metadata": metadata})
def search(self, query_vector, top_k=5):
# 假设所有向量都是 1536 维，但实际可能不同
return sorted(
self.vectors,
key=lambda x: self._cosine_similarity(x["vector"], query_vector),
reverse=True
)[:top_k]
✅ 正确示例：使用固定的 embedding 模型并验证维度
import httpx
class VectorStore:
def __init__(self, api_key: str, model: str = "text-embedding-3-small"):
self.api_key = api_key
self.model = model
self.expected_dimensions = self._get_embedding_dimensions()
self.vectors = []
def _get_embedding_dimensions(self) -> int:
"""获取模型的向量维度"""
client = httpx.Client()
response = client.post(
"https://api.holysheep.ai/v1/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.model,
"input": "dimension test"
}
)
return len(response.json()["data"][0]["embedding"])
def add(self, vector, metadata):
"""添加向量，带维度验证"""
if len(vector) != self.expected_dimensions:
raise ValueError(
f"向量维度不匹配：期望 {self.expected_dimensions} 维，"
f"实际 {len(vector)} 维"
)
self.vectors.append({"vector": vector, "metadata": metadata})
def search(self, query_vector, top_k=5):
"""搜索，验证查询向量维度"""
if len(query_vector) != self.expected_dimensions:
raise ValueError(
f"查询向量维度不匹配：期望 {self.expected_dimensions} 维，"
f"实际 {len(query_vector)} 维"
)
scored = [
(item, self._cosine_similarity(item["vector"], query_vector))
for item in self.vectors
]
return sorted(scored, key=lambda x: x[1], reverse=True)[:top_k]
使用示例
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
store = VectorStore(API_KEY)
print(f"Embedding 模型维度：{store.expected_dimensions}")错误三：批量处理时超出速率限制
# ❌ 错误示例：无限制批量请求，导致限流
import httpx
def batch_embed(texts, api_key):
client = httpx.Client()
results = []
for text in texts: # 10000 条文本
response = client.post(
"https://api.holysheep.ai/v1/embeddings",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json