📝 Session 4: HyDE Implementation Guide¶

def _classify_query_type(self, query: str) -> str:
    """Classify query type for appropriate HyDE template selection."""

    classification_prompt = f"""
    Classify the following query into one of these categories:
    1. factual - Questions seeking specific facts or information
    2. procedural - Questions asking how to do something
    3. analytical - Questions requiring analysis or comparison
    4. creative - Questions requiring creative responses

    Query: {query}

    Return only the category name:
    """

    try:
        response = self.llm_model.predict(classification_prompt).strip().lower()
        if response in self.hyde_templates:
            return response
    except Exception as e:
        print(f"Classification error: {e}")

    # Default to factual
    return 'factual'

def _factual_hyde_template(self, query: str) -> str:
    """Template for factual query types."""
    return f"""
    Write a detailed, informative document that answers: {query}

    The document should:
    - Provide specific facts and data points
    - Include relevant context and background
    - Use authoritative tone and precise language
    - Cover multiple aspects of the topic
    - Include examples where relevant

    Document:
    """

def _procedural_hyde_template(self, query: str) -> str:
    """Template for procedural query types."""
    return f"""
    Write a detailed how-to guide that explains: {query}

    The guide should:
    - Provide step-by-step instructions
    - Include prerequisite requirements
    - Mention common pitfalls and solutions
    - Explain the reasoning behind each step
    - Include tips for success

    Guide:
    """

def _analytical_hyde_template(self, query: str) -> str:
    """Template for analytical query types."""
    return f"""
    Write an analytical document that examines: {query}

    The analysis should:
    - Present multiple perspectives or approaches
    - Compare and contrast different options
    - Discuss pros and cons
    - Provide evidence-based reasoning
    - Draw insightful conclusions

    Analysis:
    """

def _generate_hypothetical_documents(self, query: str,
                                   query_type: str,
                                   num_docs: int) -> List[Dict]:
    """Generate multiple hypothetical documents with variations."""

    base_template = self.hyde_templates[query_type]
    hypothetical_docs = []

    for i in range(num_docs):
        # Add variation to temperature for diversity
        varied_temperature = self.temperature + (i * 0.1)
        varied_temperature = min(varied_temperature, 1.0)

        try:
            # Generate hypothetical document
            prompt = base_template(query)

            # Use varied temperature for diversity
            doc_text = self.llm_model.predict(
                prompt,
                temperature=varied_temperature
            )

            hypothetical_docs.append({
                'document': doc_text.strip(),
                'temperature': varied_temperature,
                'variation': i + 1,
                'word_count': len(doc_text.split()),
                'quality_score': self._assess_document_quality(doc_text, query)
            })

            print(f"Generated hypothetical document {i+1}/{num_docs}")

        except Exception as e:
            print(f"Error generating document {i+1}: {e}")
            continue

    # Sort by quality score for best results
    hypothetical_docs.sort(key=lambda x: x['quality_score'], reverse=True)

    return hypothetical_docs

def _assess_document_quality(self, document: str, query: str) -> float:
    """Assess the quality of a generated hypothetical document."""

    quality_prompt = f"""
    Rate the quality of this hypothetical document for answering the query.
    Consider relevance, completeness, and accuracy.

    Query: {query}
    Document: {document[:500]}...

    Rate from 0.0 to 1.0 where:
    1.0 = Excellent, comprehensive answer
    0.5 = Adequate, partial answer
    0.0 = Poor, irrelevant answer

    Return only a number:
    """

    try:
        response = self.llm_model.predict(quality_prompt).strip()
        quality_score = float(response)
        return max(0.0, min(1.0, quality_score))
    except:
        return 0.5  # Default score if assessment fails

def _create_enhanced_embeddings(self, query: str,
                              hypothetical_docs: List[Dict]) -> Dict:
    """Create enhanced embeddings combining query and hypothetical docs."""

    # Original query embedding
    original_embedding = self.embedding_model.encode([query])[0]

    # Hypothetical document embeddings
    hyde_texts = [doc['document'] for doc in hypothetical_docs]
    hyde_embeddings = self.embedding_model.encode(hyde_texts)

    # Calculate quality-based weights
    weights = self._calculate_document_weights(hypothetical_docs)

    # Weighted combination of hypothetical embeddings
    weighted_hyde = np.average(hyde_embeddings, axis=0, weights=weights)

    # Combine original query with weighted hypotheticals
    # Give more weight to original query to preserve intent
    combined_embedding = (0.3 * original_embedding +
                        0.7 * weighted_hyde)

    # Normalize the combined embedding
    combined_embedding = combined_embedding / np.linalg.norm(combined_embedding)

    return {
        'original': original_embedding,
        'hyde_embeddings': hyde_embeddings,
        'weighted_hyde': weighted_hyde,
        'combined': combined_embedding,
        'weights': weights
    }

def _calculate_document_weights(self, hypothetical_docs: List[Dict]) -> List[float]:
    """Calculate weights for hypothetical documents based on quality."""

    quality_scores = [doc['quality_score'] for doc in hypothetical_docs]

    # Softmax normalization for weights
    exp_scores = np.exp(np.array(quality_scores))
    weights = exp_scores / np.sum(exp_scores)

    return weights.tolist()

def _calculate_enhancement_confidence(self, enhanced_embeddings: Dict) -> float:
    """Calculate confidence score for the enhancement process."""

    original = enhanced_embeddings['original']
    combined = enhanced_embeddings['combined']

    # Calculate similarity between original and enhanced
    similarity = np.dot(original, combined) / (
        np.linalg.norm(original) * np.linalg.norm(combined)
    )

    # Higher similarity indicates conservative enhancement
    # Lower similarity indicates more aggressive transformation
    confidence_score = 1.0 - abs(similarity - 0.7)  # Target 0.7 similarity

    return max(0.0, min(1.0, confidence_score))

# Test the HyDE system
enhancer = HyDEQueryEnhancer(llm_model, embedding_model)

# Test factual query
factual_result = enhancer.enhance_query_with_hyde(
    "What is the capital of France?",
    query_type='factual'
)

# Test procedural query
procedural_result = enhancer.enhance_query_with_hyde(
    "How do I deploy a Docker container?",
    query_type='procedural'
)

# Test analytical query
analytical_result = enhancer.enhance_query_with_hyde(
    "What are the pros and cons of microservices?",
    query_type='analytical'
)

print("Enhancement Results:")
print(f"Factual confidence: {factual_result['confidence_score']}")
print(f"Procedural confidence: {procedural_result['confidence_score']}")
print(f"Analytical confidence: {analytical_result['confidence_score']}")

def search_with_hyde(self, query: str, vector_store, top_k: int = 10):
    """Perform enhanced search using HyDE."""

    # Generate enhanced embedding
    hyde_result = self.enhance_query_with_hyde(query)
    enhanced_embedding = hyde_result['enhanced_embedding']

    # Search with enhanced embedding
    enhanced_results = vector_store.similarity_search_by_vector(
        enhanced_embedding, k=top_k
    )

    # Also search with original query for comparison
    original_results = vector_store.similarity_search(query, k=top_k)

    return {
        'enhanced_results': enhanced_results,
        'original_results': original_results,
        'hyde_metadata': hyde_result,
        'improvement_score': self._calculate_improvement(
            enhanced_results, original_results, query
        )
    }