File size: 10,011 Bytes
b30f068 | 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 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 | """
LLM Intent Classifier
Uses LLM to classify user intent and select appropriate tool
"""
import sys
from pathlib import Path
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))
import openai
import json
import hashlib
from typing import Dict, List, Optional
from src.config.credentials import CredentialsManager
import os
class LLMIntentClassifier:
"""
Classifies user queries using LLM to determine intent and select tool
"""
def __init__(self, api_key: str = None, model: str = None):
"""
Initialize LLM Intent Classifier
Args:
api_key: OpenAI API key (optional, loads from .env if not provided)
model: OpenAI model to use (default: gpt-4o-mini for cost efficiency)
"""
if api_key is None:
creds = CredentialsManager()
api_key = creds.get_api_key("openai")
self.client = openai.OpenAI(api_key=api_key)
self.model = model or os.getenv("OPENAI_MODEL_NAME", "gpt-4o-mini")
self.cache = {} # Simple in-memory cache (can be upgraded to Redis later)
self.cache_size_limit = 1000 # Limit cache size
def classify_intent(
self,
query: str,
conversation_context: list = None
) -> Dict:
"""
Classify user intent and select appropriate tool using LLM
Args:
query: User's query
conversation_context: Optional conversation history
Returns:
Dict with tool_name, confidence, reasoning, method
"""
# Check cache first
cache_key = self._generate_cache_key(query, conversation_context)
if cache_key in self.cache:
cached_result = self.cache[cache_key].copy()
cached_result["method"] = "llm_cached"
return cached_result
# Build prompt
prompt = self._build_classification_prompt(query, conversation_context)
# Call LLM
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": self._get_system_prompt()
},
{
"role": "user",
"content": prompt
}
],
temperature=0.3, # Lower temperature for more consistent classification
max_tokens=200,
response_format={"type": "json_object"} # Force JSON response
)
result = json.loads(response.choices[0].message.content)
# Validate and normalize result
result = self._validate_result(result)
# Cache result (with size limit)
if len(self.cache) >= self.cache_size_limit:
# Remove oldest entry (simple FIFO)
oldest_key = next(iter(self.cache))
del self.cache[oldest_key]
self.cache[cache_key] = result.copy()
result["method"] = "llm"
return result
except json.JSONDecodeError as e:
# If LLM doesn't return valid JSON, fallback
print(f"⚠️ LLM returned invalid JSON: {e}")
return {
"tool_name": "agriculture_web",
"confidence": 0.5,
"reasoning": "LLM classification failed: invalid JSON response",
"method": "fallback"
}
except Exception as e:
# Fallback to default
print(f"⚠️ LLM classification error: {e}")
return {
"tool_name": "agriculture_web",
"confidence": 0.5,
"reasoning": f"LLM classification failed: {str(e)}",
"method": "fallback"
}
def _get_system_prompt(self) -> str:
"""Get system prompt for tool classification"""
return """You are an expert at classifying user queries and selecting the appropriate tool.
Available tools:
1. weather - For weather, temperature, forecast, climate queries (e.g., "What's the weather in London?", "Temperature in Tokyo")
2. soil - For soil data, soil properties, agricultural soil information (e.g., "Show me soil data for Iowa", "Soil pH in California")
3. cdms_label - For pesticide labels, herbicide labels, product labels, safety data sheets (e.g., "Find Roundup label", "What's the application rate for Sevin?", "Safety precautions for 2,4-D")
4. agriculture_web - For general agriculture questions, best practices, farming advice (e.g., "How to control aphids?", "Best practices for corn fertilization")
Return a JSON object with:
- tool_name: one of the tool names above (exactly as listed)
- confidence: float between 0.0 and 1.0 indicating how confident you are
- reasoning: brief explanation (1-2 sentences) of why this tool was selected
Be especially careful with follow-up questions - use the conversation context to understand what the user is asking about."""
def _build_classification_prompt(
self,
query: str,
context: list = None
) -> str:
"""Build classification prompt"""
prompt = f"Classify this user query and select the best tool:\n\n"
prompt += f"Query: {query}\n\n"
if context:
prompt += "Conversation context:\n"
for i, msg in enumerate(context[-3:], 1): # Last 3 messages
role = msg.get("role", "user")
content = msg.get("content", "")[:200] # Truncate long messages
prompt += f"{i}. {role}: {content}\n"
prompt += "\n"
prompt += "Consider the conversation context when classifying. If this is a follow-up question, use context to understand what the user is asking about.\n\n"
prompt += "Return your response as JSON with tool_name, confidence, and reasoning fields."
return prompt
def _validate_result(self, result: Dict) -> Dict:
"""Validate and normalize LLM result"""
valid_tools = ["weather", "soil", "cdms_label", "agriculture_web"]
tool_name = result.get("tool_name", "agriculture_web")
if tool_name not in valid_tools:
# Try to map common variations
tool_name_lower = tool_name.lower()
if "weather" in tool_name_lower or "temperature" in tool_name_lower:
tool_name = "weather"
elif "soil" in tool_name_lower:
tool_name = "soil"
elif "cdms" in tool_name_lower or "label" in tool_name_lower or "pesticide" in tool_name_lower:
tool_name = "cdms_label"
else:
tool_name = "agriculture_web" # Default fallback
confidence = float(result.get("confidence", 0.5))
confidence = max(0.0, min(1.0, confidence)) # Clamp to 0-1
reasoning = result.get("reasoning", "No reasoning provided")
return {
"tool_name": tool_name,
"confidence": confidence,
"reasoning": reasoning
}
def _generate_cache_key(self, query: str, context: list = None) -> str:
"""Generate cache key for query"""
# Normalize query
query_normalized = query.lower().strip()
# Include context if present (last message only for cache key to keep it simple)
if context:
last_msg = context[-1].get("content", "")[:50] if context else ""
cache_str = f"{query_normalized}||{last_msg}"
else:
cache_str = query_normalized
# Generate hash
return hashlib.md5(cache_str.encode()).hexdigest()
def clear_cache(self):
"""Clear the classification cache"""
self.cache.clear()
def get_cache_stats(self) -> Dict:
"""Get cache statistics"""
return {
"cache_size": len(self.cache),
"cache_limit": self.cache_size_limit,
"cache_usage": len(self.cache) / self.cache_size_limit
}
# Test function
if __name__ == "__main__":
print("=" * 80)
print("Testing LLM Intent Classifier")
print("=" * 80)
classifier = LLMIntentClassifier()
test_queries = [
"What's the weather in London?",
"Show me soil data for Iowa",
"Find the Roundup pesticide label",
"How to control aphids on tomato plants?",
"What about safety?", # Follow-up (needs context)
]
# Test with context
context = [
{
"role": "user",
"content": "What's the application rate for Roundup?"
},
{
"role": "assistant",
"content": "The application rate for Roundup is 1.5-2.5 quarts per acre..."
}
]
for query in test_queries:
print(f"\n📝 Query: {query}")
print("-" * 80)
# Test without context
result = classifier.classify_intent(query)
print(f" Tool: {result['tool_name']}")
print(f" Confidence: {result['confidence']:.0%}")
print(f" Reasoning: {result['reasoning']}")
print(f" Method: {result.get('method', 'unknown')}")
# Test with context for follow-up
if "What about" in query:
print(f"\n With context:")
result_with_context = classifier.classify_intent(query, context)
print(f" Tool: {result_with_context['tool_name']}")
print(f" Confidence: {result_with_context['confidence']:.0%}")
print(f" Reasoning: {result_with_context['reasoning']}")
print("\n" + "=" * 80)
print("Cache Stats:", classifier.get_cache_stats())
print("=" * 80)
|