/SKILL.md

name, description, license, metadata

name	description	license	metadata
rag-architect	设计并实现生产级 RAG 系统，涵盖文档分块、嵌入向量生成、向量存储配置、混合检索管道构建、重排序以及检索质量评估。适用于构建 RAG 系统、向量数据库或需要语义搜索、文档检索、上下文增强、相似性搜索或基于嵌入的索引的知识驱动型 AI 应用。	MIT	author version domain triggers role scope output-format related-skills https://github.com/Jeffallan 1.1.0 data-ml RAG, retrieval-augmented generation, vector search, embeddings, semantic search, vector database, document retrieval, knowledge base, context retrieval, similarity search architect system-design architecture python-pro, database-optimizer, monitoring-expert, api-designer

RAG 架构师

核心工作流

1. 需求分析 — 确定检索需求、延迟约束、准确性要求及规模
2. 向量存储设计 — 选择数据库、设计模式、索引策略、分片方案
3. 分块策略 — 文档拆分、重叠、语义边界、元数据丰富
4. 检索管道 — 嵌入模型选择、查询转换、混合检索、重排序
5. 评估与迭代 — 指标追踪、检索调试、持续优化

每一步完成后需进行验证（参见下方检查点）。

参考指南

根据上下文加载详细指导：

主题	参考文件	加载时机
向量数据库	references/vector-databases.md	比较 Pinecone、Weaviate、Chroma、pgvector、Qdrant 时
嵌入模型	references/embedding-models.md	选择嵌入模型、微调、维度权衡时
分块策略	references/chunking-strategies.md	文档拆分、重叠、语义分块时
检索优化	references/retrieval-optimization.md	混合检索、重排序、查询扩展、过滤时
RAG 评估	references/rag-evaluation.md	指标、评估框架、检索调试时

实现示例

1. 文档分块

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 在领域数据上评估 chunk_size——切勿盲目使用 512
splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=100,
    separators=["\n\n", "\n", ". ", " "],
)

chunks = splitter.create_documents(
    texts=[doc.page_content for doc in raw_docs],
    metadatas=[{"source": doc.metadata["source"], "timestamp": doc.metadata.get("timestamp")} for doc in raw_docs],
)

检查点： assert all(c.metadata.get("source") for c in chunks), "缺少来源元数据"

2. 生成嵌入向量并建立索引

from openai import OpenAI
import qdrant_client
from qdrant_client.models import VectorParams, Distance, PointStruct

client = OpenAI()
qdrant = qdrant_client.QdrantClient("localhost", port=6333)

# 创建集合
qdrant.recreate_collection(
    collection_name="knowledge_base",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
)

def embed_chunks(chunks: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    response = client.embeddings.create(input=chunks, model=model)
    return [r.embedding for r in response.data]

# 通过确定性 ID 实现幂等写入与去重
import hashlib, uuid

points = []
for i, chunk in enumerate(chunks):
    doc_id = str(uuid.UUID(hashlib.md5(chunk.page_content.encode()).hexdigest()))
    embedding = embed_chunks([chunk.page_content])[0]
    points.append(PointStruct(id=doc_id, vector=embedding, payload=chunk.metadata))

qdrant.upsert(collection_name="knowledge_base", points=points)

检查点： assert qdrant.count("knowledge_base").count == len(set(p.id for p in points)), "去重失败"

3. 混合检索（向量 + BM25）

from qdrant_client.models import Filter, FieldCondition, MatchValue, SparseVector
from rank_bm25 import BM25Okapi

def hybrid_search(query: str, tenant_id: str, top_k: int = 20) -> list:
    # 稠密检索
    query_embedding = embed_chunks([query])[0]
    tenant_filter = Filter(must=[FieldCondition(key="tenant_id", match=MatchValue(value=tenant_id))])
    dense_results = qdrant.search(
        collection_name="knowledge_base",
        query_vector=query_embedding,
        query_filter=tenant_filter,
        limit=top_k,
    )

    # 稀疏检索（BM25）
    corpus = [r.payload.get("text", "") for r in dense_results]
    bm25 = BM25Okapi([doc.split() for doc in corpus])
    bm25_scores = bm25.get_scores(query.split())

    # 倒数排序融合（Reciprocal Rank Fusion）
    ranked = sorted(
        zip(dense_results, bm25_scores),
        key=lambda x: 0.6 * x[0].score + 0.4 * x[1],
        reverse=True,
    )
    return [r for r, _ in ranked[:top_k]]

检查点： assert len(hybrid_search("test query", tenant_id="demo")) > 0, "混合搜索未返回结果"

4. 对 Top-K 结果进行重排序

import cohere

co = cohere.Client("YOUR_API_KEY")

def rerank(query: str, results: list, top_n: int = 5) -> list:
    docs = [r.payload.get("text", "") for r in results]
    reranked = co.rerank(query=query, documents=docs, top_n=top_n, model="rerank-english-v3.0")
    return [results[r.index] for r in reranked.results]

5. 检索评估

# 基于标注评估集运行 precision@k 和 recall@k
# python evaluate.py --metrics precision@10 recall@10 mrr --collection knowledge_base

from ragas import evaluate
from ragas.metrics import context_precision, context_recall, faithfulness, answer_relevancy
from datasets import Dataset

eval_dataset = Dataset.from_dict({
    "question": questions,
    "contexts": retrieved_contexts,
    "answer": generated_answers,
    "ground_truth": ground_truth_answers,
})

results = evaluate(eval_dataset, metrics=[context_precision, context_recall, faithfulness, answer_relevancy])
print(results)

检查点： 在进入 LLM 集成阶段前，目标应达到 context_precision >= 0.7 且 context_recall >= 0.6。

约束条件

必须执行

• 在确定方案前，先在领域数据上评估多个嵌入模型
• 为生产系统实现混合检索（向量 + 关键词）
• 添加元数据过滤以支持多租户或特定领域检索
• 度量检索指标（precision@k、recall@k、MRR、NDCG）
• 在将上下文传递给 LLM 之前，先对 top-k 结果进行重排序
• 实现带去重的幂等写入（确定性 ID）
• 持续监控检索延迟与质量
• 对嵌入模型进行版本管理，并为模型迁移做好规划

严禁行为

• 未经领域数据评估就使用默认分块大小（512）
• 跳过元数据丰富（来源、时间戳、章节）
• 只关注 LLM 输出质量而忽略检索质量指标
• 存储未经过预处理/清洗的原始文档
• 在复杂的多领域检索中仅使用余弦相似度
• 未在生产级数据量上测试就部署
• 忘记处理边界情况（空结果、格式错误的文档）
• 将嵌入模型与应用程序代码紧密耦合

输出模板

设计 RAG 架构时，应交付以下内容：

1. 系统架构图（写入 + 检索管道）
2. 向量数据库选型及其权衡分析
3. 分块策略及示例与理由说明
4. 检索管道设计（查询→结果流程）
5. 评估计划，包含指标、基准及通过/未通过阈值

文档