ai-data-engineering

AI 数据工程

目的

为 AI/ML 系统构建数据基础设施，包括 RAG 流水线、特征存储和嵌入生成。提供生产级 AI 应用的架构模式、编排工作流和评估指标。

何时使用

在以下情况下使用此技能：

• 构建 RAG（检索增强生成）流水线
• 实现语义搜索或向量数据库
• 设置用于实时服务的 ML 特征存储
• 创建嵌入生成流水线
• 使用 RAGAS 指标评估 RAG 质量
• 为 AI 系统编排数据工作流
• 与前端技能（ai-chat、search-filter）集成

在以下情况下跳过此技能：

• 构建传统 CRUD 应用（使用 databases-relational）
• 简单键值存储（使用 databases-nosql）
• 应用中不包含 AI/ML 组件

RAG 流水线架构

RAG 流水线有 5 个不同阶段。了解此架构对生产环境实现至关重要。

┌─────────────────────────────────────────────────────────────┐
│                    RAG Pipeline (5 Stages)                   │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  1. INGESTION → Load documents (PDF, DOCX, Markdown)        │
│  2. INDEXING → Chunk (512 tokens) + Embed + Store           │
│  3. RETRIEVAL → Query embedding + Vector search + Filters   │
│  4. GENERATION → Context injection + LLM streaming          │
│  5. EVALUATION → RAGAS metrics (faithfulness, relevancy)    │
│                                                              │
└─────────────────────────────────────────────────────────────┘

完整的 RAG 架构和实现模式，请参阅：

• references/rag-architecture.md - 详细介绍 5 个阶段
• examples/langchain-rag/basic_rag.py - 可运行的实现

分块策略

分块是影响 RAG 质量的最关键决策。分块不当会破坏检索效果。

默认建议：

• 大小： 512 个 token
• 重叠： 50-100 个 token
• 方法： 基于固定 token 数

这些取值的原因：

• 太小（<256 个 token）：丢失上下文，需要进行多次检索
• 太大（>1024 个 token）：包含无关内容，触及 token 限制
• 重叠可防止在分块边界处丢失信息

特殊情况下的替代策略：

# Code-aware chunking (preserves functions/classes)
from langchain.text_splitter import RecursiveCharacterTextSplitter

code_splitter = RecursiveCharacterTextSplitter.from_language(
    language="python",
    chunk_size=512,
    chunk_overlap=50
)

# Semantic chunking (splits on meaning, not tokens)
from langchain.text_splitter import SemanticChunker

semantic_splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile"  # Split at semantic boundaries
)

参阅： references/chunking-strategies.md，了解完整的决策框架

嵌入生成

嵌入质量会直接影响检索准确性。目前，Voyage AI 是同类最佳。

主要建议：Voyage AI voyage-3

• 维度：1024
• MTEB 分数：69.0（截至 2025 年 12 月最高）
• 成本：$$$，但比 OpenAI 好 9.74%
• 适用场景：要求最佳检索质量的生产系统

高性价比替代方案：OpenAI text-embedding-3-small

• 维度：1536
• MTEB 分数：62.3
• 成本：$（比 voyage-3 便宜 5 倍）
• 适用场景：开发、原型设计和成本敏感型应用

实现：

from langchain_voyageai import VoyageAIEmbeddings
from langchain_openai import OpenAIEmbeddings

# Production (best quality)
embeddings = VoyageAIEmbeddings(
    model="voyage-3",
    voyage_api_key="your-api-key"
)

# Development (cost-effective)
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    openai_api_key="your-api-key"
)

参阅： references/embedding-strategies.md，了解完整的提供商比较

RAGAS 评估指标

传统指标（BLEU、ROUGE）无法衡量 RAG 质量。RAGAS 提供由大模型担任评审的评估方式。

4 个核心指标：

指标	衡量内容	良好分数
忠实度	与检索上下文的事实一致性	> 0.8
答案相关性	答案是否回应了用户的问题？	> 0.7
上下文精确率	检索到的分块是否确实相关？	> 0.6
上下文召回率	是否检索到了所有必要的分块？	> 0.7

快速评估脚本：

# Run RAGAS evaluation (TOKEN-FREE script execution)
python scripts/evaluate_rag.py --dataset eval_data.json --output results.json

手动实现：

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy

dataset = {
    "question": ["What is the capital of France?"],
    "answer": ["Paris is the capital of France."],
    "contexts": [["France's capital is Paris."]],
    "ground_truth": ["Paris"]
}

result = evaluate(dataset, metrics=[faithfulness, answer_relevancy])
print(f"Faithfulness: {result['faithfulness']}")
print(f"Answer Relevancy: {result['answer_relevancy']}")

参阅： references/evaluation-metrics.md，了解完整的 RAGAS 实现指南

特征存储

特征存储通过提供一致的特征计算，解决“训练-服务偏差”问题。

主要建议：Feast - 开源，可搭配任何后端使用（PostgreSQL、Redis、DynamoDB、S3、BigQuery、Snowflake）

基本用法：

from feast import FeatureStore
store = FeatureStore(repo_path="feature_repo/")

# Online serving (low-latency)
features = store.get_online_features(
    features=["user_features:total_orders"],
    entity_rows=[{"user_id": 1001}]
).to_dict()

参阅： references/feature-stores.md，了解完整的 Feast 设置方法和替代方案（Tecton、Hopsworks）

LangChain 编排

LangChain 是用于大模型编排的主要框架，拥有最大的生态系统（24,215+ 个 API 参考代码片段）。

Context7 库 ID： /websites/langchain_oss_python_langchain（可信度：高，代码片段：435）

基本 RAG 链：

from langchain_core.prompts import ChatPromptTemplate
from langchain_qdrant import QdrantVectorStore
from langchain_voyageai import VoyageAIEmbeddings

# Setup retriever
vectorstore = QdrantVectorStore(
    client=qdrant_client,
    embedding=VoyageAIEmbeddings(model="voyage-3")
)
retriever = vectorstore.as_retriever(search_type="mmr", search_kwargs={"k": 5})

# Build chain
prompt = ChatPromptTemplate.from_template(
    "Answer based on context:\n{context}\n\nQuestion: {question}"
)
chain = {"context": retriever, "question": lambda x: x} | prompt | ChatOpenAI() | StrOutputParser()

# Stream response
for chunk in chain.stream("What is the capital of France?"):
    print(chunk, end="", flush=True)

参阅： references/langchain-patterns.md - 完整的 LangChain 0.3+ 模式，包括流式处理和混合搜索

编排工具

现代 AI 流水线需要 cron 作业之外的工作流编排。

首选推荐：Dagster（适用于 ML/AI 流水线）——以资产为中心的设计、最佳的血缘追踪，非常适合 RAG

示例：嵌入流水线

from dagster import asset
from langchain_voyageai import VoyageAIEmbeddings

@asset
def raw_documents():
    """Load documents from S3."""
    return documents

@asset
def chunked_documents(raw_documents):
    """Split into 512-token chunks with 50-token overlap."""
    from langchain.text_splitter import RecursiveCharacterTextSplitter
    splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=50)
    return splitter.split_documents(raw_documents)

@asset
def embedded_documents(chunked_documents):
    """Generate embeddings with Voyage AI."""
    embeddings = VoyageAIEmbeddings(model="voyage-3")
    return embeddings.embed_documents([doc.page_content for doc in chunked_documents])

参见： references/orchestration-tools.md，了解完整的 Dagster 模式和替代方案（Prefect、Airflow 3.0、dbt）

与前端技能集成

ai-chat 技能 → RAG 后端

ai-chat 技能使用 RAG 流水线的输出生成流式响应。

后端 API（FastAPI）：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

@app.post("/api/rag/stream")
async def stream_rag(query: str):
    async def generate():
        chain = RetrievalQA.from_chain_type(llm=OpenAI(streaming=True), retriever=vectorstore.as_retriever())
        async for chunk in chain.astream(query):
            yield chunk
    return StreamingResponse(generate(), media_type="text/plain")

参见： references/rag-architecture.md，了解完整的前端集成模式

search-filter 技能 → 语义搜索

search-filter 技能使用语义搜索后端进行向量相似度搜索。

后端（Qdrant + Voyage AI）：

from qdrant_client import QdrantClient
from langchain_voyageai import VoyageAIEmbeddings

@app.post("/api/search/semantic")
async def semantic_search(query: str, filters: dict):
    query_vector = VoyageAIEmbeddings(model="voyage-3").embed_query(query)
    results = QdrantClient().search(
        collection_name="documents",
        query_vector=query_vector,
        query_filter=filters,
        limit=10
    )
    return {"results": results}

数据版本控制

首选推荐：LakeFS（于 2025 年 11 月收购 DVC 团队）

在数据湖上执行类似 Git 的操作：分支、提交、合并、时间旅行。支持 S3/Azure/GCS。

import lakefs

branch = lakefs.Branch("main").create("experiment-voyage-3")
branch.commit("Updated embeddings to voyage-3")
branch.merge_into("main")

参见： references/data-versioning.md，了解完整的 LakeFS 设置

快速入门工作流

1. 设置向量数据库：

# Run Qdrant setup script (TOKEN-FREE execution)
python scripts/setup_qdrant.py --collection docs --dimension 1024

2. 对文档进行分块和嵌入：

# Chunk documents (TOKEN-FREE execution)
python scripts/chunk_documents.py \
  --input data/documents/ \
  --chunk-size 512 \
  --overlap 50 \
  --output data/chunks/

3. 实现 RAG 流水线：

参见 examples/langchain-rag/basic_rag.py，了解完整的可运行示例。

4. 使用 RAGAS 进行评估：

# Run evaluation (TOKEN-FREE execution)
python scripts/evaluate_rag.py \
  --dataset data/eval_qa.json \
  --output results/ragas_metrics.json

5. 使用编排工具部署：

参见 examples/dagster-pipelines/embedding_pipeline.py，了解生产环境部署。

依赖项

必需的 Python 软件包：

# Core RAG
pip install langchain langchain-core langchain-openai langchain-voyageai langchain-qdrant

# Vector database
pip install qdrant-client

# Evaluation
pip install ragas datasets

# Feature stores
pip install feast

# Orchestration
pip install dagster dagster-webserver

# Data versioning
pip install lakefs-client

替代方案所需的可选软件包：

# LlamaIndex (alternative to LangChain)
pip install llama-index

# dbt (SQL transformations)
pip install dbt-core dbt-postgres

# Prefect (alternative orchestration)
pip install prefect

故障排除

常见问题：

1. 检索质量差——检查分块大小（尝试 512 个令牌），增加重叠量（50–100），尝试混合搜索，使用 Cohere 重新排序

2. 嵌入生成缓慢——批量处理文档（100–1000），使用异步 API，通过 Redis 缓存，在开发环境中使用较小的模型

3. LLM 成本高——减少检索的分块数（k=3），使用成本更低的重新排序模型，缓存常见查询

参见： references/rag-architecture.md，了解完整的故障排除指南

最佳实践

**分块：**默认使用 512 个令牌，重叠 50 个令牌。对于复杂文档，使用语义分块。处理源代码时保留代码结构。

**嵌入：**生产环境使用 Voyage AI voyage-3，开发环境使用 OpenAI text-embedding-3-small。切勿混用嵌入模型（更换模型时要重新嵌入所有内容）。

**评估：**每次流水线更改后都运行 RAGAS 指标。维护包含 50 组以上问答对的测试数据集。持续跟踪指标。

**编排：**使用 Dagster 编排 ML/AI 流水线，仅使用 dbt 进行 SQL 转换。对所有流水线代码进行版本控制。

**前端集成：**始终以流式方式返回 LLM 响应。实现重试逻辑。向用户显示引用/来源。妥善处理空结果。

其他资源

参考文档：

• references/rag-architecture.md - 完整的 RAG 流水线指南
• references/chunking-strategies.md - 分块决策框架
• references/embedding-strategies.md - 嵌入模型对比
• references/langchain-patterns.md - LangChain 0.3+ 模式
• references/feature-stores.md - Feast 设置和替代方案
• references/evaluation-metrics.md - RAGAS 实施指南

可运行示例：

• examples/langchain-rag/basic_rag.py - 简单的 RAG 链
• examples/langchain-rag/streaming_rag.py - 流式响应
• examples/langchain-rag/hybrid_search.py - 向量 + BM25
• examples/llamaindex-agents/query_engine.py - LlamaIndex 替代方案
• examples/feast-features/ - 完整的特征存储设置
• examples/dagster-pipelines/embedding_pipeline.py - 生产环境流水线

可执行脚本（无需令牌）：

• scripts/evaluate_rag.py - RAGAS 评估运行器
• scripts/chunk_documents.py - 文档分块工具
• scripts/benchmark_retrieval.py - 检索质量基准测试
• scripts/setup_qdrant.py - Qdrant 集合设置