在实际项目中,如果团队希望保持对技术细节的完全控制,或受限于环境无法引入重量级依赖,可以选择不使用 LangChain、LlamaIndex 等框架,直接基于 Python 与相关 API 构建 RAG 系统。这种“原生”方案虽然需要更多底层编码,但结构透明、依赖少、调试方便,非常适合快速验证原型或对定制化要求高的场景。
以下是一个可运行的最小完整实现,包含索引构建、检索和生成三个核心环节,所有代码基于 OpenAI 兼容 API(可替换为其他模型服务)和一款轻量向量数据库 chromadb。
1. 整体架构与依赖
方案仅依赖三个库:openai(模型调用)、chromadb(向量存储与检索)、tiktoken(估算文本长度,非强制)。安装方式:
pip install openai chromadb tiktoken
工作流程概括为:
- 准备文档 → 切分为小块(chunk)
- 将每个 chunk 用嵌入模型转为向量 → 存入 chromadb 集合
- 用户提问时,先将问题向量化 → 在集合中检索相似片段
- 将检索到的片段拼接进提示词 → 调用 LLM 生成回答
2. 索引构建:文档切片与向量存储
假设我们有一批企业文档,以字符串列表形式提供。以下代码完成切片、嵌入和入库。
import os
from openai import OpenAI
import chromadb
from chromadb.config import Settings
# 初始化客户端
client = OpenAI(api_key="your-api-key") # 或使用环境变量 OPENAI_API_KEY
chroma_client = chromadb.Client(Settings(anonymized_telemetry=False))
collection = chroma_client.get_or_create_collection(name="my_docs")
# 文档切片函数(简单按固定长度切分,可改为更精细的分割策略)
def chunk_text(text, chunk_size=500, overlap=50):
words = text.split()
chunks = []
for i in range(0, len(words), chunk_size - overlap):
chunk = " ".join(words[i:i + chunk_size])
chunks.append(chunk)
return chunks
# 待索引的文档
documents = [
"公司年假政策:员工每年享有15天带薪年假,工作每满一年增加1天,上限20天...",
"报销流程:差旅费需在返回后一周内提交报销单,附票据原件...",
# 更多文档...
]
# 执行切片与嵌入
for doc_id, doc_text in enumerate(documents):
chunks = chunk_text(doc_text)
for chunk_idx, chunk in enumerate(chunks):
# 生成嵌入向量
response = client.embeddings.create(
model="text-embedding-3-small",
input=chunk
)
embedding = response.data[0].embedding
# 存入chroma,附带元数据(便于溯源)
collection.add(
embeddings=[embedding],
documents=[chunk],
metadatas=[{"source": f"doc_{doc_id}", "chunk": chunk_idx}],
ids=[f"doc_{doc_id}_chunk_{chunk_idx}"]
)
print(f"索引完成,共存入 {collection.count()} 个片段")
要点说明:
- 切片大小(
chunk_size)和重叠量(overlap)需要根据文档特性调整,通常 300~800 词且保留少量重叠可保证语义完整。 - 元数据中存储来源信息,为后续溯源提供支持。
- 生产环境中文档可能为 PDF、Word 等,需先转换为纯文本。
3. 检索:根据问题召回相关片段
用户提问时,将问题向量化,然后从 chromadb 中取回最相似的 top-k 个片段。
def retrieve(query, top_k=3):
# 将问题转为向量
query_embedding = client.embeddings.create(
model="text-embedding-3-small",
input=query
).data[0].embedding
# 相似度搜索
results = collection.query(
query_embeddings=[query_embedding],
n_results=top_k,
include=["documents", "metadatas"]
)
# 提取检索结果
docs = results["documents"][0] # 文本片段列表
metas = results["metadatas"][0] # 对应元数据
return docs, metas
该函数返回两个列表:文本内容和各自的元数据,供后续生成和溯源使用。
4. 生成:基于检索结果回答
将检索到的片段与用户问题组合成一个提示词,调用聊天模型生成答案,并要求标注出处。
def generate_answer(query):
docs, metas = retrieve(query)
# 如果没有检索到任何内容
if not docs:
return "抱歉,没有找到与您问题相关的内部资料。"
# 拼接上下文
context_parts = []
for i, (doc, meta) in enumerate(zip(docs, metas)):
context_parts.append(f"[来源 {i+1}: {meta['source']}]\n{doc}")
context = "\n\n".join(context_parts)
# 构造提示词
prompt = f"""你是一个企业知识库助手。请严格根据以下资料回答问题。
如果资料不足以回答,请明确说“根据已有资料无法确定”,不要编造。
资料:
{context}
问题:{query}
请用简洁的段落回答,并在每个关键事实后标注来源编号(如[来源1])。"""
# 调用LLM
response = client.chat.completions.create(
model="gpt-4o-mini", # 可根据需要选择
messages=[
{"role": "system", "content": "你是一个准确、谨慎的助手。"},
{"role": "user", "content": prompt}
],
temperature=0.2 # 降低随机性,让输出更忠实于材料
)
answer = response.choices[0].message.content
return answer
使用时只需:
question = "年假怎么计算?"
answer = generate_answer(question)
print(answer)
输出示例:
根据公司政策,员工每年享有15天带薪年假,工作每满一年增加1天,上限为20天 [来源1]。
5. 简单的溯源与展示增强
在前端或日志中,可将生成答案与检索来源一同返回,方便用户核对:
def ask_with_sources(query):
docs, metas = retrieve(query)
answer = generate_answer(query) # 内部已调用 retrieve,但这里我们也可复用
return {
"question": query,
"answer": answer,
"sources": [{"content": doc, "source": meta["source"]} for doc, meta in zip(docs, metas)]
}
这样,任何一个调用方都能同时获得答案和支撑证据,满足可溯源需求。
6. 生产化注意事项
- 嵌入模型选择:OpenAI 的 text-embedding-3-small 性价比高,离线场景可换用本地部署的嵌入模型(如 sentence-transformers)。
- 向量数据库:chromadb 适合轻量使用和开发测试;数据量增大后,可切换为 Milvus、Qdrant、Weaviate 等,同时保持相似调用逻辑。
- 切片策略优化:固定长度切片简单,但对表格、代码、段落边界不友好。可引入按标题、段落、句子分割的工具(如
spaCy、LangChain的文本分割器,即使不用框架也可单独引用这部分逻辑)。 - 提示词安全性:注意防止提示词注入,避免用户输入恶意指令改变助手行为。
- 成本控制:缓存常见问题的嵌入和回答,避免重复调用生成模型。
- 更新与删除:知识变更时,通过 chromadb 的
update或delete方法按 ID 精确更新片段。
总结:这个无框架原生方案仅用百余行代码便实现了 RAG 的核心链路,完全透明可定制,适合团队在理解底层原理后,根据实际需求逐步扩展为更健壮的系统。它证明了 RAG 并不必然依赖庞大框架,关键技术本身足够轻量、可控。