INTRODUCTION AND MOTIVATION
The evolution of command-line interfaces has reached an inflection point with the advent of Large Language Models. Traditional Unix shells excel at precise command execution but struggle with discoverability and user-friendly interaction. By integrating LLMs into shell architecture, we can create a hybrid system that maintains the power and precision of traditional shells while adding natural language understanding capabilities.
This article presents a comprehensive approach to building an LLM-integrated Unix shell that seamlessly combines traditional command execution with intelligent natural language processing. The resulting system allows users to execute standard Unix commands alongside natural language queries, receive intelligent suggestions, and benefit from contextual assistance.
ARCHITECTURAL FOUNDATION
The foundation of an LLM-integrated shell rests on a modular architecture that separates concerns while maintaining tight integration between components. The system consists of five primary layers: the presentation layer handles user interaction, the command processing layer manages input parsing and routing, the execution layer handles both traditional commands and LLM interactions, the context management layer maintains conversation state, and the storage layer manages history and configuration.
+------------------+
| Presentation | <- User Interface and I/O
+------------------+
| Command Proc. | <- Input parsing and routing
+------------------+
| Execution | <- Command and LLM execution
+------------------+
| Context Mgmt. | <- State and conversation management
+------------------+
| Storage | <- History and configuration
+------------------+
This layered approach ensures that each component can be developed, tested, and maintained independently while supporting the complex interactions required for LLM integration.
CORE SHELL COMPONENTS
The fundamental shell components form the backbone of our LLM-integrated system. The input parser must distinguish between traditional commands and natural language queries while maintaining compatibility with existing shell scripts and command structures.
The command parser implementation begins with a tokenizer that handles both structured commands and free-form text:
class CommandParser:
def __init__(self):
self.llm_triggers = ['ask', 'explain', 'help', '?']
self.command_history = []
def parse_input(self, user_input):
"""Parse user input and determine processing strategy"""
tokens = self.tokenize(user_input.strip())
if self.is_llm_query(tokens):
return self.create_llm_command(tokens, user_input)
else:
return self.create_unix_command(tokens)
def is_llm_query(self, tokens):
"""Determine if input should be processed as LLM query"""
if not tokens:
return False
# Check for explicit LLM triggers
if tokens[0] in self.llm_triggers:
return True
# Check for natural language patterns
return self.contains_natural_language(tokens)
The parser must intelligently differentiate between commands intended for traditional execution and those requiring LLM processing. This differentiation relies on pattern recognition, explicit triggers, and contextual analysis of the input structure.
The command execution engine maintains separate pathways for Unix commands and LLM interactions while providing a unified interface for result handling:
class CommandExecutor:
def __init__(self, llm_client, shell_environment):
self.llm_client = llm_client
self.shell_env = shell_environment
self.execution_context = ExecutionContext()
def execute_command(self, parsed_command):
"""Execute parsed command through appropriate pathway"""
try:
if parsed_command.type == CommandType.UNIX:
return self.execute_unix_command(parsed_command)
elif parsed_command.type == CommandType.LLM:
return self.execute_llm_query(parsed_command)
else:
return self.execute_hybrid_command(parsed_command)
except Exception as e:
return self.handle_execution_error(e, parsed_command)
The execution context maintains state information that allows the LLM to understand the current shell environment, recent command history, and user preferences. This context proves crucial for providing relevant and accurate responses.
LLM INTEGRATION LAYER
The LLM integration layer provides abstraction over different language model providers while maintaining consistent functionality across the shell system. This abstraction allows the shell to work with various LLM services without requiring changes to the core shell logic.
The LLM client implementation supports multiple providers through a common interface:
class LLMClient:
def __init__(self, provider_config):
self.provider = self.initialize_provider(provider_config)
self.conversation_manager = ConversationManager()
self.response_cache = ResponseCache()
def process_query(self, query, context):
"""Process natural language query with full context"""
# Check cache first for performance
cache_key = self.generate_cache_key(query, context)
cached_response = self.response_cache.get(cache_key)
if cached_response:
return cached_response
# Prepare context for LLM
enriched_context = self.enrich_context(context)
# Execute LLM query
response = self.provider.complete_chat(
messages=self.build_message_history(query, enriched_context),
temperature=0.7,
max_tokens=1000
)
# Cache and return response
processed_response = self.process_response(response)
self.response_cache.set(cache_key, processed_response)
return processed_response
Context enrichment plays a vital role in providing the LLM with sufficient information to generate helpful responses. The enrichment process includes current working directory, recent command history, system information, and user preferences.
def enrich_context(self, base_context):
"""Enrich context with shell-specific information"""
enriched = base_context.copy()
# Add current shell state
enriched['current_directory'] = os.getcwd()
enriched['environment_variables'] = dict(os.environ)
enriched['recent_commands'] = self.get_recent_commands(10)
# Add system information
enriched['system_info'] = {
'os': platform.system(),
'architecture': platform.machine(),
'python_version': platform.python_version()
}
# Add user preferences and shell configuration
enriched['user_preferences'] = self.load_user_preferences()
return enriched
The conversation manager maintains dialogue state across multiple interactions, allowing the LLM to reference previous exchanges and maintain context continuity throughout the session.
COMMAND PROCESSING PIPELINE
The command processing pipeline orchestrates the flow of user input through parsing, context enrichment, execution, and result presentation. This pipeline must handle both synchronous Unix command execution and asynchronous LLM processing while maintaining responsive user interaction.
The pipeline controller manages the complete processing workflow:
class ProcessingPipeline:
def __init__(self, parser, executor, context_manager):
self.parser = parser
self.executor = executor
self.context_manager = context_manager
self.middleware_stack = []
def process_user_input(self, user_input):
"""Process user input through complete pipeline"""
# Stage 1: Parse and classify input
parsed_command = self.parser.parse_input(user_input)
# Stage 2: Enrich with context
enriched_command = self.context_manager.enrich_command(parsed_command)
# Stage 3: Apply middleware (logging, security, etc.)
processed_command = self.apply_middleware(enriched_command)
# Stage 4: Execute command
execution_result = self.executor.execute_command(processed_command)
# Stage 5: Process and format result
formatted_result = self.format_result(execution_result)
# Stage 6: Update context and history
self.context_manager.update_context(
processed_command,
execution_result
)
return formatted_result
Middleware components provide cross-cutting concerns such as security validation, command logging, performance monitoring, and error handling. The middleware stack allows for flexible extension of pipeline functionality without modifying core processing logic.
class SecurityMiddleware:
def __init__(self, security_config):
self.allowed_commands = security_config.get('allowed_commands', [])
self.blocked_patterns = security_config.get('blocked_patterns', [])
def process_command(self, command, next_handler):
"""Apply security validation to command"""
if self.is_command_allowed(command):
if not self.contains_blocked_pattern(command):
return next_handler(command)
else:
raise SecurityException("Command contains blocked pattern")
else:
raise SecurityException("Command not in allowed list")
Context management requires sophisticated state tracking to maintain useful conversation history while preventing context pollution. The context manager maintains separate contexts for different types of interactions.
INPUT/OUTPUT HANDLING AND USER INTERACTION
The input/output handling system must provide a seamless user experience that accommodates both traditional shell interaction patterns and the more conversational nature of LLM interactions. This requires careful attention to prompt design, result formatting, and interactive features.
The interactive shell controller manages the main user interaction loop:
class InteractiveShell:
def __init__(self, pipeline, config):
self.pipeline = pipeline
self.config = config
self.prompt_manager = PromptManager()
self.output_formatter = OutputFormatter()
self.running = False
def start_interactive_session(self):
"""Start main interactive shell loop"""
self.running = True
self.display_welcome_message()
while self.running:
try:
# Get user input with dynamic prompt
prompt = self.prompt_manager.generate_prompt()
user_input = input(prompt)
if self.is_exit_command(user_input):
break
# Process input through pipeline
result = self.pipeline.process_user_input(user_input)
# Format and display result
formatted_output = self.output_formatter.format_result(result)
print(formatted_output)
except KeyboardInterrupt:
self.handle_interrupt()
except Exception as e:
self.handle_error(e)
self.display_goodbye_message()
The prompt manager creates dynamic prompts that reflect current shell state and provide visual cues about available functionality:
class PromptManager:
def __init__(self):
self.prompt_style = PromptStyle.ENHANCED
self.show_context_info = True
def generate_prompt(self):
"""Generate dynamic prompt based on current state"""
components = []
# Add current directory
cwd = os.path.basename(os.getcwd())
components.append(f"[{cwd}]")
# Add LLM status indicator
if self.llm_available():
components.append("🤖")
# Add git branch if in git repository
git_branch = self.get_git_branch()
if git_branch:
components.append(f"({git_branch})")
# Construct final prompt
prompt_prefix = " ".join(components)
return f"{prompt_prefix} $ "
Result formatting must handle diverse output types from simple command results to complex LLM responses. The formatter provides consistent presentation while preserving important formatting and structure.
SECURITY CONSIDERATIONS AND IMPLEMENTATION
Security represents a critical concern when integrating LLMs into shell environments. The system must protect against prompt injection attacks, unauthorized command execution, and data leakage while maintaining functionality and usability.
Command sanitization and validation form the first line of defense:
class SecurityValidator:
def __init__(self, security_policy):
self.policy = security_policy
self.command_whitelist = self.load_command_whitelist()
self.pattern_blacklist = self.load_pattern_blacklist()
def validate_command(self, command):
"""Comprehensive command security validation"""
validation_result = ValidationResult()
# Check against command whitelist
if not self.is_whitelisted_command(command):
validation_result.add_violation(
"Command not in whitelist",
SecurityLevel.HIGH
)
# Scan for dangerous patterns
dangerous_patterns = self.scan_for_dangerous_patterns(command)
for pattern in dangerous_patterns:
validation_result.add_violation(
f"Contains dangerous pattern: {pattern}",
SecurityLevel.CRITICAL
)
# Validate LLM query safety
if command.type == CommandType.LLM:
llm_safety = self.validate_llm_query_safety(command)
validation_result.merge(llm_safety)
return validation_result
LLM-specific security measures must address prompt injection attempts and ensure that the language model cannot be manipulated into executing unauthorized actions:
def validate_llm_query_safety(self, llm_command):
"""Validate LLM query for security threats"""
safety_result = ValidationResult()
query_text = llm_command.query_text
# Check for prompt injection patterns
injection_patterns = [
r"ignore.*previous.*instructions",
r"system.*role.*admin",
r"execute.*command.*as.*root"
]
for pattern in injection_patterns:
if re.search(pattern, query_text, re.IGNORECASE):
safety_result.add_violation(
f"Potential prompt injection: {pattern}",
SecurityLevel.CRITICAL
)
# Validate query length and complexity
if len(query_text) > self.policy.max_query_length:
safety_result.add_violation(
"Query exceeds maximum length",
SecurityLevel.MEDIUM
)
return safety_result
Data privacy protection ensures that sensitive information from the user’s environment is not inadvertently sent to external LLM services:
class PrivacyFilter:
def __init__(self, privacy_config):
self.sensitive_patterns = privacy_config.get('sensitive_patterns', [])
self.environment_filters = privacy_config.get('env_filters', [])
def filter_context(self, context):
"""Remove sensitive information from context"""
filtered_context = context.copy()
# Filter environment variables
if 'environment_variables' in filtered_context:
filtered_env = {}
for key, value in filtered_context['environment_variables'].items():
if not self.is_sensitive_env_var(key):
filtered_env[key] = self.sanitize_value(value)
filtered_context['environment_variables'] = filtered_env
# Filter command history
if 'recent_commands' in filtered_context:
filtered_commands = []
for cmd in filtered_context['recent_commands']:
sanitized_cmd = self.sanitize_command(cmd)
if sanitized_cmd:
filtered_commands.append(sanitized_cmd)
filtered_context['recent_commands'] = filtered_commands
return filtered_context
PERFORMANCE OPTIMIZATION STRATEGIES
Performance optimization in an LLM-integrated shell requires careful attention to both traditional shell responsiveness and the inherently slower nature of language model interactions. The system must maintain snappy response times for traditional commands while providing smooth user experience for LLM operations.
Caching strategies provide significant performance improvements for repeated queries:
class IntelligentCache:
def __init__(self, cache_config):
self.memory_cache = {}
self.persistent_cache = self.initialize_persistent_storage(cache_config)
self.cache_ttl = cache_config.get('ttl_seconds', 3600)
def get_cached_response(self, query, context_hash):
"""Retrieve cached response if available and valid"""
cache_key = self.generate_cache_key(query, context_hash)
# Check memory cache first
if cache_key in self.memory_cache:
cached_item = self.memory_cache[cache_key]
if self.is_cache_valid(cached_item):
return cached_item['response']
# Check persistent cache
persistent_item = self.persistent_cache.get(cache_key)
if persistent_item and self.is_cache_valid(persistent_item):
# Promote to memory cache
self.memory_cache[cache_key] = persistent_item
return persistent_item['response']
return None
def cache_response(self, query, context_hash, response):
"""Cache response with appropriate TTL"""
cache_key = self.generate_cache_key(query, context_hash)
cache_item = {
'response': response,
'timestamp': time.time(),
'context_hash': context_hash
}
self.memory_cache[cache_key] = cache_item
self.persistent_cache.set(cache_key, cache_item)
Asynchronous processing allows the shell to remain responsive while processing LLM queries:
class AsynchronousExecutor:
def __init__(self, thread_pool_size=4):
self.executor = ThreadPoolExecutor(max_workers=thread_pool_size)
self.active_queries = {}
def execute_llm_query_async(self, query, context, callback):
"""Execute LLM query asynchronously with callback"""
query_id = self.generate_query_id()
future = self.executor.submit(
self.process_llm_query,
query,
context
)
future.add_done_callback(
lambda f: self.handle_query_completion(f, query_id, callback)
)
self.active_queries[query_id] = {
'future': future,
'query': query,
'start_time': time.time()
}
return query_id
The system provides progress indicators and allows users to continue with other tasks while LLM processing occurs in the background.
COMPLETE RUNNING EXAMPLE - AISHELL IMPLEMENTATION
The following complete implementation demonstrates all concepts discussed in this article. This working example provides a fully functional LLM-integrated shell that can execute both traditional Unix commands and natural language queries.
#!/usr/bin/env python3
“””
AIShell - A Unix shell with integrated Large Language Model capabilities
Complete implementation demonstrating LLM-shell integration concepts
“””
import os
import sys
import re
import time
import json
import subprocess
import threading
import platform
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from enum import Enum
from concurrent.futures import ThreadPoolExecutor
import readline # For command history and editing
class CommandType(Enum):
UNIX = “unix”
LLM = “llm”
HYBRID = “hybrid”
class SecurityLevel(Enum):
LOW = 1
MEDIUM = 2
HIGH = 3
CRITICAL = 4
@dataclass
class ParsedCommand:
“”“Represents a parsed command with metadata”””
type: CommandType
raw_input: str
tokens: List[str]
command_name: str
arguments: List[str]
query_text: Optional[str] = None
needs_llm: bool = False
@dataclass
class ExecutionResult:
“”“Represents the result of command execution”””
success: bool
output: str
error: str
execution_time: float
command_type: CommandType
@dataclass
class ValidationResult:
“”“Security validation result”””
is_valid: bool = True
violations: List[Dict[str, Any]] = None
```
def __post_init__(self):
if self.violations is None:
self.violations = []
def add_violation(self, message: str, level: SecurityLevel):
"""Add security violation"""
self.violations.append({
'message': message,
'level': level,
'timestamp': time.time()
})
if level in [SecurityLevel.HIGH, SecurityLevel.CRITICAL]:
self.is_valid = False
```
class MockLLMProvider:
“”“Mock LLM provider for demonstration purposes”””
```
def complete_chat(self, messages: List[Dict], temperature: float = 0.7, max_tokens: int = 1000) -> Dict:
"""Simulate LLM response generation"""
time.sleep(0.5) # Simulate network delay
last_message = messages[-1]['content'] if messages else ""
# Simple response generation based on query patterns
if 'list files' in last_message.lower():
response = "You can list files using 'ls' command. For detailed listing, use 'ls -la'."
elif 'current directory' in last_message.lower():
response = f"You are currently in: {os.getcwd()}"
elif 'help' in last_message.lower():
response = "I can help you with Unix commands and system information. Try asking about files, directories, or specific commands."
else:
response = f"I understand you're asking: '{last_message}'. In a real implementation, this would be processed by an actual LLM."
return {
'choices': [{'message': {'content': response}}],
'usage': {'total_tokens': len(response.split())}
}
```
class SecurityValidator:
“”“Security validation for commands and queries”””
```
def __init__(self):
self.dangerous_commands = ['rm -rf /', 'sudo rm -rf', 'mkfs', 'dd if=']
self.injection_patterns = [
r'ignore.*previous.*instructions',
r'system.*role.*admin',
r'execute.*command.*as.*root'
]
def validate_command(self, command: ParsedCommand) -> ValidationResult:
"""Validate command for security threats"""
result = ValidationResult()
# Check for dangerous Unix commands
if command.type == CommandType.UNIX:
cmd_string = ' '.join([command.command_name] + command.arguments)
for dangerous_cmd in self.dangerous_commands:
if dangerous_cmd in cmd_string:
result.add_violation(
f"Dangerous command detected: {dangerous_cmd}",
SecurityLevel.CRITICAL
)
# Check for LLM prompt injection
if command.type == CommandType.LLM and command.query_text:
for pattern in self.injection_patterns:
if re.search(pattern, command.query_text, re.IGNORECASE):
result.add_violation(
f"Potential prompt injection: {pattern}",
SecurityLevel.HIGH
)
return result
```
class ResponseCache:
“”“Simple in-memory cache for LLM responses”””
```
def __init__(self, ttl_seconds: int = 3600):
self.cache = {}
self.ttl = ttl_seconds
def get(self, key: str) -> Optional[Any]:
"""Get cached item if valid"""
if key in self.cache:
item = self.cache[key]
if time.time() - item['timestamp'] < self.ttl:
return item['data']
else:
del self.cache[key]
return None
def set(self, key: str, data: Any):
"""Cache data with timestamp"""
self.cache[key] = {
'data': data,
'timestamp': time.time()
}
```
class ConversationManager:
“”“Manages conversation context and history”””
```
def __init__(self):
self.conversation_history = []
self.max_history_length = 10
def add_exchange(self, user_input: str, assistant_response: str):
"""Add conversation exchange to history"""
self.conversation_history.append({
'user': user_input,
'assistant': assistant_response,
'timestamp': time.time()
})
# Trim history if too long
if len(self.conversation_history) > self.max_history_length:
self.conversation_history = self.conversation_history[-self.max_history_length:]
def get_conversation_context(self) -> List[Dict]:
"""Get conversation history formatted for LLM"""
context = []
for exchange in self.conversation_history:
context.append({'role': 'user', 'content': exchange['user']})
context.append({'role': 'assistant', 'content': exchange['assistant']})
return context
```
class LLMClient:
“”“Client for LLM integration with caching and context management”””
```
def __init__(self):
self.provider = MockLLMProvider()
self.conversation_manager = ConversationManager()
self.cache = ResponseCache()
def process_query(self, query: str, context: Dict) -> str:
"""Process natural language query with context"""
# Generate cache key
cache_key = f"{query}:{hash(str(context))}"
# Check cache first
cached_response = self.cache.get(cache_key)
if cached_response:
return cached_response
# Prepare messages for LLM
messages = self.build_message_history(query, context)
# Call LLM provider
response = self.provider.complete_chat(messages, temperature=0.7)
# Extract response text
response_text = response['choices'][0]['message']['content']
# Cache response
self.cache.set(cache_key, response_text)
# Update conversation history
self.conversation_manager.add_exchange(query, response_text)
return response_text
def build_message_history(self, query: str, context: Dict) -> List[Dict]:
"""Build message history including context"""
messages = [
{
'role': 'system',
'content': f'''You are an AI assistant integrated into a Unix shell.
Current working directory: {context.get('current_directory', 'unknown')}
Recent commands: {context.get('recent_commands', [])}
System: {context.get('system_info', {})}
Provide helpful, concise responses about Unix commands and system administration.'''
}
]
# Add conversation history
messages.extend(self.conversation_manager.get_conversation_context())
# Add current query
messages.append({'role': 'user', 'content': query})
return messages
```
class CommandParser:
“”“Parser for both Unix commands and natural language queries”””
```
def __init__(self):
self.llm_triggers = ['ask', 'explain', 'help', '?', 'what', 'how', 'why']
def parse_input(self, user_input: str) -> ParsedCommand:
"""Parse user input and determine command type"""
tokens = user_input.strip().split()
if not tokens:
return ParsedCommand(
type=CommandType.UNIX,
raw_input=user_input,
tokens=tokens,
command_name="",
arguments=[]
)
# Check if this is an LLM query
if self.is_llm_query(tokens, user_input):
return ParsedCommand(
type=CommandType.LLM,
raw_input=user_input,
tokens=tokens,
command_name="llm_query",
arguments=[],
query_text=user_input,
needs_llm=True
)
# Parse as Unix command
command_name = tokens[0]
arguments = tokens[1:] if len(tokens) > 1 else []
return ParsedCommand(
type=CommandType.UNIX,
raw_input=user_input,
tokens=tokens,
command_name=command_name,
arguments=arguments
)
def is_llm_query(self, tokens: List[str], full_input: str) -> bool:
"""Determine if input should be processed as LLM query"""
if not tokens:
return False
# Check for explicit triggers
if tokens[0].lower() in self.llm_triggers:
return True
# Check for question patterns
if full_input.strip().endswith('?'):
return True
# Check for natural language patterns
natural_indicators = ['can you', 'could you', 'please', 'i need', 'i want']
full_lower = full_input.lower()
return any(indicator in full_lower for indicator in natural_indicators)
```
class CommandExecutor:
“”“Executes both Unix commands and LLM queries”””
```
def __init__(self, llm_client: LLMClient):
self.llm_client = llm_client
self.command_history = []
def execute_command(self, command: ParsedCommand) -> ExecutionResult:
"""Execute parsed command through appropriate pathway"""
start_time = time.time()
try:
if command.type == CommandType.LLM:
result = self.execute_llm_query(command)
else:
result = self.execute_unix_command(command)
execution_time = time.time() - start_time
result.execution_time = execution_time
# Update command history
self.command_history.append({
'command': command.raw_input,
'timestamp': time.time(),
'success': result.success
})
return result
except Exception as e:
return ExecutionResult(
success=False,
output="",
error=str(e),
execution_time=time.time() - start_time,
command_type=command.type
)
def execute_unix_command(self, command: ParsedCommand) -> ExecutionResult:
"""Execute Unix command using subprocess"""
if not command.command_name:
return ExecutionResult(
success=True,
output="",
error="",
execution_time=0,
command_type=CommandType.UNIX
)
try:
# Handle built-in commands
if command.command_name == 'cd':
return self.handle_cd_command(command.arguments)
elif command.command_name == 'exit':
sys.exit(0)
# Execute external command
cmd_args = [command.command_name] + command.arguments
process = subprocess.run(
cmd_args,
capture_output=True,
text=True,
timeout=30 # 30 second timeout
)
return ExecutionResult(
success=process.returncode == 0,
output=process.stdout,
error=process.stderr,
execution_time=0, # Will be set by caller
command_type=CommandType.UNIX
)
except subprocess.TimeoutExpired:
return ExecutionResult(
success=False,
output="",
error="Command timed out after 30 seconds",
execution_time=0,
command_type=CommandType.UNIX
)
except FileNotFoundError:
return ExecutionResult(
success=False,
output="",
error=f"Command not found: {command.command_name}",
execution_time=0,
command_type=CommandType.UNIX
)
def handle_cd_command(self, arguments: List[str]) -> ExecutionResult:
"""Handle built-in cd command"""
try:
if not arguments:
# cd with no arguments goes to home directory
os.chdir(os.path.expanduser('~'))
else:
os.chdir(arguments[0])
return ExecutionResult(
success=True,
output=f"Changed directory to: {os.getcwd()}",
error="",
execution_time=0,
command_type=CommandType.UNIX
)
except (FileNotFoundError, PermissionError) as e:
return ExecutionResult(
success=False,
output="",
error=str(e),
execution_time=0,
command_type=CommandType.UNIX
)
def execute_llm_query(self, command: ParsedCommand) -> ExecutionResult:
"""Execute LLM query with context"""
context = self.build_execution_context()
try:
response = self.llm_client.process_query(command.query_text, context)
return ExecutionResult(
success=True,
output=response,
error="",
execution_time=0, # Will be set by caller
command_type=CommandType.LLM
)
except Exception as e:
return ExecutionResult(
success=False,
output="",
error=f"LLM query failed: {str(e)}",
execution_time=0,
command_type=CommandType.LLM
)
def build_execution_context(self) -> Dict:
"""Build context information for LLM queries"""
return {
'current_directory': os.getcwd(),
'recent_commands': [cmd['command'] for cmd in self.command_history[-5:]],
'system_info': {
'os': platform.system(),
'architecture': platform.machine(),
'python_version': platform.python_version()
}
}
```
class OutputFormatter:
“”“Formats command output for display”””
```
def format_result(self, result: ExecutionResult) -> str:
"""Format execution result for display"""
if not result.success:
return f"Error: {result.error}"
if result.command_type == CommandType.LLM:
return f"🤖 {result.output}"
return result.output
```
class PromptManager:
“”“Manages shell prompt generation”””
```
def generate_prompt(self) -> str:
"""Generate dynamic prompt with current directory"""
cwd = os.path.basename(os.getcwd())
if not cwd:
cwd = "/"
return f"[{cwd}] 🤖 $ "
class ProcessingPipeline:
“”“Main processing pipeline for user input”””
```
def __init__(self):
self.parser = CommandParser()
self.llm_client = LLMClient()
self.executor = CommandExecutor(self.llm_client)
self.security_validator = SecurityValidator()
self.output_formatter = OutputFormatter()
def process_user_input(self, user_input: str) -> str:
"""Process user input through complete pipeline"""
# Stage 1: Parse input
parsed_command = self.parser.parse_input(user_input)
# Stage 2: Security validation
validation_result = self.security_validator.validate_command(parsed_command)
if not validation_result.is_valid:
violations = [v['message'] for v in validation_result.violations]
return f"Security violation: {'; '.join(violations)}"
# Stage 3: Execute command
execution_result = self.executor.execute_command(parsed_command)
# Stage 4: Format output
formatted_result = self.output_formatter.format_result(execution_result)
return formatted_result
```
class AIShell:
“”“Main AIShell application class”””
```
def __init__(self):
self.pipeline = ProcessingPipeline()
self.prompt_manager = PromptManager()
self.running = False
# Configure readline for command history
readline.set_history_length(1000)
def start_interactive_session(self):
"""Start main interactive shell loop"""
self.running = True
self.display_welcome_message()
while self.running:
try:
# Generate and display prompt
prompt = self.prompt_manager.generate_prompt()
user_input = input(prompt).strip()
if not user_input:
continue
if user_input.lower() in ['exit', 'quit', 'bye']:
break
# Process input
result = self.pipeline.process_user_input(user_input)
# Display result
if result:
print(result)
except KeyboardInterrupt:
print("\nUse 'exit' or 'quit' to leave the shell.")
except EOFError:
break
except Exception as e:
print(f"Unexpected error: {e}")
self.display_goodbye_message()
def display_welcome_message(self):
"""Display welcome message"""
print("=" * 60)
print("Welcome to AIShell - Unix Shell with LLM Integration")
print("=" * 60)
print("You can execute regular Unix commands or ask natural language questions.")
print("Examples:")
print(" ls -la (regular Unix command)")
print(" help me list files (natural language query)")
print(" what is my current directory? (question)")
print("Type 'exit' or 'quit' to leave.")
print("-" * 60)
def display_goodbye_message(self):
"""Display goodbye message"""
print("\nGoodbye! Thanks for using AIShell.")
```
def main():
“”“Main entry point for AIShell application”””
shell = AIShell()
shell.start_interactive_session()
if __name__ == “__main__’:
main()
This complete implementation demonstrates a working LLM-integrated shell that combines traditional Unix command execution with natural language processing capabilities. The system maintains security through validation mechanisms, provides performance optimization through caching, and offers a seamless user experience that bridges the gap between traditional shell interfaces and modern AI assistance.
The implementation showcases all major concepts discussed in this article while providing a foundation that can be extended with additional features such as more sophisticated LLM providers, enhanced security measures, and advanced context management capabilities.
