import requests
import json
from typing import Optional, Dict, Any, List
def update_smart_rag_configuration(
flow_name: str,
node_id: str,
config: Dict[str, Any],
api_token: str
) -> Dict[str, Any]:
url = f"https://{flow_name}.flows.graphorlm.com/smart-rag/{node_id}"
headers = {
"Authorization": f"Bearer {api_token}",
"Content-Type": "application/json"
}
payload = {"config": config}
response = requests.patch(url, headers=headers, json=payload)
response.raise_for_status()
return response.json()
# Configuration examples
def demonstrate_smart_rag_configurations():
api_token = "YOUR_API_TOKEN"
flow_name = "my-rag-pipeline"
node_id = "smart-rag-1748287628685"
configurations = [
{"name": "High Precision", "config": {"topK": 3}},
{"name": "Balanced", "config": {"topK": 10}},
{"name": "Comprehensive", "config": {"topK": 20}},
{"name": "Unlimited", "config": {"topK": None}},
]
for config_info in configurations:
try:
result = update_smart_rag_configuration(
flow_name, node_id, config_info["config"], api_token
)
print(f"✅ {config_info['name']} configuration applied successfully")
print(f" Message: {result['message']}")
except requests.exceptions.HTTPError as e:
print(f"❌ Failed to apply {config_info['name']} configuration: {e}")
# Advanced configuration validator
class SmartRagConfigValidator:
def __init__(self, flow_name: str, api_token: str):
self.flow_name = flow_name
self.api_token = api_token
self.base_url = f"https://{flow_name}.flows.graphorlm.com"
def validate_configuration(self, config: Dict[str, Any]) -> Dict[str, Any]:
"""Validate smart RAG configuration and provide recommendations"""
validation_result = {
"is_valid": True,
"errors": [],
"warnings": [],
"recommendations": [],
"resource_analysis": {},
"performance_analysis": {}
}
top_k = config.get('topK')
# Validate topK parameter
if top_k is not None:
if not isinstance(top_k, int):
validation_result["errors"].append("topK must be an integer or null")
validation_result["is_valid"] = False
elif top_k <= 0:
validation_result["errors"].append("topK must be greater than 0")
validation_result["is_valid"] = False
elif top_k == 1:
validation_result["warnings"].append("topK = 1 may be too restrictive for most use cases")
elif top_k > 50:
validation_result["warnings"].append("topK > 50 may significantly impact performance and resource usage")
elif top_k > 100:
validation_result["errors"].append("topK > 100 is not recommended due to performance constraints")
validation_result["is_valid"] = False
# Resource analysis
validation_result["resource_analysis"] = self._analyze_resource_usage(top_k)
# Performance analysis
validation_result["performance_analysis"] = self._analyze_performance(top_k)
# Generate recommendations
validation_result["recommendations"] = self._generate_recommendations(top_k)
return validation_result
def _analyze_resource_usage(self, top_k: Optional[int]) -> dict:
"""Analyze resource usage implications of Top K setting"""
if top_k is None:
return {
"level": "very_high",
"description": "Unlimited processing - highest resource consumption",
"relative_usage": "5x baseline"
}
elif top_k <= 5:
return {
"level": "low",
"description": "Minimal resource consumption",
"relative_usage": "1x baseline"
}
elif top_k <= 15:
return {
"level": "medium",
"description": "Moderate resource consumption",
"relative_usage": f"{top_k/5:.1f}x baseline"
}
elif top_k <= 30:
return {
"level": "high",
"description": "High resource consumption",
"relative_usage": f"{top_k/5:.1f}x baseline"
}
else:
return {
"level": "very_high",
"description": "Very high resource consumption",
"relative_usage": f"{top_k/5:.1f}x baseline"
}
def _analyze_performance(self, top_k: Optional[int]) -> dict:
"""Analyze performance implications of Top K setting"""
if top_k is None:
return {
"speed": "slowest",
"description": "Processes all documents - longest processing time",
"estimated_time": "high"
}
elif top_k <= 5:
return {
"speed": "fastest",
"description": "Minimal processing overhead",
"estimated_time": "low"
}
elif top_k <= 15:
return {
"speed": "good",
"description": "Balanced processing time",
"estimated_time": "medium"
}
elif top_k <= 30:
return {
"speed": "moderate",
"description": "Higher processing overhead",
"estimated_time": "medium-high"
}
else:
return {
"speed": "slow",
"description": "Significant processing overhead",
"estimated_time": "high"
}
def _generate_recommendations(self, top_k: Optional[int]) -> List[str]:
"""Generate configuration recommendations based on Top K value"""
recommendations = []
if top_k is None:
recommendations.extend([
"Consider setting a specific Top K limit to control resource usage",
"Use unlimited only for comprehensive research applications",
"Monitor processing time and resource usage closely"
])
elif top_k <= 3:
recommendations.extend([
"Excellent for high-precision applications",
"Consider increasing if recall is important",
"Monitor result completeness"
])
elif top_k <= 10:
recommendations.extend([
"Good balance of precision and coverage",
"Suitable for most general applications",
"Monitor quality metrics for optimization"
])
elif top_k <= 20:
recommendations.extend([
"Good for comprehensive coverage",
"Monitor resource implications",
"Consider if all results are actually needed"
])
else:
recommendations.extend([
"Very high Top K - ensure this is necessary",
"Consider breaking into multiple queries",
"Monitor processing resource usage and time carefully"
])
return recommendations
def update_with_validation(self, node_id: str, config: Dict[str, Any]) -> Dict[str, Any]:
"""Update configuration with validation and confirmation"""
# Validate configuration
validation = self.validate_configuration(config)
print("🔍 Configuration Validation Results:")
print("=" * 40)
if validation["errors"]:
print("❌ Errors found:")
for error in validation["errors"]:
print(f" • {error}")
return {"success": False, "validation": validation}
if validation["warnings"]:
print("⚠️ Warnings:")
for warning in validation["warnings"]:
print(f" • {warning}")
print(f"\n⚡ Resource Analysis:")
resource = validation["resource_analysis"]
print(f" Level: {resource['level']}")
print(f" Impact: {resource['description']}")
print(f" Relative Usage: {resource['relative_usage']}")
print(f"\n⚡ Performance Analysis:")
performance = validation["performance_analysis"]
print(f" Speed: {performance['speed']}")
print(f" Description: {performance['description']}")
print(f" Processing Time: {performance['estimated_time']}")
# Ask for confirmation for high-impact configurations
top_k = config.get('topK')
if top_k is None or top_k > 20:
response = input(f"\n⚠️ This configuration may have significant resource/performance impact. Continue? (y/N): ")
if response.lower() != 'y':
print("Configuration update cancelled.")
return {"success": False, "cancelled": True}
# Apply configuration
try:
result = update_smart_rag_configuration(
self.flow_name, node_id, config, self.api_token
)
print(f"\n✅ Configuration updated successfully!")
print(f"Message: {result['message']}")
return {"success": True, "result": result, "validation": validation}
except Exception as e:
print(f"\n❌ Failed to update configuration: {e}")
return {"success": False, "error": str(e)}
def print_validation_summary(self, validation: Dict[str, Any]):
"""Print a summary of validation results"""
print("\n📊 Configuration Summary:")
print("-" * 25)
if validation["is_valid"]:
print("✅ Configuration is valid")
print(f"⚡ Resources: {validation['resource_analysis']['level']}")
print(f"⚡ Performance: {validation['performance_analysis']['speed']}")
else:
print("❌ Configuration has errors")
if validation["recommendations"]:
print(f"\n💡 Recommendations:")
for rec in validation["recommendations"]:
print(f" • {rec}")
# Usage example
validator = SmartRagConfigValidator("my-rag-pipeline", "YOUR_API_TOKEN")
# Test different configurations
test_configs = [
{"topK": 5},
{"topK": 15},
{"topK": 25},
{"topK": None}
]
for config in test_configs:
print(f"\n{'='*50}")
print(f"Testing configuration: {config}")
validation = validator.validate_configuration(config)
validator.print_validation_summary(validation)