Introduction To Vector Databases And How To Use AI For SEO via @sejournal, @vahandev

pip install openai google-cloud-aiplatform google-auth pandas pinecone-client tabulate ipython numpy

import pandas as pd
from openai import OpenAI
from pinecone import Pinecone
from IPython.display import clear_output # Setup your OpenAI and Pinecone API keys
openai_client = OpenAI(api_key='YOUR_OPENAI_API_KEY') # Instantiate OpenAI client
pinecone = Pinecone(api_key='YOUR_PINECON_API_KEY') # Connect to an existing Pinecone index
index_name = "article-index-all-ada"
index = pinecone.Index(index_name) def generate_embeddings(text): """ Generates an embedding for the given text using OpenAI's API. Returns None if text is invalid or an error occurs. """ try: if not text or not isinstance(text, str): raise ValueError("Input text must be a non-empty string.") result = openai_client.embeddings.create( input=text, model="text-embedding-ada-002" ) clear_output(wait=True) # Clear output for a fresh display if hasattr(result, 'data') and len(result.data) > 0: print("API Response:", result) return result.data[0].embedding else: raise ValueError("Invalid response from the OpenAI API. No data returned.") except ValueError as ve: print(f"ValueError: {ve}") return None except Exception as e: print(f"An error occurred while generating embeddings: {e}") return None # Load your articles from a CSV
df = pd.read_csv('Sample Export File.csv') # Process each article
for idx, row in df.iterrows(): try: clear_output(wait=True) content = row["Content"] vector = generate_embeddings(content) if vector is None: print(f"Skipping article ID {row['ID']} due to empty or invalid embedding.") continue index.upsert(vectors=[ ( row['Permalink'], # Unique ID vector, # The embedding { 'title': row['Title'], 'category': row['Category'], 'type': row['Type'], 'publish_date': row['Publish Date'], 'publish_year': row['Publish Year'] } ) ]) except Exception as e: clear_output(wait=True) print(f"Error processing article ID {row['ID']}: {str(e)}") print("Embeddings are successfully stored in the vector database.")

from openai import OpenAI
from pinecone import Pinecone
from IPython.display import clear_output
from tabulate import tabulate # Import tabulate for table formatting # Setup your OpenAI and Pinecone API keys
openai_client = OpenAI(api_key='YOUR_OPENAI_API_KEY') # Instantiate OpenAI client
pinecone = Pinecone(api_key='YOUR_OPENAI_API_KEY') # Connect to an existing Pinecone index
index_name = "article-index-all-ada"
index = pinecone.Index(index_name) # Function to generate embeddings using OpenAI's API
def generate_embeddings(text): """ Generates an embedding for a given text using OpenAI's API. """ try: if not text or not isinstance(text, str): raise ValueError("Input text must be a non-empty string.") result = openai_client.embeddings.create( input=text, model="text-embedding-ada-002" ) # Debugging: Print the response to understand its structure clear_output(wait=True) #print("API Response:", result) if hasattr(result, 'data') and len(result.data) > 0: return result.data[0].embedding else: raise ValueError("Invalid response from the OpenAI API. No data returned.") except ValueError as ve: print(f"ValueError: {ve}") return None except Exception as e: print(f"An error occurred while generating embeddings: {e}") return None # Function to query the Pinecone index with keywords and metadata
def match_keywords_to_index(keywords): """ Matches a list of keywords to the closest article in the Pinecone index, filtering by metadata dynamically. """ results = [] for keyword_pair in keywords: try: clear_output(wait=True) # Extract the keyword and category from the sub-array keyword = keyword_pair[0] category = keyword_pair[1] # Generate embedding for the current keyword vector = generate_embeddings(keyword) if vector is None: print(f"Skipping keyword '{keyword}' due to embedding error.") continue # Query the Pinecone index for the closest vector with metadata filter query_results = index.query( vector=vector, # The embedding of the keyword top_k=1, # Retrieve only the closest match include_metadata=True, # Include metadata in the results filter={"category": category} # Filter results by metadata category dynamically ) # Store the closest match if query_results['matches']: closest_match = query_results['matches'][0] results.append({ 'Keyword': keyword, # The searched keyword 'Category': category, # The category used for filtering 'Match Score': f"{closest_match['score']:.2f}", # Similarity score (formatted to 2 decimal places) 'Title': closest_match['metadata'].get('title', 'N/A'), # Title of the article 'URL': closest_match['id'] # Using 'id' as the URL }) else: results.append({ 'Keyword': keyword, 'Category': category, 'Match Score': 'N/A', 'Title': 'No match found', 'URL': 'N/A' }) except Exception as e: clear_output(wait=True) print(f"Error processing keyword '{keyword}' with category '{category}': {e}") results.append({ 'Keyword': keyword, 'Category': category, 'Match Score': 'Error', 'Title': 'Error occurred', 'URL': 'N/A' }) return results # Example usage: Find matches for an array of keywords and categories
keywords = [["SEO Tools", "SEO"], ["TikTok", "TikTok"], ["SEO Consultant", "SEO"]] # Replace with your keywords and categories
matches = match_keywords_to_index(keywords) # Display the results in a table
print(tabulate(matches, headers="keys", tablefmt="fancy_grid"))

import os
import sys
import time
import numpy as np
import pandas as pd
from typing import List, Optional from google.auth import load_credentials_from_file
from google.cloud import aiplatform
from google.api_core.exceptions import ServiceUnavailable from pinecone import Pinecone
from vertexai.language_models import TextEmbeddingModel, TextEmbeddingInput # Set up your Google Cloud credentials
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "config.json" # Replace with your JSON key file
credentials, project_id = load_credentials_from_file(os.environ["GOOGLE_APPLICATION_CREDENTIALS"]) # Initialize Pinecone
pinecone = Pinecone(api_key='YOUR_PINECON_API_KEY') # Replace with your Pinecone API key
index = pinecone.Index("article-index-vertex") # Replace with your Pinecone index name # Initialize Vertex AI
aiplatform.init(project=project_id, credentials=credentials, location="us-central1") def generate_embeddings( text: str, task: str = "RETRIEVAL_DOCUMENT", model_id: str = "text-embedding-005", dimensions: Optional[int] = 768
) -> Optional[List[float]]: if not text or not text.strip(): print("Text input is empty. Skipping.") return None try: model = TextEmbeddingModel.from_pretrained(model_id) input_data = TextEmbeddingInput(text, task_type=task) vectors = model.get_embeddings([input_data], output_dimensionality=dimensions) return vectors[0].values except ServiceUnavailable as e: print(f"Vertex AI service is unavailable: {e}") return None except Exception as e: print(f"Error generating embeddings: {e}") return None # Load data from CSV
data = pd.read_csv("Sample Export File.csv") # Replace with your CSV file path for idx, row in data.iterrows(): try: permalink = str(row["Permalink"]) content = row["Content"] embedding = generate_embeddings(content) if not embedding: print(f"Skipping article ID {row['ID']} due to empty or failed embedding.") continue print(f"Embedding for {permalink}: {embedding[:5]}...") sys.stdout.flush() index.upsert(vectors=[ ( permalink, embedding, { 'category': row['Category'], 'title': row['Title'], 'publish_date': row['Publish Date'], 'type': row['Type'], 'publish_year': row['Publish Year'] } ) ]) time.sleep(1) # Optional: Sleep to avoid rate limits except Exception as e: print(f"Error processing article ID {row['ID']}: {e}") print("All embeddings are stored in the vector database.") 

import os
import pandas as pd
from google.cloud import aiplatform
from google.auth import load_credentials_from_file
from google.api_core.exceptions import ServiceUnavailable
from vertexai.language_models import TextEmbeddingModel from pinecone import Pinecone
from tabulate import tabulate # For table formatting # Set up your Google Cloud credentials
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "config.json" # Replace with your JSON key file
credentials, project_id = load_credentials_from_file(os.environ["GOOGLE_APPLICATION_CREDENTIALS"]) # Initialize Pinecone client
pinecone = Pinecone(api_key='YOUR_PINECON_API_KEY') # Add your Pinecone API key
index_name = "article-index-vertex" # Replace with your Pinecone index name
index = pinecone.Index(index_name) # Initialize Vertex AI
aiplatform.init(project=project_id, credentials=credentials, location="us-central1") def generate_embeddings( text: str, model_id: str = "text-embedding-005"
) -> list: """ Generates embeddings for the input text using Google Vertex AI's embedding model. Returns None if text is empty or an error occurs. """ if not text or not text.strip(): print("Text input is empty. Skipping.") return None try: model = TextEmbeddingModel.from_pretrained(model_id) vector = model.get_embeddings([text]) # Removed 'task_type' and 'output_dimensionality' return vector[0].values except ServiceUnavailable as e: print(f"Vertex AI service is unavailable: {e}") return None except Exception as e: print(f"Error generating embeddings: {e}") return None def match_keywords_to_index(keywords): """ Matches a list of keyword-category pairs to the closest articles in the Pinecone index, filtering by metadata if specified. """ results = [] for keyword_pair in keywords: keyword = keyword_pair[0] category = keyword_pair[1] try: keyword_vector = generate_embeddings(keyword) if not keyword_vector: print(f"No embedding generated for keyword '{keyword}' in category '{category}'.") results.append({ 'Keyword': keyword, 'Category': category, 'Match Score': 'Error/Empty', 'Title': 'No match', 'URL': 'N/A' }) continue query_results = index.query( vector=keyword_vector, top_k=1, include_metadata=True, filter={"category": category} ) if query_results['matches']: closest_match = query_results['matches'][0] results.append({ 'Keyword': keyword, 'Category': category, 'Match Score': f"{closest_match['score']:.2f}", 'Title': closest_match['metadata'].get('title', 'N/A'), 'URL': closest_match['id'] }) else: results.append({ 'Keyword': keyword, 'Category': category, 'Match Score': 'N/A', 'Title': 'No match found', 'URL': 'N/A' }) except Exception as e: print(f"Error processing keyword '{keyword}' with category '{category}': {e}") results.append({ 'Keyword': keyword, 'Category': category, 'Match Score': 'Error', 'Title': 'Error occurred', 'URL': 'N/A' }) return results # Example usage: keywords = [["SEO Tools", "Tools"], ["TikTok", "TikTok"], ["SEO Consultant", "SEO"]] matches = match_keywords_to_index(keywords) # Display the results in a table
print(tabulate(matches, headers="keys", tablefmt="fancy_grid"))