Pinecone Vector Database Setup Guide for Australian Businesses

Vector databases fundamentally change how we search and retrieve information. Unlike traditional databases that match keywords, vector databases find semantically similar content—understanding that "automobile" and "car" are related even without explicit keywords.

How Vector Search Works

Why Pinecone?

Common Use Cases

• RAG Systems: Ground LLM responses in your organisation's knowledge
• Semantic Search: Find documents by meaning, not just keywords
• Recommendations: Suggest similar products, articles, or content
• Duplicate Detection: Identify similar records or near-duplicates
• Anomaly Detection: Find outliers in high-dimensional data

Setting up Pinecone involves creating an account, configuring your first index, and understanding the key concepts that govern performance and cost.

Account Setup

1. Visit pinecone.io and create an account
2. Choose your plan (Starter free tier available)
3. Create your first project
4. Generate an API key from the console

Install the Python Client

# Install Pinecone client
pip install pinecone-client

# For latest features
pip install pinecone-client[grpc]

Initialize Connection

from pinecone import Pinecone

# Initialize client
pc = Pinecone(api_key="your-api-key")

# List existing indexes
print(pc.list_indexes())

Understanding Pinecone Concepts

Creating Your First Index

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="your-api-key")

# Create a serverless index (recommended for most use cases)
pc.create_index(
    name="australian-business-docs",
    dimension=1536,  # OpenAI text-embedding-3-small dimension
    metric="cosine",
    spec=ServerlessSpec(
        cloud="aws",
        region="us-east-1"  # Choose closest available region
    )
)

# Wait for index to be ready
import time
while not pc.describe_index("australian-business-docs").status['ready']:
    time.sleep(1)

# Connect to the index
index = pc.Index("australian-business-docs")

Before storing data in Pinecone, you need to convert it to vector embeddings. The quality of your embeddings directly impacts search relevance.

Choosing an Embedding Model

Generating Embeddings with OpenAI

from openai import OpenAI

client = OpenAI(api_key="your-openai-key")

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

# Example: Embed a business document
document = """
Clever Ops provides AI automation solutions for Australian businesses.
We specialise in workflow automation, custom AI development, and
enterprise integrations across Sydney, Melbourne, and Brisbane.
"""

embedding = get_embedding(document)
print(f"Embedding dimension: {len(embedding)}")  # 1536

Batch Ingestion Pipeline

from pinecone import Pinecone
from openai import OpenAI
import uuid

pc = Pinecone(api_key="pinecone-api-key")
openai_client = OpenAI(api_key="openai-api-key")
index = pc.Index("australian-business-docs")

def prepare_documents(documents: list[dict]) -> list[dict]:
    """Prepare documents for Pinecone ingestion."""
    vectors = []

    # Batch embeddings for efficiency
    texts = [doc["content"] for doc in documents]

    # OpenAI supports up to 2048 inputs per batch
    batch_size = 100
    all_embeddings = []

    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        response = openai_client.embeddings.create(
            input=batch,
            model="text-embedding-3-small"
        )
        all_embeddings.extend([e.embedding for e in response.data])

    # Prepare vectors with metadata
    for doc, embedding in zip(documents, all_embeddings):
        vectors.append({
            "id": doc.get("id", str(uuid.uuid4())),
            "values": embedding,
            "metadata": {
                "title": doc.get("title", ""),
                "category": doc.get("category", ""),
                "source": doc.get("source", ""),
                "date": doc.get("date", ""),
                "text": doc["content"][:1000]  # Store truncated text
            }
        })

    return vectors

def upsert_documents(documents: list[dict], namespace: str = ""):
    """Upload documents to Pinecone."""
    vectors = prepare_documents(documents)

    # Upsert in batches of 100
    batch_size = 100
    for i in range(0, len(vectors), batch_size):
        batch = vectors[i:i + batch_size]
        index.upsert(vectors=batch, namespace=namespace)

    print(f"Upserted {len(vectors)} vectors to namespace '{namespace}'")

Chunking Strategies

Long documents need to be split into chunks before embedding:

from langchain.text_splitter import RecursiveCharacterTextSplitter

def chunk_document(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    """Split document into overlapping chunks."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["

", "
", ". ", " ", ""]
    )
    return splitter.split_text(text)

# Optimal chunk sizes
# - FAQ/Support: 200-500 characters (precise answers)
# - Documentation: 500-1000 characters (balanced)
# - Long-form content: 1000-2000 characters (more context)

Effective querying is crucial for building responsive AI applications. Pinecone offers multiple query strategies to optimise for different use cases.

Basic Similarity Search

# Query with a text embedding
query_text = "How do I implement GST calculations for my business?"
query_embedding = get_embedding(query_text)

results = index.query(
    vector=query_embedding,
    top_k=5,
    include_metadata=True
)

for match in results.matches:
    print(f"Score: {match.score:.4f}")
    print(f"Title: {match.metadata.get('title')}")
    print(f"Text: {match.metadata.get('text')[:200]}...")
    print("---")

Filtered Queries (Hybrid Search)

Combine semantic search with metadata filtering:

# Filter by category and date
results = index.query(
    vector=query_embedding,
    top_k=10,
    include_metadata=True,
    filter={
        "category": {"$eq": "compliance"},
        "date": {"$gte": "2024-01-01"}
    }
)

# Complex filters with AND/OR
results = index.query(
    vector=query_embedding,
    top_k=10,
    filter={
        "$and": [
            {"category": {"$in": ["compliance", "legal"]}},
            {"$or": [
                {"region": {"$eq": "NSW"}},
                {"region": {"$eq": "VIC"}}
            ]}
        ]
    }
)

Namespace Queries

# Query specific namespace (e.g., per-tenant data)
results = index.query(
    vector=query_embedding,
    top_k=5,
    namespace="tenant_acme_corp"
)

# Query across all namespaces not directly supported
# Use separate queries and merge results if needed

Query Performance Optimisation

Async Queries for High Throughput

import asyncio
from pinecone import Pinecone

async def batch_query(queries: list[str], index) -> list[dict]:
    """Execute multiple queries concurrently."""

    async def single_query(query_text: str):
        embedding = get_embedding(query_text)
        return index.query(
            vector=embedding,
            top_k=5,
            include_metadata=True
        )

    tasks = [single_query(q) for q in queries]
    results = await asyncio.gather(*tasks)
    return results

# Run batch queries
queries = [
    "Australian tax compliance",
    "Privacy Act requirements",
    "APRA regulations"
]
results = asyncio.run(batch_query(queries, index))

Let's build a production-ready RAG system using Pinecone that can answer questions about Australian business regulations.

System Architecture

┌─────────────┐    ┌──────────────┐    ┌─────────────┐
│   User      │───▶│   Query      │───▶│  Pinecone   │
│   Query     │    │   Embedding  │    │   Search    │
└─────────────┘    └──────────────┘    └──────────────┘
                                              │
                                              ▼
┌─────────────┐    ┌──────────────┐    ┌─────────────┐
│   Response  │◀───│   GPT-4      │◀───│  Context    │
│   to User   │    │   Generation │    │  Assembly   │
└─────────────┘    └──────────────┘    └─────────────┘
          

Complete Implementation

from pinecone import Pinecone
from openai import OpenAI
from typing import Optional

class AustralianBusinessRAG:
    """RAG system for Australian business knowledge."""

    def __init__(
        self,
        pinecone_api_key: str,
        openai_api_key: str,
        index_name: str
    ):
        self.pc = Pinecone(api_key=pinecone_api_key)
        self.openai = OpenAI(api_key=openai_api_key)
        self.index = self.pc.Index(index_name)

    def get_embedding(self, text: str) -> list[float]:
        """Generate embedding for query."""
        response = self.openai.embeddings.create(
            input=text,
            model="text-embedding-3-small"
        )
        return response.data[0].embedding

    def retrieve(
        self,
        query: str,
        top_k: int = 5,
        filter_dict: Optional[dict] = None
    ) -> list[dict]:
        """Retrieve relevant documents."""
        query_embedding = self.get_embedding(query)

        results = self.index.query(
            vector=query_embedding,
            top_k=top_k,
            include_metadata=True,
            filter=filter_dict
        )

        return [
            {
                "text": match.metadata.get("text", ""),
                "title": match.metadata.get("title", ""),
                "source": match.metadata.get("source", ""),
                "score": match.score
            }
            for match in results.matches
        ]

    def generate_response(
        self,
        query: str,
        context_docs: list[dict]
    ) -> str:
        """Generate response using retrieved context."""

        # Format context
        context = "

".join([
            f"Source: {doc['title']}
{doc['text']}"
            for doc in context_docs
        ])

        # Create prompt
        system_prompt = """You are an expert on Australian business regulations and compliance.
Answer questions based on the provided context. Use Australian English spelling.
If the context doesn't contain relevant information, say so clearly.
Always cite your sources."""

        user_prompt = f"""Context:
{context}

Question: {query}

Answer:"""

        response = self.openai.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ],
            temperature=0.3
        )

        return response.choices[0].message.content

    def ask(
        self,
        query: str,
        top_k: int = 5,
        filter_dict: Optional[dict] = None
    ) -> dict:
        """Complete RAG pipeline."""

        # Retrieve relevant documents
        docs = self.retrieve(query, top_k, filter_dict)

        # Generate response
        response = self.generate_response(query, docs)

        return {
            "query": query,
            "response": response,
            "sources": [
                {"title": d["title"], "source": d["source"]}
                for d in docs
            ]
        }

# Usage
rag = AustralianBusinessRAG(
    pinecone_api_key="your-key",
    openai_api_key="your-key",
    index_name="australian-business-docs"
)

result = rag.ask(
    "What are the record-keeping requirements under the Privacy Act?",
    filter_dict={"category": "privacy"}
)

print(result["response"])
print("
Sources:", result["sources"])

Moving to production requires attention to reliability, monitoring, and cost management.

Error Handling

from pinecone.exceptions import PineconeException
import time

def robust_query(index, embedding, retries=3):
    """Query with retry logic."""
    for attempt in range(retries):
        try:
            return index.query(
                vector=embedding,
                top_k=5,
                include_metadata=True
            )
        except PineconeException as e:
            if attempt < retries - 1:
                time.sleep(2 ** attempt)  # Exponential backoff
                continue
            raise e

def robust_upsert(index, vectors, batch_size=100, retries=3):
    """Upsert with batching and retry logic."""
    for i in range(0, len(vectors), batch_size):
        batch = vectors[i:i + batch_size]
        for attempt in range(retries):
            try:
                index.upsert(vectors=batch)
                break
            except PineconeException as e:
                if attempt < retries - 1:
                    time.sleep(2 ** attempt)
                    continue
                raise e

Monitoring and Observability

import logging
import time
from functools import wraps

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def monitor_query(func):
    """Decorator to monitor query performance."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        try:
            result = func(*args, **kwargs)
            duration = time.time() - start
            logger.info(f"Query completed in {duration:.3f}s")
            return result
        except Exception as e:
            logger.error(f"Query failed: {e}")
            raise
    return wrapper

# Check index stats
def get_index_stats(index):
    """Get detailed index statistics."""
    stats = index.describe_index_stats()
    logger.info(f"Total vectors: {stats.total_vector_count}")
    logger.info(f"Namespaces: {list(stats.namespaces.keys())}")
    return stats

Cost Optimisation Strategies

Data Management

# Delete vectors by ID
index.delete(ids=["doc_1", "doc_2", "doc_3"])

# Delete by metadata filter
index.delete(
    filter={"date": {"$lt": "2023-01-01"}}
)

# Delete entire namespace
index.delete(delete_all=True, namespace="old_tenant")

# Update metadata without re-embedding
index.update(
    id="doc_1",
    set_metadata={"status": "archived", "reviewed": True}
)

A Brisbane-based professional services firm implemented Pinecone to power their internal knowledge management system, demonstrating practical vector database deployment for Australian enterprises.

# Namespace structure
namespaces = {
    "general": "Company-wide knowledge",
    "client_acme": "ACME Corp project docs",
    "client_bigbank": "BigBank engagement",
    "compliance": "Regulatory documents",
    "templates": "Proposal and report templates"
}

# Access control via namespace selection
def get_accessible_namespaces(user_role: str, user_clients: list) -> list:
    namespaces = ["general", "compliance", "templates"]
    if user_role in ["partner", "senior_manager"]:
        namespaces.extend([f"client_{c}" for c in user_clients])
    return namespaces