Introduction aux bases de données vectorielles et à l’utilisation de l’IA pour le référencement

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"))