Spaces:
Sleeping
Sleeping
File size: 1,814 Bytes
6afc01a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 |
"""
Query Router - Stage 1
Simple router that always queries all MCP servers for alert generation.
The intelligence is in the compiler, not the router.
"""
from typing import Dict, Any
class QueryRouter:
"""
Router for alert generation system.
For alert generation, ALWAYS queries all MCP servers since we need
comprehensive data to identify potential issues. The compiler handles
the intelligence of extracting only alerting/concerning information.
"""
def __init__(self):
"""Initialize router - no LLM needed for simple all-server routing"""
pass
def route_alert_query(self, location: Dict[str, float]) -> Dict[str, bool]:
"""
Route for alert generation - always query ALL servers.
Args:
location: Dict with 'latitude' and 'longitude' keys
Returns:
Dict with all servers set to True
"""
return {
"weather": True,
"soil": True,
"water": True,
"elevation": True,
"pests": True
}
def route_query(self, query: str, location: Dict[str, float]) -> Dict[str, bool]:
"""
Route a general query - for now, also queries all servers.
In the future, could use LLM to determine which servers are relevant
to the specific query. But for alert generation, we always want all data.
Args:
query: User's query
location: Dict with 'latitude' and 'longitude' keys
Returns:
Dict indicating which servers to query
"""
# For now, query all servers for any query
# The compiler will extract relevant information
return self.route_alert_query(location) |