"""
ModularOllamaAgent implementation.
This module provides the ModularOllamaAgent class, which extends
the base OllamaAgent with modular components, detailed feedback, and
additional features like search integration, command management,
issue tracking, and semantic function selection feedback.
"""
import os
import sys
import json
import time
import logging
import re
from typing import Dict, Any, List, Optional, Set, Union, Tuple
from ollamatools import OllamaAgent, Config
from ollamatools.core.utils.validators.command import DEFAULT_ALLOWLIST
from ollamatools.utils.logger import setup_logger
from ollamatools.plugins.integrations.opensearch.tools import (
opensearch_query,
opensearch_fetch_url,
get_opensearch_info,
DEFAULT_OPENSEARCH_CONFIG
)
from ollamatools.core.errors import (
OllamaToolsError,
CommandError,
FileError,
WebError,
handle_exception
)
from ollamatools.plugins.tools.system import execute_command
# Import agent components
from .search import detect_factual_intent, extract_search_context, execute_search
from .commands import list_allowed_commands, add_allowed_command, remove_allowed_command
from .issues import load_issues, save_issues, add_issue, list_issues, get_issue, update_issue
from .utils import print_status, print_success, print_error, print_warning, print_info, suggest_fix
from .feedback import (
FunctionSelectionFeedback,
detect_command_intent,
extract_command_from_similar_queries,
format_command_result
)
# Set up logging
logger = setup_logger("modular_ollama_agent", level="INFO")
# Extend the default command allowlist with useful commands
EXTENDED_ALLOWLIST = DEFAULT_ALLOWLIST.union({
"python", "python3", "calc", "bc", "find", "grep", "awk", "sed",
"head", "tail", "more", "less", "cat", "echo", "cd", "pwd"
})
# Configuration for OpenSearchNG
OPENSEARCH_CONFIG = DEFAULT_OPENSEARCH_CONFIG.copy()
OPENSEARCH_CONFIG.update({
"base_url": "http://localhost:8081", # Updated to your container's port
"engines": ["google", "bing", "duckduckgo", "brave"],
"num_results": 5
})
class ModularOllamaAgent:
"""
A modular OllamaAgent that provides transparent execution and better feedback.
This agent is designed with a modular architecture that separates functionality
into logical components for better maintainability and extensibility. It includes
features like search integration, command management, issue tracking, and
semantic function selection feedback.
"""
def __init__(
self,
model: str = "llama3.1:latest",
permission_prompts: bool = True,
opensearch_config: Optional[Dict[str, Any]] = None,
command_allowlist: Optional[Set[str]] = None,
feedback_file_path: Optional[str] = None
):
"""
Initialize the agent.
Args:
model: Ollama model to use
permission_prompts: Whether to display permission prompts
opensearch_config: OpenSearchNG configuration
command_allowlist: Custom command allowlist
feedback_file_path: Path to the feedback file (default: ~/.ollamatools/function_selection_feedback.jsonl)
"""
# Initialize OllamaAgent
self.agent = OllamaAgent(
model=model,
permission_prompts=permission_prompts
)
# Store configuration
self.opensearch_config = opensearch_config or OPENSEARCH_CONFIG
self.command_allowlist = command_allowlist or EXTENDED_ALLOWLIST
# Initialize state
self.context = []
self.execution_history = []
# Initialize issue tracking
self.issues = []
self.issue_id_counter = 1
# Path for saving issues
self.issues_file = os.path.join(
os.path.expanduser('~'),
'.ollamatools',
'issues.json'
)
# Create directory if it doesn't exist
os.makedirs(os.path.dirname(self.issues_file), exist_ok=True)
# Load existing issues if any
self.load_issues()
# Initialize function selection feedback system
self.function_feedback = FunctionSelectionFeedback(feedback_file_path)
logger.info("Initialized function selection feedback system")
# Define system prompt
self.system_prompt = f"""
You are an AI assistant with modular access to various tools.
You always explain what you're doing step by step and show your work.
When a user asks you a question, you should:
1. Identify the most appropriate tool for the task
2. Describe what you're going to do before you do it
3. Show the results of any commands or functions you execute
4. Explain how you interpret these results
MOST IMPORTANTLY: When calculating math problems, ALWAYS use the 'calc' command rather than
doing the calculation yourself or using python -c. For example: calc 2+2
For searching the web:
- Use the 'opensearch_query' function to perform web searches
- Use the 'opensearch_fetch_url' function to fetch web page content safely
- Specify search engines if needed (available: {', '.join(self.opensearch_config['engines'])})
For file and system operations:
- Use 'find_files' to locate files
- Use 'read_file' to view file content
- Use 'execute_command' for running shell commands like 'ls', 'pwd', etc.
For system information queries:
- When a user asks about the current directory, use 'pwd' command
- When a user asks about file listings, use 'ls' command
- When a user asks about the current user, use 'whoami' command
- When a user asks about system information, prefer shell commands over other methods
Always check command permissions carefully and explain what each command does.
For directory structure, use 'find .' or 'ls -la' commands.
When a user asks for information that might require searching the web or retrieving external data,
automatically determine if you should use the 'opensearch_query' function instead of requiring explicit search commands.
For example, if a user asks "can you give me a recipe for chilli sauce?", you should:
1. Recognize this as a request that requires external information
2. Use the opensearch_query function with an appropriate search query
3. Present the results in a helpful format
Only use tools when necessary - for factual questions you can answer directly, don't use search.
"""
# Check if OpenSearchNG is available
try:
self.print_status("Checking OpenSearchNG availability...")
info = get_opensearch_info(config=self.opensearch_config)
self.print_success(f"OpenSearchNG running at {self.opensearch_config['base_url']}")
# Log available engines
if 'engines' in info and info['engines']:
self.print_info(f"Available search engines: {', '.join(info['engines'])}")
except Exception as e:
self.print_error(f"OpenSearchNG not available: {str(e)}")
self.print_warning("Falling back to regular search methods")
self.suggest_fix("Make sure your OpenSearchNG instance is running at " +
f"{self.opensearch_config['base_url']}")
# Utility methods redirected to utils module
def print_status(self, message: str) -> None:
print_status(message)
def print_success(self, message: str) -> None:
print_success(message)
def print_error(self, message: str) -> None:
print_error(message)
def print_warning(self, message: str) -> None:
print_warning(message)
def print_info(self, message: str) -> None:
print_info(message)
def suggest_fix(self, suggestion: str) -> None:
suggest_fix(suggestion)
# Search methods redirected to search module
def detect_factual_intent(self, message: str) -> bool:
return detect_factual_intent(message)
def extract_search_context(self, search_results: Dict[str, Any]) -> str:
return extract_search_context(search_results)
def execute_search(self, query: str) -> Dict[str, Any]:
return execute_search(query, self.opensearch_config, self.agent.permission_manager, self.execution_history)
# Command management methods redirected to commands module
def list_allowed_commands(self) -> List[str]:
return list_allowed_commands(self.command_allowlist)
def add_allowed_command(self, command: str) -> bool:
return add_allowed_command(command, self.command_allowlist)
def remove_allowed_command(self, command: str) -> bool:
return remove_allowed_command(command, self.command_allowlist)
# Issue tracking methods redirected to issues module
def load_issues(self) -> None:
load_issues(self)
def save_issues(self) -> None:
save_issues(self)
def add_issue(self, title: str, description: str, severity: str = 'medium') -> int:
return add_issue(self, title, description, severity)
def list_issues(self, status: Optional[str] = None) -> List[Dict[str, Any]]:
return list_issues(self, status)
def get_issue(self, issue_id: int) -> Optional[Dict[str, Any]]:
return get_issue(self, issue_id)
def update_issue(self, issue_id: int, status: Optional[str] = None,
severity: Optional[str] = None, description: Optional[str] = None) -> bool:
return update_issue(self, issue_id, status, severity, description)
def show_execution_history(self, max_entries: int = 10) -> None:
"""
Display the execution history.
Args:
max_entries: Maximum number of history entries to show
"""
if not self.execution_history:
print("\n📜 Execution history is empty.")
return
# Get the most recent entries
recent_entries = self.execution_history[-max_entries:]
print("\n📜 Execution History:")
print("=" * 60)
for i, entry in enumerate(recent_entries, 1):
# Format timestamp
timestamp = time.strftime("%H:%M:%S", time.localtime(entry["timestamp"]))
# Format function and args
function_name = entry["function"]
args_str = ", ".join(f"{k}={v!r}" for k, v in entry["args"].items())
# Format status
status = "✅ Success" if entry.get("success", False) else "❌ Failed"
# Print the entry
print(f"{i}. [{timestamp}] {function_name}({args_str}) -> {status}")
# Show additional details for errors
if not entry.get("success", False) and "error" in entry:
print(f" Error: {entry['error']}")
# Show additional details for commands
if function_name == "execute_command" and "returncode" in entry:
print(f" Exit code: {entry['returncode']}")
# Show additional details for search
if function_name == "opensearch_query" and "num_results" in entry:
print(f" Results: {entry['num_results']}")
print("=" * 60)
def get_directory_structure(self, path: str = '.', max_depth: int = 2) -> Dict[str, Any]:
"""
Get a formatted directory structure.
Args:
path: Starting directory path
max_depth: Maximum directory depth to show
Returns:
dict: Directory structure information
"""
try:
self.print_status(f"Getting directory structure for: {path} (depth: {max_depth})")
# Execute find command to get the structure
from ollamatools.plugins.tools.system import execute_command
result = execute_command(
command="find",
args=[path, "-maxdepth", str(max_depth), "-type", "f", "-o", "-type", "d"],
permission_manager=self.agent.permission_manager
)
if "error" in result:
return result
# Process the output into a more structured format
lines = result.get("stdout", "").strip().split('\n')
structure = {
"path": path,
"depth": max_depth,
"entries": lines,
"files": [line for line in lines if os.path.isfile(line)],
"directories": [line for line in lines if os.path.isdir(line)],
"count": {
"total": len(lines),
"files": len([line for line in lines if os.path.isfile(line)]),
"directories": len([line for line in lines if os.path.isdir(line)])
}
}
self.print_success(f"Found {structure['count']['total']} entries "
f"({structure['count']['files']} files, "
f"{structure['count']['directories']} directories)")
return structure
except Exception as e:
self.print_error(f"Failed to get directory structure: {str(e)}")
self.suggest_fix(self._generate_error_fix_suggestion(e))
return {
"error": str(e),
"path": path,
"depth": max_depth
}
def _generate_error_fix_suggestion(self, error: Exception) -> str:
"""
Generate a fix suggestion based on the error type.
Args:
error: The exception
Returns:
str: Suggested fix
"""
if isinstance(error, CommandError):
return self._generate_command_fix_suggestion(error)
elif isinstance(error, FileError):
return self._generate_file_fix_suggestion(error)
elif isinstance(error, WebError):
return self._generate_web_fix_suggestion(error)
else:
# Generic suggestion
return f"Try a different approach or check if the required resources are available."
def _generate_command_fix_suggestion(self, error: Union[CommandError, Dict[str, Any]]) -> str:
"""
Generate a fix suggestion for command errors.
Args:
error: Error object or dictionary
Returns:
str: Suggested fix
"""
# Extract error details
if isinstance(error, CommandError):
error_msg = str(error)
details = error.details or {}
command = details.get('command', '')
stderr = details.get('stderr', '')
else:
error_msg = error.get('error', '')
command = error.get('command', '')
stderr = error.get('stderr', '')
# Check for common error patterns
if "not in allowed command list" in error_msg:
return (f"The command '{command}' is not in the allowed command list. "
f"Try using one of these commands instead: {', '.join(sorted(list(self.command_allowlist)[:10]))}... "
f"or add this command to the allowlist when initializing the agent.")
elif "dangerous pattern" in error_msg:
pattern = error_msg.split("pattern matching '")[1].split("'")[0] if "pattern matching '" in error_msg else ""
return (f"The command contains potentially dangerous pattern: {pattern}. "
f"Try reformulating the command to avoid shell metacharacters.")
elif "not found" in error_msg or "command not found" in stderr:
return f"The command '{command}' does not exist on this system. Check the spelling or install the command."
elif "permission denied" in stderr.lower():
return f"Permission denied when executing '{command}'. Try running with appropriate permissions."
else:
return f"Try a different command syntax or check command documentation."
def _generate_file_fix_suggestion(self, error: FileError) -> str:
"""
Generate a fix suggestion for file errors.
Args:
error: File error
Returns:
str: Suggested fix
"""
error_msg = str(error)
details = error.details or {}
path = details.get('path', '')
if "not found" in error_msg:
return f"The file '{path}' does not exist. Check the path or create the file."
elif "is a directory" in error_msg:
return f"'{path}' is a directory, not a file. Try specifying a file path instead."
elif "permission denied" in error_msg.lower():
return f"Permission denied when accessing '{path}'. Check file permissions."
elif "exceeds maximum size" in error_msg:
return f"The file '{path}' is too large. Try reading a smaller file or increasing max_size."
elif "outside the allowed directories" in error_msg:
return f"The path '{path}' is outside the allowed directories. Try a path within the sandbox."
else:
return f"Check if the file exists and you have appropriate permissions."
def _generate_web_fix_suggestion(self, error: WebError) -> str:
"""
Generate a fix suggestion for web errors.
Args:
error: Web error
Returns:
str: Suggested fix
"""
error_msg = str(error)
details = error.details or {}
url = details.get('url', '')
if "domain" in error_msg and "blocked" in error_msg:
return f"The domain in '{url}' is blocked for security reasons. Try a different domain."
elif "scheme" in error_msg and "not allowed" in error_msg:
return f"The URL scheme is not allowed. Use http:// or https:// instead."
elif "Connection refused" in error_msg:
return f"Could not connect to '{url}'. Check if the service is running and accessible."
elif "timeout" in error_msg.lower():
return f"The request to '{url}' timed out. Try again later or check your internet connection."
elif "Status code:" in error_msg and "404" in error_msg:
return f"The resource at '{url}' was not found (404). Check the URL for typos."
elif "Status code:" in error_msg and "403" in error_msg:
return f"Access to '{url}' is forbidden (403). The site may require authentication."
else:
return f"Try a different URL or check your internet connection."
def detect_command_intent(self, message: str) -> Tuple[bool, Optional[str]]:
"""
Detect if a message is asking for system information that requires a command.
Args:
message: User message
Returns:
Tuple[bool, Optional[str]]: (has_command_intent, suggested_command)
"""
return detect_command_intent(message, self.function_feedback)
def provide_function_feedback(self, query: str, correct_function: str, actual_function: str = None) -> bool:
"""
Allow users to provide feedback on function selection.
Args:
query: The original query
correct_function: The function that should have been used
actual_function: The function that was actually used (if known)
Returns:
bool: True if feedback was recorded successfully
"""
if actual_function:
# Record explicit feedback
self.function_feedback.record_explicit_feedback(
query, correct_function, actual_function, user_correction=True
)
self.print_success(f"Thank you for your feedback! I'll remember to use {correct_function} for similar queries in the future.")
return True
else:
# Try to find the most recent function used for this query
for entry in reversed(self.execution_history):
if entry.get("query", "") == query:
actual_function = entry.get("function")
if actual_function and actual_function != correct_function:
self.function_feedback.record_explicit_feedback(
query, correct_function, actual_function, user_correction=True
)
self.print_success(f"Thank you for your feedback! I'll remember to use {correct_function} for similar queries in the future.")
return True
# If we couldn't find a matching entry, just record the correct function
self.function_feedback.record_explicit_feedback(
query, correct_function, "unknown", user_correction=True
)
self.print_success(f"Thank you for your feedback! I'll remember to use {correct_function} for similar queries in the future.")
return True
def chat(self, message: str) -> str:
"""
Process a chat message, showing the execution steps transparently.
Args:
message: User message
Returns:
str: Response message
"""
try:
# First check if the message is asking for system information
has_command_intent, suggested_command = self.detect_command_intent(message)
if has_command_intent and suggested_command:
self.print_info(f"This appears to be a question about system information. Using '{suggested_command}' command...")
try:
# Execute the command
result = execute_command(
command=suggested_command,
permission_manager=self.agent.permission_manager
)
# Record this in execution history
self.execution_history.append({
"function": "execute_command",
"args": {"command": suggested_command},
"timestamp": time.time(),
"success": "error" not in result,
"returncode": result.get("returncode"),
"query": message
})
# Record implicit feedback
self.function_feedback.record_implicit_feedback(
message, "execute_command", 1.0, result
)
if "error" in result:
self.print_error(f"Command execution failed: {result['error']}")
return f"I tried to run the '{suggested_command}' command to answer your question, but encountered an error: {result['error']}"
else:
# Format the response based on the command
formatted_result = format_command_result(suggested_command, result)
return formatted_result
except Exception as e:
self.print_error(f"Command execution failed: {str(e)}")
return f"I tried to run the '{suggested_command}' command to answer your question, but encountered an error: {str(e)}"
# If not a command, check if the message likely requires a search but doesn't explicitly ask for one
if not message.lower().startswith('search '):
search_indicators = [
"what is", "how to", "who is", "where can", "when did", "why does",
"recipe for", "information about", "tell me about", "find", "look up",
"latest", "news on", "current", "recent"
]
# Check if the message contains search indicators
needs_search = any(indicator in message.lower() for indicator in search_indicators)
# Check for question structure
question_structure = message.lower().startswith(("what", "how", "who", "where", "when", "why", "can", "could"))
if needs_search or question_structure:
# Check if message is asking for external information we likely don't have
factual_intent = self.detect_factual_intent(message)
if factual_intent:
self.print_info("This appears to be a question that might benefit from search. Searching automatically...")
try:
results = self.execute_search(message)
# Process search results before continuing with normal chat flow
context = self.extract_search_context(results)
# Add context to the user message
message = f"{message}\n\nSearch context: {context}"
# Record implicit feedback
self.function_feedback.record_implicit_feedback(
message, "opensearch_query", 0.9, results
)
except Exception as e:
self.print_error(f"Automatic search failed: {str(e)}")
self.print_info("Continuing without search results")
# Print what we're doing
self.print_status(f"Processing message: {message}")
# Start timer for perf measurement
start_time = time.time()
# Process LLM response
try:
response = self.agent.chat(
message=message,
system_prompt=self.system_prompt,
context=self.context
)
# Check if the response looks like a JSON function call
if response.strip().startswith('{') and ('function' in response or 'name' in response):
self.print_info("Detected potential function call in response")
try:
# Parse the JSON response
func_data = json.loads(response)
# Extract function name and args
function_name = None
args = {}
# Handle different JSON formats
if 'function' in func_data:
function_name = func_data['function'].get('name')
args_str = func_data['function'].get('arguments', '{}')
if isinstance(args_str, str):
args = json.loads(args_str)
else:
args = args_str
elif 'name' in func_data:
function_name = func_data.get('name')
args = func_data.get('arguments', {})
if function_name:
self.print_status(f"Executing function: {function_name} with args: {args}")
# Execute the function
if function_name == 'opensearch_query' or function_name == 'search_web':
results = self.execute_search(args.get('query', message))
context = self.extract_search_context(results)
response = f"Here's what I found about '{args.get('query', message)}':\n\n{context}"
elif hasattr(self, function_name) and callable(getattr(self, function_name)):
# Execute a method of this class
method = getattr(self, function_name)
result = method(**args)
response = f"Here are the results for '{function_name}':\n\n{result}"
elif function_name in self.agent.registered_functions:
# Execute a registered function
result = self.agent.registered_functions[function_name](**args)
response = f"Here are the results for '{function_name}':\n\n{result}"
else:
self.print_error(f"Function '{function_name}' not found")
response = f"I tried to call a function '{function_name}' but it's not available. I'll try to help you directly instead."
except json.JSONDecodeError:
self.print_info("Response looked like JSON but couldn't be parsed as a function call")
except Exception as e:
self.print_error(f"Error executing function from response: {str(e)}")
response = f"I tried to execute a function to help with your request, but encountered an error: {str(e)}"
except Exception as e:
self.print_error(f"Error communicating with Ollama: {str(e)}")
response = f"I encountered an error communicating with the language model: {str(e)}"
# Update context
self.context.append({"role": "user", "content": message})
self.context.append({"role": "assistant", "content": response})
# Limit context size
if len(self.context) > 10:
self.context = self.context[-10:]
# Calculate execution time
execution_time = time.time() - start_time
self.print_success(f"Response generated in {execution_time:.2f} seconds")
return response
except OllamaToolsError as e:
error_details = handle_exception(e, log_error=True)
self.print_error(f"Error: {error_details['error_message']}")
self.suggest_fix(self._generate_error_fix_suggestion(e))
return f"I encountered an error: {error_details['error_message']}\n\n" + \
f"Suggested fix: {self._generate_error_fix_suggestion(e)}"
except Exception as e:
error_details = handle_exception(e, log_error=True)
self.print_error(f"Unexpected error: {str(e)}")
return f"I encountered an unexpected error: {str(e)}"