Queries

The queries namespace provides methods for executing natural language queries against your knowledge graphs. It features automatic domain detection, intelligent pipeline routing, and comprehensive result formatting.

Basic Query Execution

Synchronous Queries

from functor_sdk import FunctorClient

client = FunctorClient()

# Simple query
result = client.queries.execute("What is machine learning?")
print(result.answer)
print(f"Pipeline: {result.pipeline_used}")
print(f"Processing time: {result.processing_time_ms}ms")

Asynchronous Queries

import asyncio
from functor_sdk import FunctorClient

async def main():
    async with FunctorClient() as client:
        result = await client.queries.execute_async("What is machine learning?")
        print(result.answer)

asyncio.run(main())

Query Parameters

Complete Parameter Reference

result = client.queries.execute(
    query="What are the symptoms of diabetes?",
    user_id="alice",                    # Optional: User ID for personalization
    max_results=10,                     # Optional: Max results (1-100, default: 10)
    validate_answer=True,               # Optional: Validate answer quality
    include_citations=True,             # Optional: Include source citations
    kg_names=["KG_Universal"]           # Optional: Specific knowledge graphs
)

Parameter Details

query (required)

The natural language query to execute. Supports complex questions and multi-part queries.

# Simple questions
result = client.queries.execute("What is AI?")

# Complex questions
result = client.queries.execute(
    "What are the differences between supervised and unsupervised learning?"
)

# Multi-part queries
result = client.queries.execute(
    "Compare the symptoms of diabetes and hypertension, and explain their treatments"
)

user_id (optional)

User identifier for personalized responses and query history.

# Personalized query
result = client.queries.execute(
    query="What medications am I taking?",
    user_id="patient_123"
)

max_results (optional)

Maximum number of results to return. Range: 1-100, default: 10.

# Few results for focused answers
result = client.queries.execute(
    query="What is diabetes?",
    max_results=3
)

# More results for comprehensive answers
result = client.queries.execute(
    query="List all machine learning algorithms",
    max_results=50
)

validate_answer (optional)

Whether to validate answer quality and coherence. Default: True.

# With validation (default)
result = client.queries.execute(
    query="What is quantum computing?",
    validate_answer=True
)

# Without validation (faster)
result = client.queries.execute(
    query="What is quantum computing?",
    validate_answer=False
)

include_citations (optional)

Whether to include source citations in the response. Default: True.

# With citations (default)
result = client.queries.execute(
    query="What are the side effects of aspirin?",
    include_citations=True
)

# Access citations
for citation in result.citations:
    print(f"Source: {citation.source}")
    print(f"Relevance: {citation.relevance_score}")

# Without citations (faster)
result = client.queries.execute(
    query="What are the side effects of aspirin?",
    include_citations=False
)

kg_names (optional)

Specific knowledge graphs to search. If not provided, searches all available KGs.

# Search specific knowledge graphs
result = client.queries.execute(
    query="What is diabetes?",
    kg_names=["KG_Medical", "KG_Universal"]
)

# Search all available knowledge graphs (default)
result = client.queries.execute(
    query="What is diabetes?"
)

Response Format

QueryResult Object

result = client.queries.execute("What is machine learning?")

# Access response properties
print(f"Answer: {result.answer}")
print(f"Pipeline: {result.pipeline_used}")
print(f"Processing time: {result.processing_time_ms}ms")
print(f"Confidence: {result.confidence}")
print(f"Citations: {len(result.citations)}")

# Access metadata
print(f"KGs searched: {result.metadata.kgs_searched}")
print(f"Entities found: {result.metadata.entities_found}")

Citation Objects

result = client.queries.execute("What is diabetes?")

for citation in result.citations:
    print(f"Source: {citation.source}")
    print(f"Type: {citation.source_type}")
    print(f"Relevance: {citation.relevance_score}")
    print(f"Chunk: {citation.chunk_text[:100]}...")
    print(f"Metadata: {citation.metadata}")
    print("---")

Advanced Usage Patterns

Batch Queries

# Synchronous batch processing
queries = [
    "What is machine learning?",
    "What is deep learning?",
    "What is natural language processing?"
]

results = []
for query in queries:
    result = client.queries.execute(query)
    results.append(result)

# Process results
for i, result in enumerate(results):
    print(f"Query {i+1}: {result.answer[:100]}...")

Concurrent Async Queries

import asyncio
from functor_sdk import FunctorClient

async def execute_queries():
    async with FunctorClient() as client:
        queries = [
            "What is AI?",
            "What is ML?",
            "What is DL?"
        ]
        
        # Execute queries concurrently
        tasks = [
            client.queries.execute_async(query)
            for query in queries
        ]
        
        results = await asyncio.gather(*tasks)
        
        for i, result in enumerate(results):
            print(f"Query {i+1}: {result.answer[:100]}...")

asyncio.run(execute_queries())

Query with Error Handling

from functor_sdk import (
    FunctorClient,
    FunctorAPIError,
    FunctorTimeoutError
)

client = FunctorClient()

try:
    result = client.queries.execute(
        query="What is quantum computing?",
        max_results=5,
        timeout=60  # Custom timeout for complex queries
    )
    
    if result.confidence < 0.7:
        print("Warning: Low confidence answer")
    
    print(f"Answer: {result.answer}")
    
except FunctorTimeoutError:
    print("Query timed out - try a simpler question")
    
except FunctorAPIError as e:
    print(f"Query failed: {e.message}")

Domain-Specific Queries

# Medical queries
medical_result = client.queries.execute(
    query="What are the treatment options for hypertension?",
    kg_names=["KG_Medical"],
    validate_answer=True,
    include_citations=True
)

# Technical queries
tech_result = client.queries.execute(
    query="How does gradient descent work in neural networks?",
    kg_names=["KG_Technical"],
    max_results=15
)

# General knowledge queries
general_result = client.queries.execute(
    query="What is the history of artificial intelligence?",
    kg_names=["KG_Universal"]
)

Performance Optimization

Query Optimization Tips

Be specific: More specific queries yield better results
Use appropriate max_results: Don't request more results than needed
Disable validation for simple queries: Set validate_answer=False for faster responses
Target specific KGs: Use kg_names to search only relevant knowledge graphs
Use async for concurrent operations: Execute multiple queries simultaneously

Example Optimizations

# Optimized for speed
fast_result = client.queries.execute(
    query="What is diabetes?",
    max_results=3,
    validate_answer=False,
    include_citations=False
)

# Optimized for accuracy
accurate_result = client.queries.execute(
    query="What are the latest treatment protocols for type 2 diabetes?",
    max_results=10,
    validate_answer=True,
    include_citations=True,
    kg_names=["KG_Medical"]
)

Common Use Cases

Question Answering

# Direct questions
result = client.queries.execute("What is the capital of France?")
print(result.answer)

# How/Why questions
result = client.queries.execute("How does photosynthesis work?")
print(result.answer)

# Comparison questions
result = client.queries.execute("What's the difference between Python and Java?")
print(result.answer)

Information Retrieval

# List requests
result = client.queries.execute(
    query="List all machine learning algorithms",
    max_results=20
)

# Specific information
result = client.queries.execute(
    query="What are the side effects of metformin?",
    kg_names=["KG_Medical"]
)

Research Assistance

# Research questions
result = client.queries.execute(
    query="What are the latest developments in quantum computing?",
    include_citations=True,
    max_results=15
)

# Access sources for further research
for citation in result.citations:
    print(f"Source: {citation.source}")
    print(f"Relevance: {citation.relevance_score}")
    print(f"Text: {citation.chunk_text[:200]}...")
    print("---")

Error Handling

Common Error Scenarios

from functor_sdk import (
    FunctorClient,
    FunctorAPIError,
    FunctorAuthenticationError,
    FunctorNotFoundError,
    FunctorTimeoutError
)

client = FunctorClient()

try:
    result = client.queries.execute("What is AI?")
    
except FunctorAuthenticationError:
    print("Authentication failed - check your API key")
    
except FunctorNotFoundError:
    print("No knowledge graphs found")
    
except FunctorTimeoutError:
    print("Query timed out - try a simpler question")
    
except FunctorAPIError as e:
    print(f"API error {e.status_code}: {e.message}")
    
except Exception as e:
    print(f"Unexpected error: {e}")

🆕 Enhanced Query Methods (NEW)

The Functor SDK now includes enhanced query methods with advanced features including intelligent auto-routing, quality filtering, and comprehensive analytics.

Enhanced Smart Query

Execute queries with intelligent auto-routing, quality filtering, and advanced analytics. Returns EnhancedQueryResponse with comprehensive execution metadata.

from functor_sdk import FunctorClient

client = FunctorClient()

# Enhanced smart query with auto-routing
result = client.queries.execute_smart(
    query="What is deep learning and how does it relate to neural networks?",
    mode="auto",
    enable_reranking=True,
    max_results=10,
    quality_threshold=0.7,
    enable_analytics=True
)

# Access enhanced metadata
print(f"Mode used: {result.execution_metadata.actual_mode}")
print(f"Routing confidence: {result.execution_metadata.routing_confidence}")
print(f"Response time: {result.execution_metadata.total_response_time_ms}ms")
print(f"Quality score: {result.analytics.result_quality_score}")
print(f"Duplicates removed: {result.analytics.duplicate_documents_removed}")

# Access routing explanation
print(f"Selected mode: {result.routing_explanation.selected_mode}")
print(f"Reasoning: {result.routing_explanation.reasoning}")

Enhanced SQL Processing

Advanced SQL query generation with comprehensive analytics, query explanation, validation, and multiple output formats.

# Enhanced SQL processing
sql_result = client.queries.execute_enhanced_sql(
    query="Show me the first 5 entities and their relationships",
    max_results=5,
    include_explanation=True,
    query_type="auto",
    output_format="summary",
    include_schema_info=True
)

# Access SQL results
print(f"Generated SQL: {sql_result.generated_sql}")
print(f"Query type: {sql_result.query_type}")
print(f"Validation status: {sql_result.validation_status}")
print(f"Execution time: {sql_result.execution_time_ms}ms")

# Access schema information
for table_name, table_info in sql_result.schema_info.items():
    print(f"Table: {table_name}")
    print(f"  Columns: {table_info.columns}")
    print(f"  Total rows: {table_info.total_rows}")

# Access analytics
print(f"Query complexity: {sql_result.analytics.query_complexity}")
print(f"Join operations: {sql_result.analytics.join_operations}")

Enhanced Smart Re-ranking

Re-rank documents with quality filtering, deduplication, and advanced analytics. Returns EnhancedReRankResponse with comprehensive quality metrics.

# Enhanced smart re-ranking
documents = [
    {"content": "AI is transforming industries...", "title": "AI Revolution"},
    {"content": "Machine learning algorithms...", "title": "ML Guide"}
]

rerank_result = client.queries.rerank_smart(
    query="artificial intelligence and machine learning trends",
    documents=documents,
    method="listwise",
    quality_filter="auto",
    max_results=10,
    return_analytics=True,
    deduplication_threshold=0.85,
    min_quality_score=0.6
)

# Access ranked documents
for doc in rerank_result.ranked_documents:
    print(f"Rank {doc.rank}: {doc.document['title']}")
    print(f"  Score: {doc.score}")
    print(f"  Quality: {doc.quality_score}")

# Access analytics
print(f"Avg quality score: {rerank_result.analytics.quality_metrics.avg_quality_score}")
print(f"Duplicates removed: {rerank_result.analytics.deduplication_stats.duplicates_removed}")
print(f"Processing time: {rerank_result.processing_time_ms}ms")

Async Enhanced Methods

All enhanced methods are also available in async versions:

import asyncio
from functor_sdk import FunctorClient

async def main():
    async with FunctorClient() as client:
        # Enhanced smart query (async)
        result = await client.queries.execute_smart_async(
            query="What is machine learning?",
            mode="auto",
            enable_analytics=True
        )
        
        # Enhanced SQL (async)
        sql_result = await client.queries.execute_enhanced_sql_async(
            query="Count all entities by type",
            output_format="statistics"
        )
        
        # Enhanced rerank (async)
        rerank_result = await client.queries.rerank_smart_async(
            query="AI trends",
            documents=documents,
            quality_filter="auto"
        )

asyncio.run(main())

Next Steps

Ingestion Namespace - Upload data to build your knowledge graphs
Sources Namespace - Manage your data sources
Error Handling - Comprehensive error management
Async vs Sync - Choose the right interface