Skip to content

teff.rag.stores.factory

teff.rag.stores.factory

Build a :class:VectorStore from a declarative store: config dict.

Shared by the rag (search) and rag_ingest (write) tools so both read the same {type, ...} block in a workflow YAML.

Functions:

Name Description
store_from_config

Instantiate the store named by config["type"].

store_from_config 

store_from_config(config)

Instantiate the store named by config["type"].

Supported types: in_memory (default), sqlite, chroma, qdrant, pgvector, faiss, lance/lancedb, milvus, weaviate, pinecone. Raises ValueError for unknown types.

Source code in teff/rag/stores/factory.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def store_from_config(config: dict) -> VectorStore:
    """Instantiate the store named by ``config["type"]``.

    Supported types: ``in_memory`` (default), ``sqlite``, ``chroma``,
    ``qdrant``, ``pgvector``, ``faiss``, ``lance``/``lancedb``, ``milvus``,
    ``weaviate``, ``pinecone``.  Raises ``ValueError`` for unknown types.
    """
    store_type = config.get("type", "in_memory")
    if store_type == "in_memory":
        from teff.rag.stores import InMemoryVectorStore

        return InMemoryVectorStore(dim=config.get("dim", 768))
    if store_type == "sqlite":
        from teff.rag.stores import SQLiteVectorStore

        return SQLiteVectorStore(
            path=config.get("path", "./vectors.db"),
            dim=config.get("dim"),
        )
    if store_type == "chroma":
        from teff.rag.stores import ChromaVectorStore

        return ChromaVectorStore(
            path=config.get("path", "./chroma"),
            collection=config.get("collection", "teff"),
        )
    if store_type == "qdrant":
        from teff.rag.stores import QdrantVectorStore

        return QdrantVectorStore(
            host=config.get("host", "localhost"),
            port=config.get("port", 6333),
            collection=config.get("collection", "teff"),
        )
    if store_type == "pgvector":
        from teff.rag.stores import PGVectorStore

        return PGVectorStore(
            dsn=config.get("dsn", ""),
            table=config.get("table", "teff_vectors"),
        )
    if store_type == "faiss":
        from teff.rag.stores import FAISSVectorStore

        return FAISSVectorStore(
            dim=config.get("dim", 1536),
            path=config.get("path"),
        )
    if store_type in ("lance", "lancedb"):
        from teff.rag.stores import LanceVectorStore

        return LanceVectorStore(
            path=config.get("path", "./lance"),
            table=config.get("table", "vectors"),
            dim=config.get("dim"),
        )
    if store_type == "milvus":
        from teff.rag.stores import MilvusVectorStore

        return MilvusVectorStore(
            uri=config.get("uri", "./milvus.db"),
            token=config.get("token", ""),
            collection=config.get("collection", "teff"),
            dim=config.get("dim"),
        )
    if store_type == "weaviate":
        from teff.rag.stores import WeaviateVectorStore

        return WeaviateVectorStore(
            collection=config.get("collection", "teff"),
            embedded=bool(config.get("embedded", False)),
            host=config.get("host", "localhost"),
            http_port=config.get("http_port", 8080),
            http_secure=bool(config.get("http_secure", False)),
            grpc_port=config.get("grpc_port", 50051),
            grpc_secure=bool(config.get("grpc_secure", False)),
            api_key=config.get("api_key", ""),
            headers=config.get("headers"),
            dim=config.get("dim"),
        )
    if store_type == "pinecone":
        from teff.rag.stores import PineconeVectorStore

        return PineconeVectorStore(
            index_name=config.get("index_name", "teff"),
            api_key=config.get("api_key", ""),
            host=config.get("host", ""),
            namespace=config.get("namespace", ""),
            dim=config.get("dim"),
        )
    msg = f"unsupported store type: {store_type}"
    raise ValueError(msg)