"""
Feedback system for the ModularOllamaAgent.
This module provides a feedback loop system that improves the agent's
semantic understanding of user queries by learning from past interactions.
It helps map natural language queries to appropriate functions and commands.
"""
import os
import re
import json
import time
import logging
from typing import Dict, Any, List, Optional, Tuple, Set, Union
# Set up logging
logger = logging.getLogger(__name__)
class FunctionSelectionFeedback:
"""Manages feedback on function selection for continuous improvement."""
def __init__(self, feedback_file_path=None):
"""
Initialize the feedback system.
Args:
feedback_file_path: Path to the feedback file (default: ~/.ollamatools/function_selection_feedback.jsonl)
"""
if feedback_file_path is None:
self.feedback_file_path = os.path.join(
os.path.expanduser('~'),
'.ollamatools',
'function_selection_feedback.jsonl'
)
else:
self.feedback_file_path = feedback_file_path
# Create directory if it doesn't exist
os.makedirs(os.path.dirname(self.feedback_file_path), exist_ok=True)
# Cache of recent feedback for quick lookup
self.recent_feedback = []
self.max_cache_size = 1000
# Load existing feedback into cache
self._load_recent_feedback()
def _load_recent_feedback(self):
"""Load recent feedback entries into memory cache."""
try:
if os.path.exists(self.feedback_file_path):
with open(self.feedback_file_path, 'r') as f:
# Load the most recent entries (up to max_cache_size)
lines = f.readlines()
for line in lines[-self.max_cache_size:]:
try:
entry = json.loads(line.strip())
self.recent_feedback.append(entry)
except json.JSONDecodeError:
continue
logger.info(f"Loaded {len(self.recent_feedback)} feedback entries from {self.feedback_file_path}")
else:
logger.info(f"No feedback file found at {self.feedback_file_path}, starting with empty feedback")
except Exception as e:
logger.error(f"Error loading feedback data: {str(e)}")
def record_explicit_feedback(self, query, intended_function, actual_function, user_correction=True):
"""
Record explicit feedback when a function selection was incorrect.
Args:
query: The original user query
intended_function: The function that should have been used
actual_function: The function that was actually used
user_correction: Whether this feedback came directly from a user
Returns:
dict: The recorded feedback entry
"""
feedback_entry = {
"query": query,
"intended_function": intended_function,
"actual_function": actual_function,
"timestamp": time.time(),
"source": "user_correction" if user_correction else "system_detection",
"similarity_score": self._calculate_query_similarity(query)
}
# Add to cache
self.recent_feedback.append(feedback_entry)
if len(self.recent_feedback) > self.max_cache_size:
self.recent_feedback.pop(0)
# Write to file
try:
with open(self.feedback_file_path, 'a') as f:
f.write(json.dumps(feedback_entry) + '\n')
logger.info(f"Recorded explicit feedback for query: {query}")
except Exception as e:
logger.error(f"Error recording feedback: {str(e)}")
return feedback_entry
def record_implicit_feedback(self, query, function, confidence_score, execution_result):
"""
Record implicit feedback based on execution results.
Args:
query: The original user query
function: The function that was used
confidence_score: Confidence score for the function selection (0-1)
execution_result: Result of executing the function
Returns:
dict: The recorded feedback entry
"""
# Determine if the execution was likely successful
success_indicators = {
"execute_command": lambda r: r.get("returncode", 1) == 0 and not r.get("stderr"),
"read_file": lambda r: not (isinstance(r, str) and (r.startswith("Error") or r.startswith("Path validation error"))),
"search_web": lambda r: len(r) > 0 if isinstance(r, list) else True
}
likely_success = False
if function in success_indicators:
likely_success = success_indicators[function](execution_result)
feedback_entry = {
"query": query,
"function": function,
"timestamp": time.time(),
"source": "implicit_feedback",
"confidence_score": confidence_score,
"likely_success": likely_success,
"similarity_score": self._calculate_query_similarity(query)
}
# Only record if confidence was low or execution likely failed
if confidence_score < 0.7 or not likely_success:
# Add to cache
self.recent_feedback.append(feedback_entry)
if len(self.recent_feedback) > self.max_cache_size:
self.recent_feedback.pop(0)
# Write to file
try:
with open(self.feedback_file_path, 'a') as f:
f.write(json.dumps(feedback_entry) + '\n')
logger.info(f"Recorded implicit feedback for query: {query}")
except Exception as e:
logger.error(f"Error recording feedback: {str(e)}")
return feedback_entry
def _calculate_query_similarity(self, query):
"""
Calculate similarity between this query and previous queries.
Args:
query: The query to compare
Returns:
list: List of similar queries with their similarity scores
"""
# Simple word-based similarity for now
query_words = set(query.lower().split())
similarities = []
for entry in self.recent_feedback:
past_query = entry.get("query", "")
past_words = set(past_query.lower().split())
# Calculate Jaccard similarity
if not query_words or not past_words:
similarity = 0
else:
intersection = len(query_words.intersection(past_words))
union = len(query_words.union(past_words))
similarity = intersection / union if union > 0 else 0
if similarity > 0.3: # Only include reasonably similar queries
similarities.append({
"query": past_query,
"similarity": similarity,
"intended_function": entry.get("intended_function") or entry.get("function"),
"timestamp": entry.get("timestamp")
})
# Sort by similarity score
similarities.sort(key=lambda x: x["similarity"], reverse=True)
return similarities[:5] # Return top 5 similar queries
def get_function_suggestion(self, query):
"""
Get function suggestion based on feedback history.
Args:
query: The user query
Returns:
tuple: (suggested_function, confidence_score, similar_queries)
"""
# Get similar queries
similar_queries = self._calculate_query_similarity(query)
if not similar_queries:
return None, 0, []
# Count function occurrences weighted by similarity
function_scores = {}
for item in similar_queries:
function = item["intended_function"]
similarity = item["similarity"]
if function not in function_scores:
function_scores[function] = 0
function_scores[function] += similarity
# Find the function with the highest score
best_function = None
highest_score = 0
for function, score in function_scores.items():
if score > highest_score:
highest_score = score
best_function = function
# Calculate confidence based on the highest score and number of similar queries
confidence = highest_score / len(similar_queries) if similar_queries else 0
return best_function, confidence, similar_queries
def generate_improved_patterns(self):
"""
Generate improved regex patterns based on feedback data.
Returns:
dict: Dictionary mapping regex patterns to functions
"""
# Analyze the feedback data
analysis = self.analyze_function_selection_patterns()
if "error" in analysis:
return {}
improved_patterns = {}
# Process mismatches to generate patterns
for mismatch_info in analysis.get("top_mismatches", []):
mismatch, data = mismatch_info
actual, intended = mismatch.split(" → ")
# Skip if we don't have enough examples
if len(data["examples"]) < 2:
continue
# Generate a pattern from the examples
pattern = self._generate_regex_from_examples(data["examples"], intended)
if pattern:
improved_patterns[pattern] = intended
# Process query patterns
for pattern_info in analysis.get("top_query_patterns", []):
pattern_text, data = pattern_info
functions = data["functions"]
if not functions:
continue
# Find the most common function
most_common = max(functions.items(), key=lambda x: x[1])
function_name, count = most_common
# Only include if it's a strong pattern (used for the same function >70% of the time)
confidence = count / data["count"]
if confidence > 0.7:
# Create a regex pattern from the text pattern
regex_pattern = r"\b" + re.escape(pattern_text) + r".*"
improved_patterns[regex_pattern] = function_name
return improved_patterns
def _generate_regex_from_examples(self, examples, function):
"""
Generate a regex pattern from example queries.
Args:
examples: List of example queries
function: The target function
Returns:
str: Regex pattern
"""
if not examples:
return None
# Extract common words across examples
common_words = set(examples[0].lower().split())
for example in examples[1:]:
common_words &= set(example.lower().split())
# Remove very common words
stop_words = {"what", "is", "the", "a", "an", "in", "on", "at", "to", "for", "of", "with", "by"}
common_words -= stop_words
if not common_words:
# If no common words, try to find common starting words
starts_with = []
for example in examples:
words = example.lower().split()
if words:
starts_with.append(words[0])
# If we have consistent starting words
if len(set(starts_with)) == 1 and starts_with[0] not in stop_words:
return r"\b" + re.escape(starts_with[0]) + r".*"
return None
# Create a pattern with the common words
pattern_parts = []
for word in common_words:
pattern_parts.append(r"\b" + re.escape(word) + r"\b")
# Combine with lookaheads for flexibility in word order
pattern = "".join(f"(?=.*{part})" for part in pattern_parts) + ".*"
return pattern
def analyze_function_selection_patterns(self):
"""
Analyze patterns in function selection feedback.
Returns:
dict: Analysis results
"""
try:
# Load all feedback data
all_feedback = []
if os.path.exists(self.feedback_file_path):
with open(self.feedback_file_path, 'r') as f:
for line in f:
try:
entry = json.loads(line.strip())
all_feedback.append(entry)
except json.JSONDecodeError:
continue
if not all_feedback:
return {"error": "No feedback data available"}
# Analyze common mismatches
mismatches = {}
for entry in all_feedback:
if "intended_function" in entry and "actual_function" in entry:
if entry["intended_function"] != entry["actual_function"]:
key = f"{entry['actual_function']} → {entry['intended_function']}"
if key not in mismatches:
mismatches[key] = {
"count": 0,
"examples": []
}
mismatches[key]["count"] += 1
if len(mismatches[key]["examples"]) < 5: # Limit examples
mismatches[key]["examples"].append(entry["query"])
# Sort mismatches by frequency
sorted_mismatches = sorted(
[(k, v) for k, v in mismatches.items()],
key=lambda x: x[1]["count"],
reverse=True
)
# Analyze query patterns
query_patterns = {}
for entry in all_feedback:
query = entry.get("query", "")
words = query.lower().split()
# Extract potential patterns (first 2-3 words)
if len(words) >= 2:
pattern = " ".join(words[:2])
if pattern not in query_patterns:
query_patterns[pattern] = {
"count": 0,
"functions": {}
}
query_patterns[pattern]["count"] += 1
function = entry.get("intended_function") or entry.get("function")
if function:
if function not in query_patterns[pattern]["functions"]:
query_patterns[pattern]["functions"][function] = 0
query_patterns[pattern]["functions"][function] += 1
# Sort patterns by frequency
sorted_patterns = sorted(
[(k, v) for k, v in query_patterns.items()],
key=lambda x: x[1]["count"],
reverse=True
)
return {
"total_feedback_entries": len(all_feedback),
"top_mismatches": sorted_mismatches[:10],
"top_query_patterns": sorted_patterns[:10],
"improvement_suggestions": self._generate_improvement_suggestions(sorted_mismatches, sorted_patterns)
}
except Exception as e:
logger.error(f"Error analyzing feedback: {str(e)}")
return {"error": f"Error analyzing feedback: {str(e)}"}
def _generate_improvement_suggestions(self, mismatches, patterns):
"""
Generate suggestions for improving function selection.
Args:
mismatches: List of function mismatches
patterns: List of query patterns
Returns:
list: Improvement suggestions
"""
suggestions = []
# Suggest improvements based on common mismatches
if mismatches:
for mismatch, data in mismatches[:5]:
actual, intended = mismatch.split(" → ")
examples = data["examples"]
if examples:
pattern_words = set()
for example in examples:
pattern_words.update(example.lower().split())
# Filter out common words
common_words = {"what", "is", "the", "a", "an", "in", "on", "at", "to", "for", "of", "with", "by"}
pattern_words = pattern_words - common_words
if pattern_words:
suggestion = {
"type": "function_mapping",
"description": f"Improve mapping from '{actual}' to '{intended}'",
"examples": examples[:3],
"suggested_keywords": list(pattern_words)[:5],
"implementation_hint": f"Add pattern matching for these keywords to select '{intended}'"
}
suggestions.append(suggestion)
# Suggest improvements based on query patterns
if patterns:
for pattern, data in patterns[:5]:
functions = data["functions"]
if functions:
# Find the most common function for this pattern
most_common = max(functions.items(), key=lambda x: x[1])
function_name, count = most_common
suggestion = {
"type": "query_pattern",
"description": f"Add explicit handling for queries starting with '{pattern}'",
"suggested_function": function_name,
"confidence": count / data["count"],
"implementation_hint": f"Add pattern matching for queries starting with '{pattern}' to select '{function_name}'"
}
suggestions.append(suggestion)
return suggestions
def detect_command_intent(message, feedback_system=None):
"""
Detect if a message is asking for system information that requires a command.
Args:
message: User message
feedback_system: Optional FunctionSelectionFeedback instance
Returns:
Tuple[bool, Optional[str]]: (has_command_intent, suggested_command)
"""
# First check feedback-based patterns if available
if feedback_system:
suggested_function, confidence, similar_queries = feedback_system.get_function_suggestion(message)
if suggested_function == "execute_command" and confidence > 0.7:
# Extract command from similar queries
command = extract_command_from_similar_queries(message, similar_queries)
if command:
return True, command
# Map common questions to shell commands
command_patterns = {
r"(?:what|which)(?:\s+is)?(?:\s+the)?(?:\s+current)?\s+directory": "pwd",
r"where\s+am\s+i": "pwd",
r"list\s+(?:the\s+)?(?:files|contents)": "ls",
r"show\s+(?:the\s+)?(?:files|contents)": "ls",
r"(?:what|which)(?:\s+is)?(?:\s+the)?(?:\s+current)?\s+user": "whoami",
r"who\s+am\s+i": "whoami",
r"(?:what|which)(?:\s+is)?(?:\s+the)?(?:\s+current)?\s+date": "date",
r"(?:what|which)(?:\s+is)?(?:\s+the)?(?:\s+current)?\s+time": "date",
# Add more patterns as needed
}
# Check if message matches any pattern
for pattern, command in command_patterns.items():
if re.search(pattern, message.lower()):
return True, command
return False, None
def extract_command_from_similar_queries(message, similar_queries):
"""
Extract the most appropriate command from similar queries.
Args:
message: Current user message
similar_queries: List of similar queries with their metadata
Returns:
str: Extracted command or None
"""
# Count command occurrences in similar queries
command_counts = {}
for query_info in similar_queries:
# Skip if not execute_command
if query_info.get("intended_function") != "execute_command":
continue
# Try to extract the command from the query
query = query_info.get("query", "")
# Check for common command patterns
command_mapping = {
r"directory|where\s+am\s+i": "pwd",
r"list\s+files|show\s+files|contents": "ls",
r"user|who\s+am\s+i": "whoami",
r"date|time": "date"
}
for pattern, cmd in command_mapping.items():
if re.search(pattern, query.lower()):
if cmd not in command_counts:
command_counts[cmd] = 0
command_counts[cmd] += query_info.get("similarity", 0.5)
# Return the most common command
if command_counts:
return max(command_counts.items(), key=lambda x: x[1])[0]
return None
def format_command_result(command, result):
"""
Format the result of a command execution in a user-friendly way.
Args:
command: The command that was executed
result: The command execution result
Returns:
str: Formatted result
"""
if "error" in result:
return f"Error executing command: {result['error']}"
stdout = result.get("stdout", "").strip()
stderr = result.get("stderr", "").strip()
if command == "pwd":
return f"You are currently in: {stdout}"
elif command == "ls":
return f"Files in the current directory:\n{stdout}"
elif command == "whoami":
return f"Current user: {stdout}"
elif command == "date":
return f"Current date and time: {stdout}"
else:
# Generic formatting
response = f"Command result:\n{stdout}"
if stderr:
response += f"\nErrors/Warnings:\n{stderr}"
return response