Skip to main content

Query Understanding System

The Query Understanding System is a comprehensive solution designed to address the challenges enterprise users face when interacting with RAG systems. It provides intelligent query processing that understands user intent, reformulates queries for better retrieval, and continuously improves through user feedback.

Overview​

Enterprise users often ask questions that lack context, use internal jargon, or combine multiple intents. The Query Understanding System solves these problems by:

  • Intent Classification: Identifying question types (factual, procedural, comparative, exploratory)
  • Query Expansion: Using synonyms and domain terminology to improve retrieval
  • Query Rewriting: Optimizing queries for better search performance
  • Multi-Intent Decomposition: Breaking down complex queries into simpler components
  • Context-Aware Processing: Using conversation history to improve understanding
  • Query Validation: Detecting ambiguous queries and requesting clarification
  • Domain-Specific Preprocessing: Handling industry-specific terminology

Architecture​

The system is built with a modular architecture that allows for easy customization and extension:

Core Components​

Intent Classification​

The intent classifier identifies the type of question being asked:

  • Factual: "What is machine learning?"
  • Procedural: "How do I implement a neural network?"
  • Comparative: "Compare Python vs R for data science"
  • Exploratory: "Show me cloud computing options"
from packages.rag import QueryUnderstandingPipeline

pipeline = QueryUnderstandingPipeline()
await pipeline.initialize()

analysis = await pipeline.process_query("What is machine learning?")
print(f"Intent: {analysis.intent}")
print(f"Confidence: {analysis.confidence}")

Query Expansion​

Expands queries using synonyms and domain-specific terminology:

# With domain terms file
pipeline = QueryUnderstandingPipeline(domain_terms_file="domain_terms.yaml")

analysis = await pipeline.process_query("ML algorithms")
print(f"Expanded terms: {analysis.expanded_terms}")

Query Rewriting​

Rewrites queries for better retrieval performance:

analysis = await pipeline.process_query("What is API?")
print(f"Original: {analysis.original_query}")
print(f"Rewritten: {analysis.processed_query}")
# Output: "What is application programming interface?"

Multi-Intent Decomposition​

Breaks down complex queries into simpler components:

analysis = await pipeline.process_query(
    "What is machine learning and how do I implement it?"
)
print(f"Decomposed queries: {analysis.decomposed_queries}")

Context-Aware Processing​

Uses conversation history to improve query understanding:

# With conversation context
analysis = await pipeline.process_query(
    "How do I install it?", 
    thread_id="conversation_123"
)
print(f"Context relevance: {analysis.context_relevance}")

Configuration​

Domain Terms Configuration​

Create a YAML file with domain-specific terminology:

terms:
  - term: "API"
    synonyms: ["application programming interface", "web service"]
    definition: "Application Programming Interface"
    category: "technical"
    weight: 0.9
  - term: "ML"
    synonyms: ["machine learning", "artificial intelligence"]
    definition: "Machine Learning"
    category: "technical"
    weight: 0.8

Domain Configuration​

Configure domain-specific preprocessing:

domain_config = {
    "technical": {
        "preprocessing_rules": [
            {
                "pattern": r"\bAPI\b",
                "replacement": "application programming interface"
            }
        ]
    },
    "medical": {
        "preprocessing_rules": [
            {
                "pattern": r"\bBP\b",
                "replacement": "blood pressure"
            }
        ]
    }
}

pipeline = QueryUnderstandingPipeline(domain_config=domain_config)

A/B Testing​

The system includes comprehensive A/B testing capabilities for continuous improvement:

Creating an A/B Test​

from packages.rag import ABTestManager, ABTest, TestVariant, TestType

manager = ABTestManager()

test = ABTest(
    test_id="intent_classification_v2",
    name="Improved Intent Classification",
    description="Test new intent classification model",
    test_type=TestType.INTENT_CLASSIFICATION,
    variants=[
        TestVariant(
            name="control",
            description="Current model",
            config={"model": "current"},
            is_control=True
        ),
        TestVariant(
            name="treatment",
            description="New model",
            config={"model": "improved"}
        )
    ],
    success_metrics=["accuracy", "confidence"],
    minimum_sample_size=1000
)

await manager.create_test(test)
await manager.start_test("intent_classification_v2")

Assigning Variants​

# Automatically assign variant to user
variant = await manager.assign_variant("intent_classification_v2", "user_123")

Analyzing Results​

# Get test results
results = await manager.analyze_test_results("intent_classification_v2")
recommendation = await manager.get_test_recommendation("intent_classification_v2")

User Feedback Integration​

Collect and analyze user feedback for continuous improvement:

Collecting Feedback​

from packages.rag import FeedbackIntegrationManager, FeedbackChannel

feedback_manager = FeedbackIntegrationManager()

# Collect feedback after query processing
feedback_items = await feedback_manager.collect_and_store_feedback(
    analysis, 
    user_id="user_123", 
    session_id="session_456",
    channels=[FeedbackChannel.INLINE, FeedbackChannel.SURVEY]
)

Analyzing Feedback​

# Get feedback summary
summary = await feedback_manager.get_feedback_summary(analysis.query_analysis_id)

# Get insights and recommendations
insights = await feedback_manager.get_insights(days_back=30)

Query Analysis and Metrics​

Monitor system performance with comprehensive analytics:

Basic Metrics​

from packages.rag import QueryAnalyzer

analyzer = QueryAnalyzer()

# Store analysis
query_id = await analyzer.store_analysis(analysis)

# Get metrics
metrics = await analyzer.get_metrics(days_back=30)
print(f"Total queries: {metrics.total_queries}")
print(f"Average confidence: {metrics.avg_confidence}")
print(f"Intent accuracy: {metrics.intent_accuracy}")

Detailed Reports​

# Intent classification report
intent_report = await analyzer.generate_intent_classification_report(days_back=30)
print(intent_report.classification_report)

# Query transformation report
transform_report = await analyzer.generate_transformation_report(days_back=30)
print(f"Rewrite effectiveness: {transform_report.rewrite_effectiveness}")

Visualizations​

# Generate analysis plots
analyzer.generate_visualizations("analysis_plots/")

Troubleshooting​

The system includes comprehensive troubleshooting capabilities:

Running Diagnostics​

from packages.rag import TroubleshootingGuide

guide = TroubleshootingGuide(analyzer, feedback_manager)

# Run full diagnosis
diagnosis = await guide.run_full_diagnosis(days_back=7)

print(f"Health score: {diagnosis['summary']['overall_health_score']}")
print(f"Issues found: {diagnosis['summary']['total_issues_found']}")

Generating Reports​

# Generate troubleshooting report
await guide.generate_troubleshooting_report("troubleshooting_report.txt", days_back=7)

Best Practices​

1. Domain-Specific Configuration​

  • Create comprehensive domain term dictionaries
  • Configure preprocessing rules for your industry
  • Regularly update terminology based on user feedback

2. A/B Testing Strategy​

  • Start with high-impact, low-risk changes
  • Ensure sufficient sample sizes for statistical significance
  • Monitor multiple metrics, not just accuracy

3. Feedback Collection​

  • Implement multiple feedback channels
  • Make feedback collection non-intrusive
  • Act on feedback insights promptly

4. Performance Monitoring​

  • Set up automated monitoring and alerting
  • Track both technical and user satisfaction metrics
  • Regular performance reviews and optimization

5. Continuous Improvement​

  • Use diagnostic insights to guide improvements
  • Implement changes incrementally
  • Measure impact of each change

Common Issues and Solutions​

Low Confidence Scores​

Problem: Many queries have low confidence scores Solutions:

  • Add more training data for intent classification
  • Improve query preprocessing rules
  • Implement query clarification for ambiguous cases

High Processing Times​

Problem: Query processing takes too long Solutions:

  • Implement caching for common queries
  • Optimize model inference
  • Use faster, lighter models where appropriate

Poor Intent Classification​

Problem: Intent classification accuracy is low Solutions:

  • Retrain the classification model
  • Add domain-specific intent patterns
  • Use ensemble methods for better accuracy

Context Mismatch​

Problem: Queries don't use conversation context effectively Solutions:

  • Improve context extraction algorithms
  • Increase context window size
  • Better topic tracking across conversations

API Reference​

QueryUnderstandingPipeline​

Main pipeline for query understanding.

class QueryUnderstandingPipeline:
    def __init__(
        self,
        domain_terms_file: Optional[str] = None,
        history_api: Optional[Any] = None,
        domain_config: Optional[Dict[str, Any]] = None
    )
    
    async def process_query(
        self,
        query: str,
        thread_id: Optional[str] = None,
        user_id: Optional[str] = None,
        domain: str = "general"
    ) -> QueryAnalysis

QueryAnalysis​

Result of query processing.

@dataclass
class QueryAnalysis:
    original_query: str
    processed_query: str
    intent: QueryIntent
    complexity: QueryComplexity
    confidence: float
    entities: List[Dict[str, Any]]
    keywords: List[str]
    expanded_terms: List[str]
    decomposed_queries: List[str]
    context_relevance: float
    needs_clarification: bool
    clarification_questions: List[str]
    domain_tags: List[str]
    processing_time_ms: float
    metadata: Dict[str, Any]

Examples​

Basic Usage​

from packages.rag import create_query_understanding_pipeline

# Create pipeline
pipeline = create_query_understanding_pipeline()
await pipeline.initialize()

# Process query
analysis = await pipeline.process_query("What is machine learning?")
print(f"Intent: {analysis.intent}")
print(f"Confidence: {analysis.confidence}")

With Domain Terms​

# Create pipeline with domain terms
pipeline = create_query_understanding_pipeline(
    domain_terms_file="medical_terms.yaml"
)
await pipeline.initialize()

# Process medical query
analysis = await pipeline.process_query("What is BP?", domain="medical")

With Conversation Context​

# Create pipeline with history API
pipeline = create_query_understanding_pipeline(history_api=history_api)
await pipeline.initialize()

# Process query with context
analysis = await pipeline.process_query(
    "How do I install it?", 
    thread_id="conversation_123"
)

A/B Testing​

from packages.rag import create_ab_test_manager

# Create A/B test manager
manager = create_ab_test_manager()

# Create and start test
test = ABTest(...)
await manager.create_test(test)
await manager.start_test(test.test_id)

# Assign variant and process query
variant = await manager.assign_variant(test.test_id, "user_123")
# Use variant configuration for query processing

This comprehensive query understanding system provides enterprise-grade capabilities for handling complex user queries, ensuring better retrieval performance and user satisfaction.