rag-skills

LlamaFarm 的 RAG 技能

针对 RAG 组件的框架专用模式和代码审查清单。

扩展：python-skills — 此处同样适用所有 Python 最佳实践。

组件概览

方面	技术	版本
Python	Python	3.11+
文档处理	LlamaIndex	0.13+
向量存储	ChromaDB	1.0+
任务队列	Celery	5.5+
嵌入	Universal/Ollama/OpenAI	多种

目录结构

rag/
├── api.py                 # Search and database APIs
├── celery_app.py          # Celery configuration
├── main.py                # Entry point
├── core/
│   ├── base.py            # Document, Component, Pipeline ABCs
│   ├── factories.py       # Component factories
│   ├── ingest_handler.py  # File ingestion with safety checks
│   ├── blob_processor.py  # Binary file processing
│   ├── settings.py        # Pydantic settings
│   └── logging.py         # RAGStructLogger
├── components/
│   ├── embedders/         # Embedding providers
│   ├── extractors/        # Metadata extractors
│   ├── parsers/           # Document parsers (LlamaIndex)
│   ├── retrievers/        # Retrieval strategies
│   └── stores/            # Vector stores (ChromaDB, FAISS)
├── tasks/                 # Celery tasks
│   ├── ingest_tasks.py    # File ingestion
│   ├── search_tasks.py    # Database search
│   ├── query_tasks.py     # Complex queries
│   ├── health_tasks.py    # Health checks
│   └── stats_tasks.py     # Statistics
└── utils/
    └── embedding_safety.py  # Circuit breaker, validation

快速参考

主题	文件	要点
LlamaIndex	llamaindex.md	文档解析、分块、节点转换
ChromaDB	chromadb.md	集合、嵌入、距离度量
Celery	celery.md	任务路由、错误处理、工作进程配置
性能	performance.md	批处理、缓存、去重

核心模式

文档数据类

from dataclasses import dataclass, field
from typing import Any

@dataclass
class Document:
    content: str
    metadata: dict[str, Any] = field(default_factory=dict)
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    source: str | None = None
    embeddings: list[float] | None = None

组件抽象基类

from abc import ABC, abstractmethod

class Component(ABC):
    def __init__(
        self,
        name: str | None = None,
        config: dict[str, Any] | None = None,
        project_dir: Path | None = None,
    ):
        self.name = name or self.__class__.__name__
        self.config = config or {}
        self.logger = RAGStructLogger(__name__).bind(name=self.name)
        self.project_dir = project_dir

    @abstractmethod
    def process(self, documents: list[Document]) -> ProcessingResult:
        pass

检索策略模式

class RetrievalStrategy(Component, ABC):
    @abstractmethod
    def retrieve(
        self,
        query_embedding: list[float],
        vector_store,
        top_k: int = 5,
        **kwargs
    ) -> RetrievalResult:
        pass

    @abstractmethod
    def supports_vector_store(self, vector_store_type: str) -> bool:
        pass

带熔断器的嵌入器

class Embedder(Component):
    DEFAULT_FAILURE_THRESHOLD = 5
    DEFAULT_RESET_TIMEOUT = 60.0

    def __init__(self, ...):
        super().__init__(...)
        self._circuit_breaker = CircuitBreaker(
            failure_threshold=config.get("failure_threshold", 5),
            reset_timeout=config.get("reset_timeout", 60.0),
        )
        self._fail_fast = config.get("fail_fast", True)

    def embed_text(self, text: str) -> list[float]:
        self.check_circuit_breaker()
        try:
            embedding = self._call_embedding_api(text)
            self.record_success()
            return embedding
        except Exception as e:
            self.record_failure(e)
            if self._fail_fast:
                raise EmbedderUnavailableError(str(e)) from e
            return [0.0] * self.get_embedding_dimension()

审查清单摘要

审查 RAG 代码时：

1. LlamaIndex（中优先级）

   • 合理配置分块
   • 解析过程中保留元数据
   • 处理不支持格式时的错误

2. ChromaDB（高优先级）

   • 线程安全地访问客户端
   • 正确选择距离度量
   • 兼容元数据类型

3. Celery（高优先级）

   • 将任务路由到正确的队列
   • 记录包含上下文的错误日志
   • 正确序列化

4. 性能（中优先级）

   • 对嵌入进行批处理
   • 启用去重
   • 合理缓存

有关包含 grep 模式的详细清单，请参阅各主题文件。