Migrate your vector embeddings from Pinecone to Antfly.

This example also covers the Antfly-side pattern for other systems that already have vectors, such as Milvus, Weaviate, pgvector exports, or custom offline embedding pipelines. Only the fetch step in main.py is Pinecone-specific.

Use Cases#

• Cost reduction: Move from cloud-based Pinecone to local Antfly
• Local development: Run your vector search locally without API calls
• Sunsetting Pinecone: Full migration with preserved embeddings

Prerequisites#

• A running Antfly server (antfly standalone)
• Python 3.11+
• Your Pinecone API key

pip install -r requirements.txt

Vector Compatibility#

Your source index stores embeddings at a specific dimension and uses a specific distance metric. To preserve search behavior in Antfly, create an external embeddings index with the same dimension and metric.

Check Your Pinecone Dimension#

In the Pinecone dashboard, find your index and note its dimension. Or via API:

from pinecone import Pinecone
pc = Pinecone(api_key="...")
stats = pc.Index("your-index").describe_index_stats()
print(f"Dimension: {stats['dimension']}")

What must match#

• dimension: Vector length (e.g. 768, 1024, 1536)
• distance_metric: Usually cosine, inner_product, or l2_squared

This example creates an Antfly embeddings index with:

• external: true
• the source vector dimension
• the configured distance_metric

No embedder, field, or template is used for this migration index, because the vectors are already computed upstream.

Schemaless Storage#

Antfly tables are schemaless. When migrating from Pinecone:

• Each vector's metadata becomes top-level document fields
• The id becomes the document key
• Pre-computed embeddings go in the _embeddings field for an external: true index

Example:

Pinecone:                          Antfly:
  id: "doc-123"                      _id: "doc-123"
  values: [0.1, 0.2, ...]    →       text: "Hello"
  metadata: {text: "Hello"}          _embeddings: {nomic_index: [0.1, ...]}

Configuration#

Update these values in main.py for your migration:

# --- Configuration ---
# Customize these values for your migration
PINECONE_INDEX_NAME = "my-pinecone-index"  # Your Pinecone index name
TABLE_NAME = PINECONE_INDEX_NAME  # Antfly table name (using same name)
INDEX_NAME = "nomic_index"  # Name for the vector index in Antfly
DISTANCE_METRIC = "cosine"  # Match the metric used by your source index

ANTFLY_BASE_URL = os.getenv("ANTFLY_BASE_URL", "http://localhost:8080/db/v1")

Running the Migration#

export PINECONE_API_KEY="your-key-here"
python main.py

How It Works#

1. Initialize Clients#

def main():
    load_dotenv()

    # --- Initialize Pinecone ---
    pinecone_api_key = os.getenv("PINECONE_API_KEY")
    if not pinecone_api_key:
        print("FATAL: PINECONE_API_KEY not found in environment.", file=sys.stderr)
        sys.exit(1)

    pc = Pinecone(api_key=pinecone_api_key)
    pinecone_index = pc.Index(PINECONE_INDEX_NAME)
    print(f"Connected to Pinecone index '{PINECONE_INDEX_NAME}'.")

    # --- Initialize Antfly ---
    client = AntflyClient(base_url=ANTFLY_BASE_URL)
    try:
        print(f"Connecting to Antfly at {ANTFLY_BASE_URL}...")
        client.list_tables()
        print("Connected to Antfly.")
    except Exception as e:
        print(f"Could not connect to Antfly: {e}")
        print("Ensure the Antfly server is running.")
        sys.exit(1)

2. Create Table#

Tables are schemaless - no schema definition needed:

    # --- Create Antfly table ---
    # Tables are schemaless - no schema definition required
    try:
        client.drop_table(TABLE_NAME)
        print(f"Deleted existing table '{TABLE_NAME}'")
    except AntflyException:
        pass

    try:
        client.create_table(name=TABLE_NAME)
        print(f"Created table '{TABLE_NAME}'")
    except Exception as e:
        print(f"Error creating table: {e}")
        sys.exit(1)

    print("Waiting for table shards to initialize...")
    time.sleep(5)

3. Create Vector Index#

The index is created before inserting data so Antfly knows where to store imported vectors. This is an external index, meaning Antfly will accept vectors written through _embeddings and will not try to generate them from document fields:

    # --- Create vector index BEFORE inserting data ---
    # This tells Antfly where to store pre-computed embeddings
    try:
        print(f"\nCreating vector index '{INDEX_NAME}' with dimension {dimension}...")
        index_config = {
            "name": INDEX_NAME,
            "type": "embeddings",
            "external": True,
            "dimension": dimension,
            "distance_metric": DISTANCE_METRIC,
        }
        print(
            f"  -> Creating external index for imported vectors (metric: {DISTANCE_METRIC})"
        )

        resp = httpx.post(
            f"{ANTFLY_BASE_URL}/tables/{TABLE_NAME}/indexes/{INDEX_NAME}",
            json=index_config,
            timeout=30.0,
        )
        if resp.status_code in (200, 201):
            print("Index created successfully.")
        else:
            print(f"Index creation response: {resp.status_code} - {resp.text}")
    except Exception as e:
        print(f"Error creating index: {e}")

    print("Waiting for index to propagate...")
    time.sleep(5)

4. Fetch Vectors from Pinecone#

Uses a zero-vector query with high top_k - a common pattern since Pinecone doesn't have a "list all" API:

    def fetch_all_vectors(pc_index, namespace: str, dim: int):
        """
        Fetch all vectors from a Pinecone namespace.

        Uses a zero-vector query with high top_k - a common pattern
        since Pinecone doesn't have a direct "list all" API.
        """
        print(f"Fetching vectors from namespace: '{namespace}'...")

        # Query with zero vector to get all vectors
        query_response = pc_index.query(
            namespace=namespace,
            vector=[0.0] * dim,
            top_k=1000,
            include_values=True,
            include_metadata=True,
        )

        vectors = query_response.get("matches", [])
        if len(vectors) == 1000:
            print("  WARNING: Hit 1000 limit - may have more vectors.", file=sys.stderr)

        print(f"  -> Found {len(vectors)} vectors.")
        return vectors

    # Fetch from all namespaces
    all_vectors = {}
    for namespace in namespaces:
        vectors = fetch_all_vectors(pinecone_index, namespace, dimension)

        if not vectors:
            continue

        all_vectors[namespace] = []
        for v in vectors:
            # Flatten: metadata becomes top-level fields
            entry = {
                "id": v["id"],
                "values": v["values"],
                **v.get("metadata", {}),
                "namespace": namespace,
            }
            all_vectors[namespace].append(entry)

5. Upsert to Antfly#

Vectors are batch-upserted with the _embeddings field:

    # --- Upsert into Antfly ---
    for namespace, items in all_vectors.items():
        try:
            print(f"Upserting {len(items)} vectors for namespace '{namespace}'...")

            inserts = {}
            for item in items:
                key = item["id"]
                properties = {k: v for k, v in item.items() if k != "id"}

                # Use _embeddings format for pre-computed embeddings
                # Format: {"_embeddings": {"<index_name>": [vector values]}}
                if "values" in properties:
                    embedding_vector = properties.pop("values")
                    properties["_embeddings"] = {INDEX_NAME: embedding_vector}

                inserts[key] = properties

            resp = httpx.post(
                f"{ANTFLY_BASE_URL}/tables/{TABLE_NAME}/batch",
                json={"inserts": inserts},
                timeout=60.0,
            )
            if resp.status_code in (200, 201):
                print(f"  -> Successfully upserted {len(items)} vectors.")
            else:
                print(f"  -> Batch failed: {resp.status_code} - {resp.text}")

        except Exception as e:
            print(f"  -> ERROR: {e}")
            break

    print("\nMigration complete. Waiting for indexing...")
    time.sleep(10)

6. Verify#

    # --- Verify migration ---
    print("\n=== Verifying Migration ===")

    try:
        resp = httpx.post(
            f"{ANTFLY_BASE_URL}/tables/{TABLE_NAME}/query",
            json={"limit": 200},
            timeout=30.0,
        )
        if resp.status_code == 200:
            data = resp.json()
            hits = data.get("responses", [{}])[0].get("hits", {}).get("hits", [])
            print(f"Total records in Antfly: {len(hits)}")

            if hits:
                print("\nSample records:")
                for hit in hits[:3]:
                    key = hit.get("_id", "unknown")
                    source = hit.get("_source", {})
                    text = source.get("text", "")[:80] if source else ""
                    print(f"  - {key}: {text}...")
    except Exception as e:
        print(f"Verification failed: {e}")

    # --- Test vector search using one of the imported vectors ---
    sample_key = None
    sample_vector = None
    for items in all_vectors.values():
        if items:
            sample_key = items[0]["id"]
            sample_vector = items[0]["values"]
            break

    if sample_vector is not None:
        print(
            f"\nTesting vector search using imported vector from key '{sample_key}'..."
        )
        try:
            resp = httpx.post(
                f"{ANTFLY_BASE_URL}/tables/{TABLE_NAME}/query",
                json={"embeddings": {INDEX_NAME: sample_vector}, "limit": 3},
                timeout=30.0,
            )
            if resp.status_code == 200:
                data = resp.json()
                hits = data.get("responses", [{}])[0].get("hits", {}).get("hits", [])
                if hits:
                    print("Results:")
                    for hit in hits:
                        score = hit.get("_score", 0)
                        key = hit.get("_id", "unknown")
                        print(f"  [{score:.4f}] {key}")
                else:
                    print("  (no results)")
            else:
                print(f"  Search failed: {resp.status_code} - {resp.text}")
        except Exception as e:
            print(f"  Search failed: {e}")
    else:
        print("\nSkipping verification search because no vectors were imported.")

    print("\n=== Done ===")
    print(f"Table: {TABLE_NAME}")
    print(f"Index: {INDEX_NAME} (dimension: {dimension}, metric: {DISTANCE_METRIC})")

The verification query uses one of the imported vectors directly via the query embeddings field. This keeps the example consistent with external: true indexes, which do not have an embedder for text-to-vector conversion.

Troubleshooting#

"manual _embeddings are not allowed"#

The Antfly index was probably created as a managed index. For imported vectors, create the index with external: true and do not set field, template, or embedder.

"Hit 1000 limit"#

Increase top_k in fetch_all_vectors() or implement pagination for larger datasets.

Adapting this example for Milvus / Weaviate / pgvector#

Reuse the same Antfly-side steps:

• create the table
• create an external: true embeddings index
• write vectors via _embeddings
• verify with a query-time embeddings vector

Only the export/fetch step changes based on your source system.