Building an AI-Powered Product Recommendation System

import json
import logging
from llama_index.core import Document, VectorStoreIndex
from llama_index.core.vector_stores.types import MetadataFilters, ExactMatchFilter

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

def get_documents_from_json(products: list) -> list[Document]:
    documents = []
    for item in products:
        species_list = []
        if hasattr(item, 'species') and item.species:
            species_list = [s.lower().strip() for s in item.species.split(',') if s.strip()]
        
        if not species_list:
            logger.warning(f"Product ID {getattr(item, 'id', 'Unknown')} has no valid species. Skipping.")
            continue

        for single_species in species_list:
            enhanced_text = f"""
Product ID: {item.id}
Name: {item.name}
Species: {single_species}
Weight Range: {getattr(item, 'min_weight', None) or 'No minimum'} - {getattr(item, 'max_weight', None) or 'No maximum'} kg
Description: {getattr(item, 'description', 'No description available')}

This product is specifically designed for {single_species} patients.
"""
            doc_id = f"{item.id}_{single_species}"
            original_data_json = json.dumps(item.json() if hasattr(item, 'json') else item.__dict__)

            doc = Document(
                id_=doc_id,
                text=enhanced_text + "\n\nOriginal Data: " + original_data_json,
                metadata={
                    "id": item.id,
                    "name": item.name,
                    "species": single_species,
                    "min_weight": item.min_weight,
                    "max_weight": item.max_weight,
                    "description": item.description
                }
            )
            documents.append(doc)
    
    logger.info(f"Created {len(documents)} product documents for vector indexing")
    return documents

def load_products_from_request() -> VectorStoreIndex:
    """Initialize vector store from external API"""
    try:
        documents = get_products()
        logger.info("Building VectorStoreIndex from documents")
        return VectorStoreIndex.from_documents(documents)
    except Exception as e:
        logger.error(f"Error creating vector index: {e}")
        raise

def create_filtered_query_engine(index: VectorStoreIndex, patient_info):
    """Create a species-filtered query engine with optimized parameters"""
    
    species_lower = None
    if hasattr(patient_info, 'species') and patient_info.species:
        species_lower = patient_info.species.lower()
    
    if species_lower:
        filters = MetadataFilters(filters=[
            ExactMatchFilter(key="species", value=species_lower)
        ])
        query_engine = index.as_query_engine(
            verbose=True,
            similarity_top_k=15,
            filters=filters,
            alpha=0.5,
            response_mode="tree_summarize",
            streaming=False
        )
    else:
        query_engine = index.as_query_engine(
            verbose=True,
            similarity_top_k=5,
            response_mode="compact"
        )
    
    return query_engine

async def _refresh_products_core_logic(app_state, products_to_load: list):
    """Core logic to dynamically update the vector index with new product data"""
    logger = logging.getLogger(__name__)
    try:
        logger.info(f"Refreshing index with {len(products_to_load)} products")
        
        new_index = load_products_from_array(products_to_load)
        
        app_state.vector_store_index = new_index
        
        logger.info("Vector index successfully refreshed")
        return {"status": "success", "product_count": len(products_to_load)}
    except Exception as e:
        logger.error(f"Error refreshing vector index: {e}")
        return {"status": "error", "message": str(e)}

# Illustrative example of document structure:
# Assuming Document is from llama_index.core
# Instead of one document per product like this:
# Document(text="Product X for dogs, cats, birds", metadata={"id": "X"})
# We create separate, more specific documents:
# Document(id_="X_dog", text="Product X for dogs…", metadata={"id": "X", "species": "dog", …})
# Document(id_="X_cat", text="Product X for cats…", metadata={"id": "X", "species": "cat", …})
# Document(id_="X_bird", text="Product X for birds…", metadata={"id": "X", "species": "bird", …})

def query_index(prompt_template_str: str, 
                patient_info,
                prescription_info,
                index: VectorStoreIndex) -> str:
    """Execute the full query pipeline with LlamaIndex"""
    
    species = None
    if hasattr(patient_info, 'species') and patient_info.species:
        species = patient_info.species.lower()
    
    conditions = []
    if hasattr(patient_info, 'existing_conditions') and patient_info.existing_conditions:
        conditions = [c.strip() for c in patient_info.existing_conditions.split(',')]
    
    user_prompt_text = build_user_prompt(patient_info, prescription_info)
    if conditions:
        condition_text = ", ".join(conditions)
        user_prompt_text += f"\nPatient conditions: {condition_text}"

    query_engine_params = {"similarity_top_k": 5}
    if species:
        filters = MetadataFilters(filters=[ExactMatchFilter(key="species", value=species)])
        query_engine_params.update({
            "similarity_top_k": 15,
            "filters": filters,
            "alpha": 0.5
        })
    
    query_engine = index.as_query_engine(**query_engine_params)
    
    active_prompt = PromptTemplate(prompt_template_str)
    final_query_prompt = active_prompt.format(context_str="", query_str=user_prompt_text)
    
    logger.info(f"Querying index with prompt: {final_query_prompt[:200]}...")
    result = query_engine.query(final_query_prompt)
    
    return process_llm_response(str(result))

from llama_index.core import VectorStoreIndex
from llama_index.core.vector_stores.types import MetadataFilters, ExactMatchFilter

# Exact species matching prevents cross-species recommendations
species_filter = ExactMatchFilter(key="species", value="dog")
dog_specific_engine = index.as_query_engine(
    filters=MetadataFilters(filters=[species_filter])
)

# Multiple filters can be combined
filters = MetadataFilters(filters=[
    ExactMatchFilter(key="species", value="dog"),
])

# Filters are applied BEFORE semantic search, improving both speed and accuracy
query_engine = index.as_query_engine(filters=filters)

import re
import json
import logging

logger = logging.getLogger(__name__)

def clean_json(content: str) -> str:
    """Multi-stage JSON cleaning for LLM responses."""
    if content.startswith("```json"):
        content = content[len("```json"):]
    if content.endswith("```"):
        content = content[:-len("```")]
    content = content.strip()

    cleaned_content = re.sub(r',\s*([}\]])', r'\1', content)

    try:
        json_obj = json.loads(cleaned_content)
        return json.dumps(json_obj)
    except json.JSONDecodeError as e:
        logger.warning(f"Initial JSON parsing failed after trailing comma removal: {e}. Attempting aggressive cleaning.")
        
        fixed_content = cleaned_content
        
        fixed_content = re.sub(r'\}\s*$', '}', fixed_content)

        fixed_content = re.sub(r'([{,]\s*)([a-zA-Z_]\w*)(\s*:)', r'\1"\2"\3', fixed_content)
        
        fixed_content = fixed_content.replace("'", '"')
        
        try:
            json_obj = json.loads(fixed_content)
            return json.dumps(json_obj)
        except json.JSONDecodeError as e2:
            logger.error(f"Aggressive JSON cleaning also failed: {e2}. Original content snippet: {content[:200]}")
            return json.dumps({"error": "Failed to parse LLM response as JSON after multiple attempts.", "details": str(e2)})