Introduction and Conceptual Overview
The modern software development workflow involves numerous repetitive tasks that consume valuable developer time. Managing Git repositories, configuring development environments, installing dependencies, and navigating complex IDEs are all necessary but often tedious activities. An LLM-powered development assistant that runs in the background and responds to natural language commands can dramatically improve developer productivity by automating these tasks while maintaining full control and transparency.
This article presents a comprehensive architecture for building such a development assistant. The system uses tool calling to translate natural language requests into concrete actions, supports both local and remote LLM backends across multiple GPU architectures, and implements an agentic or multi-agent design to handle complex workflows that require planning, execution, and verification.
The core concept revolves around a background service that monitors for a hotkey activation. When triggered, the assistant captures the current context including the active window, the focused application, and any relevant workspace information. The developer then issues a natural language command such as "create a new feature branch called user-authentication" or "install the Python extension for VS Code and configure it for my virtual environment." The LLM processes this request, determines which tools to invoke, generates the appropriate commands, and executes them while providing feedback to the user.
System Architecture and Components
The development assistant consists of several interconnected components that work together to provide seamless functionality. Understanding each component and how they interact is essential for building a robust system.
Background Service and Hotkey Listener
The background service runs continuously with minimal resource consumption, waiting for the developer to activate it via a configurable hotkey. This service must be cross-platform, supporting Windows, macOS, and Linux. On activation, it captures the current system state including the active window title, the process name of the focused application, and the current working directory if applicable.
The hotkey listener uses platform-specific APIs to register global keyboard shortcuts without interfering with normal application behavior. On Windows, this involves the RegisterHotKey API. On macOS, the Carbon framework or newer accessibility APIs handle global shortcuts. Linux systems typically use X11 or Wayland protocols depending on the display server.
Context Awareness System
Context awareness is critical for providing relevant assistance. The system must determine what the developer is currently working on by examining multiple signals. The active window title often contains file paths or project names. The focused application determines which tools are available. For example, if Visual Studio Code is active, the assistant can interact with its command palette and extension system. If a terminal window is focused, the assistant can execute shell commands directly.
The context system also maintains a workspace model that tracks open projects, their Git repository status, installed dependencies, and recent activities. This allows the assistant to provide intelligent suggestions and avoid redundant operations. For instance, if the developer asks to "initialize a Git repository," the system first checks whether the current directory
already contains a Git repository before executing the initialization command.
LLM Integration Layer with Multi-Backend Support
Supporting both local and remote LLMs with various GPU architectures requires a flexible abstraction layer. The system must handle different inference engines including llama.cpp for CPU and various GPU backends, vLLM for high-performance serving, Ollama for easy local deployment, and remote API providers like OpenAI or Anthropic.
For GPU support, the system detects the available hardware at startup and configures the appropriate backend. NVIDIA GPUs use CUDA through libraries like cuBLAS and cuDNN. AMD GPUs leverage ROCm, which provides HIP as a CUDA-compatible interface. Intel GPUs use oneAPI and the Intel Extension for PyTorch. Apple Silicon uses Metal Performance Shaders through the MPS backend in PyTorch.
The LLM integration layer provides a unified interface regardless of the backend. Each implementation handles model loading, tokenization, inference, and response parsing according to the specific backend's requirements. This abstraction allows the rest of the system to remain backend-agnostic.
Tool Calling Framework
Tool calling is the mechanism by which the LLM translates natural language into executable actions. The framework defines a set of tools, each with a clear specification including its name, description, parameters, and return type. When the LLM receives a user request, it analyzes the intent and generates one or more tool calls with appropriate arguments.
The tool specification uses a structured format that both the LLM and the execution engine can understand. Each tool has a JSON schema describing its parameters, validation rules, and expected behavior. The LLM generates tool calls in a standardized format, which the execution engine parses and validates before invocation.
Tools are organized into categories based on their functionality. Git tools handle repository operations like cloning, branching, committing, and pushing. IDE tools interact with Visual Studio Code or IntelliJ IDEA through their respective APIs or command-line interfaces. System tools manage file operations, process execution, and environment configuration. Package manager tools handle dependency installation using npm, pip, Maven, or other ecosystem-specific tools.
Agent Architecture for Complex Workflows
Some developer tasks require multiple steps with dependencies between them. For example, setting up a new Python project might involve creating a directory, initializing a Git repository, creating a virtual environment, installing dependencies from a requirements file, configuring VS Code with the Python extension, and opening the project. A simple tool calling approach would require the developer to issue multiple commands sequentially.
An agentic architecture allows the system to plan and execute multi-step workflows autonomously. The agent receives a high-level goal, breaks it down into subtasks, executes each subtask using the available tools, and adapts the plan based on intermediate results. This requires the agent to maintain state across multiple LLM invocations, track progress, and handle errors gracefully.
For even more complex scenarios, a multi-agent architecture distributes responsibilities among specialized agents. A planning agent analyzes the user request and creates an execution plan. A Git agent handles all repository operations. An IDE agent manages editor configuration and extensions. A validation agent checks that each step completed successfully before proceeding. These agents communicate through a message-passing system, coordinating their actions to achieve the overall goal.
Implementation Details and Code Examples
Now we examine the concrete implementation of each component with detailed code examples. The examples use Python for the core system due to its excellent library support for LLM integration, system automation, and cross-platform compatibility.
Setting Up the Background Service
The background service must run with minimal overhead and respond quickly when activated. We use a combination of threading for the hotkey listener and asyncio for the main event loop to handle concurrent operations efficiently.
import asyncio
import threading
from typing import Optional, Callable
import logging
class BackgroundService:
"""
Core background service that runs continuously and coordinates
all development assistant functionality.
"""
def __init__(self, hotkey_combination: str = "ctrl+shift+a"):
"""
Initialize the background service with a configurable hotkey.
Args:
hotkey_combination: String representation of the hotkey
(e.g., "ctrl+shift+a")
"""
self.hotkey = hotkey_combination
self.running = False
self.event_loop: Optional[asyncio.AbstractEventLoop] = None
self.activation_callback: Optional[Callable] = None
self.logger = logging.getLogger(__name__)
def set_activation_callback(self, callback: Callable):
"""
Register a callback function to be invoked when the hotkey
is pressed. The callback receives the current context.
Args:
callback: Async function that handles the activation event
"""
self.activation_callback = callback
def start(self):
"""
Start the background service in a separate thread to avoid
blocking the main application.
"""
if self.running:
self.logger.warning("Service already running")
return
self.running = True
service_thread = threading.Thread(target=self._run_service, daemon=True)
service_thread.start()
self.logger.info(f"Background service started with hotkey: {self.hotkey}")
def _run_service(self):
"""
Internal method that sets up the event loop and starts
the hotkey listener. Runs in a separate thread.
"""
self.event_loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.event_loop)
try:
self.event_loop.run_until_complete(self._async_main())
except Exception as e:
self.logger.error(f"Service error: {e}", exc_info=True)
finally:
self.event_loop.close()
async def _async_main(self):
"""
Main async loop that keeps the service running and handles
activation events.
"""
from .hotkey_listener import HotkeyListener
listener = HotkeyListener(self.hotkey)
listener.set_callback(self._handle_activation)
listener.start()
# Keep the service running
while self.running:
await asyncio.sleep(0.1)
def _handle_activation(self):
"""
Called when the hotkey is pressed. Captures context and
invokes the registered callback in the event loop.
"""
if self.activation_callback and self.event_loop:
asyncio.run_coroutine_threadsafe(
self._async_activation_handler(),
self.event_loop
)
async def _async_activation_handler(self):
"""
Async handler that captures context and invokes the callback.
"""
from .context_capture import ContextCapture
context = ContextCapture.get_current_context()
self.logger.info(f"Assistant activated in context: {context.app_name}")
if self.activation_callback:
await self.activation_callback(context)
def stop(self):
"""
Gracefully stop the background service.
"""
self.running = False
self.logger.info("Background service stopped")
The BackgroundService class provides the foundation for the entire system. It manages the lifecycle of the service, handles hotkey activation, and coordinates between the synchronous hotkey listener and the asynchronous main logic. The use of threading ensures that the hotkey listener remains responsive even when the LLM is processing a request.
Platform-Specific Hotkey Listening
Each operating system requires different approaches for global hotkey registration. We create a unified interface with platform-specific implementations.
import platform
from abc import ABC, abstractmethod
from typing import Callable, Optional
class HotkeyListenerBase(ABC):
"""
Abstract base class for platform-specific hotkey listeners.
"""
def __init__(self, hotkey: str):
"""
Initialize with a hotkey combination string.
Args:
hotkey: String like "ctrl+shift+a"
"""
self.hotkey = hotkey
self.callback: Optional[Callable] = None
self.modifiers, self.key = self._parse_hotkey(hotkey)
def _parse_hotkey(self, hotkey: str) -> tuple:
"""
Parse hotkey string into modifiers and key.
Returns:
Tuple of (modifiers_list, key_string)
"""
parts = hotkey.lower().split('+')
key = parts[-1]
modifiers = parts[:-1]
return modifiers, key
def set_callback(self, callback: Callable):
"""
Set the function to call when hotkey is pressed.
"""
self.callback = callback
@abstractmethod
def start(self):
"""
Start listening for the hotkey. Platform-specific implementation.
"""
pass
@abstractmethod
def stop(self):
"""
Stop listening for the hotkey.
"""
pass
class WindowsHotkeyListener(HotkeyListenerBase):
"""
Windows implementation using win32 API.
"""
def start(self):
"""
Register the hotkey using Windows RegisterHotKey API.
"""
import win32con
import win32api
import win32gui
# Map modifier strings to Windows constants
mod_map = {
'ctrl': win32con.MOD_CONTROL,
'shift': win32con.MOD_SHIFT,
'alt': win32con.MOD_ALT,
'win': win32con.MOD_WIN
}
# Combine modifiers
mod_flags = 0
for mod in self.modifiers:
if mod in mod_map:
mod_flags |= mod_map[mod]
# Get virtual key code
vk_code = ord(self.key.upper())
# Create a message-only window for receiving hotkey messages
wc = win32gui.WNDCLASS()
wc.lpfnWndProc = self._window_proc
wc.lpszClassName = "DevAssistantHotkey"
class_atom = win32gui.RegisterClass(wc)
self.hwnd = win32gui.CreateWindow(
class_atom, "DevAssistant", 0, 0, 0, 0, 0, 0, 0, 0, None
)
# Register the hotkey
hotkey_id = 1
if not win32gui.RegisterHotKey(self.hwnd, hotkey_id, mod_flags, vk_code):
raise RuntimeError("Failed to register hotkey")
# Start message loop in separate thread
import threading
self.msg_thread = threading.Thread(target=self._message_loop, daemon=True)
self.msg_thread.start()
def _window_proc(self, hwnd, msg, wparam, lparam):
"""
Window procedure that handles hotkey messages.
"""
import win32con
if msg == win32con.WM_HOTKEY:
if self.callback:
self.callback()
return 0
def _message_loop(self):
"""
Windows message loop that processes hotkey events.
"""
import win32gui
while True:
win32gui.PumpWaitingMessages()
def stop(self):
"""
Unregister the hotkey and destroy the window.
"""
import win32gui
win32gui.UnregisterHotKey(self.hwnd, 1)
win32gui.DestroyWindow(self.hwnd)
class MacOSHotkeyListener(HotkeyListenerBase):
"""
macOS implementation using Quartz and Cocoa frameworks.
"""
def start(self):
"""
Register global hotkey using macOS Carbon or Cocoa APIs.
"""
from Quartz import (
CGEventMaskBit, CGEventTapCreate, kCGEventKeyDown,
kCGHeadInsertEventTap, kCGSessionEventTap,
CFMachPortCreateRunLoopSource, CFRunLoopAddSource,
CFRunLoopGetCurrent, kCFRunLoopCommonModes, CGEventTapEnable
)
# Create event tap for keyboard events
event_mask = CGEventMaskBit(kCGEventKeyDown)
self.tap = CGEventTapCreate(
kCGSessionEventTap,
kCGHeadInsertEventTap,
0,
event_mask,
self._event_callback,
None
)
if not self.tap:
raise RuntimeError("Failed to create event tap")
# Add to run loop
run_loop_source = CFMachPortCreateRunLoopSource(None, self.tap, 0)
CFRunLoopAddSource(CFRunLoopGetCurrent(), run_loop_source, kCFRunLoopCommonModes)
CGEventTapEnable(self.tap, True)
def _event_callback(self, proxy, event_type, event, refcon):
"""
Callback for keyboard events. Checks if it matches our hotkey.
"""
from Quartz import CGEventGetIntegerValueField, kCGKeyboardEventKeycode
from Cocoa import NSEvent
# Get key code and modifiers
keycode = CGEventGetIntegerValueField(event, kCGKeyboardEventKeycode)
ns_event = NSEvent.eventWithCGEvent_(event)
modifiers = ns_event.modifierFlags()
# Check if this matches our hotkey
if self._matches_hotkey(keycode, modifiers):
if self.callback:
self.callback()
return event
def _matches_hotkey(self, keycode, modifiers):
"""
Check if the keycode and modifiers match our registered hotkey.
"""
from Cocoa import (
NSCommandKeyMask, NSShiftKeyMask, NSAlternateKeyMask, NSControlKeyMask
)
# Map our modifier strings to Cocoa constants
required_mods = 0
if 'cmd' in self.modifiers or 'command' in self.modifiers:
required_mods |= NSCommandKeyMask
if 'shift' in self.modifiers:
required_mods |= NSShiftKeyMask
if 'alt' in self.modifiers or 'option' in self.modifiers:
required_mods |= NSAlternateKeyMask
if 'ctrl' in self.modifiers or 'control' in self.modifiers:
required_mods |= NSControlKeyMask
# Check modifiers match
if (modifiers & required_mods) != required_mods:
return False
# Map key string to keycode
key_map = {
'a': 0, 'b': 11, 'c': 8, 'd': 2, 'e': 14, 'f': 3,
'g': 5, 'h': 4, 'i': 34, 'j': 38, 'k': 40, 'l': 37,
'm': 46, 'n': 45, 'o': 31, 'p': 35, 'q': 12, 'r': 15,
's': 1, 't': 17, 'u': 32, 'v': 9, 'w': 13, 'x': 7,
'y': 16, 'z': 6
}
expected_keycode = key_map.get(self.key.lower())
return keycode == expected_keycode
def stop(self):
"""
Disable and release the event tap.
"""
from Quartz import CGEventTapEnable
if self.tap:
CGEventTapEnable(self.tap, False)
class LinuxHotkeyListener(HotkeyListenerBase):
"""
Linux implementation using X11 or python-xlib.
"""
def start(self):
"""
Register global hotkey using X11.
"""
from Xlib import X, XK, display
from Xlib.ext import record
from Xlib.protocol import rq
self.local_display = display.Display()
self.record_display = display.Display()
# Map modifier strings to X11 masks
mod_map = {
'ctrl': X.ControlMask,
'shift': X.ShiftMask,
'alt': X.Mod1Mask,
'super': X.Mod4Mask
}
# Calculate modifier mask
self.mod_mask = 0
for mod in self.modifiers:
if mod in mod_map:
self.mod_mask |= mod_map[mod]
# Get keysym for our key
keysym_str = 'XK_' + self.key.lower()
self.target_keysym = getattr(XK, keysym_str, None)
if not self.target_keysym:
raise ValueError(f"Unknown key: {self.key}")
# Set up recording context
ctx = self.record_display.record_create_context(
0,
[record.AllClients],
[{
'core_requests': (0, 0),
'core_replies': (0, 0),
'ext_requests': (0, 0, 0, 0),
'ext_replies': (0, 0, 0, 0),
'delivered_events': (0, 0),
'device_events': (X.KeyPress, X.KeyPress),
'errors': (0, 0),
'client_started': False,
'client_died': False,
}]
)
# Start recording in separate thread
import threading
self.record_thread = threading.Thread(
target=self._record_loop,
args=(ctx,),
daemon=True
)
self.record_thread.start()
def _record_loop(self, ctx):
"""
Main loop that processes X11 events.
"""
self.record_display.record_enable_context(ctx, self._event_handler)
self.record_display.record_free_context(ctx)
def _event_handler(self, reply):
"""
Handle X11 keyboard events.
"""
from Xlib import X
if reply.category != record.FromServer:
return
if reply.client_swapped:
return
data = reply.data
while len(data):
event, data = rq.EventField(None).parse_binary_value(
data, self.record_display.display, None, None
)
if event.type == X.KeyPress:
# Get keysym from keycode
keysym = self.local_display.keycode_to_keysym(event.detail, 0)
# Check if modifiers and key match
if (event.state & self.mod_mask) == self.mod_mask:
if keysym == self.target_keysym:
if self.callback:
self.callback()
def stop(self):
"""
Stop the X11 event recording.
"""
if hasattr(self, 'record_display'):
self.record_display.close()
if hasattr(self, 'local_display'):
self.local_display.close()
class HotkeyListener:
"""
Factory class that creates the appropriate platform-specific listener.
"""
def __new__(cls, hotkey: str):
"""
Create and return the appropriate listener for the current platform.
"""
system = platform.system()
if system == 'Windows':
return WindowsHotkeyListener(hotkey)
elif system == 'Darwin':
return MacOSHotkeyListener(hotkey)
elif system == 'Linux':
return LinuxHotkeyListener(hotkey)
else:
raise NotImplementedError(f"Platform {system} not supported")
The hotkey listener implementation demonstrates the complexity of cross-platform development. Each operating system provides different APIs and requires different approaches. Windows uses a message-based system with RegisterHotKey. macOS requires event taps that intercept keyboard events at a low level. Linux uses the X11 protocol to monitor keyboard activity. The factory pattern ensures that the rest of the system can use a unified interface regardless of the underlying platform.
Context Capture and Awareness
Understanding what the developer is currently doing is essential for providing relevant assistance. The context capture system gathers information about the active window, running processes, and workspace state.
import os
import platform
from dataclasses import dataclass
from typing import Optional, List, Dict
from pathlib import Path
@dataclass
class DevelopmentContext:
"""
Represents the current development context including active
application, workspace, and environment information.
"""
# Active window information
window_title: str
app_name: str
app_path: Optional[str]
process_id: int
# Workspace information
current_directory: Optional[Path]
workspace_root: Optional[Path]
open_files: List[Path]
# Git repository information
git_repository: Optional[Path]
git_branch: Optional[str]
git_status: Optional[Dict[str, any]]
# IDE-specific information
ide_type: Optional[str] # 'vscode', 'intellij', 'pycharm', etc.
ide_version: Optional[str]
active_file: Optional[Path]
cursor_position: Optional[tuple] # (line, column)
# Environment information
python_version: Optional[str]
node_version: Optional[str]
java_version: Optional[str]
virtual_env: Optional[Path]
def __str__(self):
"""
Human-readable representation of the context.
"""
parts = [f"Application: {self.app_name}"]
if self.current_directory:
parts.append(f"Directory: {self.current_directory}")
if self.git_repository:
parts.append(f"Git: {self.git_branch} in {self.git_repository}")
if self.ide_type:
parts.append(f"IDE: {self.ide_type}")
if self.active_file:
parts.append(f"File: {self.active_file}")
return " | ".join(parts)
class ContextCapture:
"""
Captures the current development context by examining active
windows, processes, and workspace state.
"""
@staticmethod
def get_current_context() -> DevelopmentContext:
"""
Capture and return the current development context.
Returns:
DevelopmentContext object with all available information
"""
system = platform.system()
if system == 'Windows':
return ContextCapture._capture_windows_context()
elif system == 'Darwin':
return ContextCapture._capture_macos_context()
elif system == 'Linux':
return ContextCapture._capture_linux_context()
else:
raise NotImplementedError(f"Platform {system} not supported")
@staticmethod
def _capture_windows_context() -> DevelopmentContext:
"""
Capture context on Windows using win32 APIs.
"""
import win32gui
import win32process
import psutil
# Get active window
hwnd = win32gui.GetForegroundWindow()
window_title = win32gui.GetWindowText(hwnd)
# Get process information
_, process_id = win32process.GetWindowThreadProcessId(hwnd)
process = psutil.Process(process_id)
app_name = process.name()
app_path = process.exe()
# Get current directory from process
try:
current_directory = Path(process.cwd())
except (psutil.AccessDenied, psutil.NoSuchProcess):
current_directory = None
# Detect IDE type
ide_type = ContextCapture._detect_ide_type(app_name, app_path)
# Build context object
context = DevelopmentContext(
window_title=window_title,
app_name=app_name,
app_path=app_path,
process_id=process_id,
current_directory=current_directory,
workspace_root=None,
open_files=[],
git_repository=None,
git_branch=None,
git_status=None,
ide_type=ide_type,
ide_version=None,
active_file=None,
cursor_position=None,
python_version=None,
node_version=None,
java_version=None,
virtual_env=None
)
# Enhance context with additional information
ContextCapture._enhance_context(context)
return context
@staticmethod
def _capture_macos_context() -> DevelopmentContext:
"""
Capture context on macOS using Cocoa and Quartz APIs.
"""
from Cocoa import NSWorkspace
from Quartz import CGWindowListCopyWindowInfo, kCGWindowListOptionOnScreenOnly, kCGNullWindowID
import psutil
# Get active application
workspace = NSWorkspace.sharedWorkspace()
active_app = workspace.activeApplication()
app_name = active_app['NSApplicationName']
process_id = active_app['NSApplicationProcessIdentifier']
app_path = active_app.get('NSApplicationPath')
# Get active window title
window_list = CGWindowListCopyWindowInfo(
kCGWindowListOptionOnScreenOnly,
kCGNullWindowID
)
window_title = ""
for window in window_list:
if window.get('kCGWindowOwnerPID') == process_id:
if window.get('kCGWindowLayer') == 0:
window_title = window.get('kCGWindowName', '')
break
# Get process information
try:
process = psutil.Process(process_id)
current_directory = Path(process.cwd())
except (psutil.AccessDenied, psutil.NoSuchProcess):
current_directory = None
# Detect IDE type
ide_type = ContextCapture._detect_ide_type(app_name, app_path)
# Build context object
context = DevelopmentContext(
window_title=window_title,
app_name=app_name,
app_path=app_path,
process_id=process_id,
current_directory=current_directory,
workspace_root=None,
open_files=[],
git_repository=None,
git_branch=None,
git_status=None,
ide_type=ide_type,
ide_version=None,
active_file=None,
cursor_position=None,
python_version=None,
node_version=None,
java_version=None,
virtual_env=None
)
# Enhance context with additional information
ContextCapture._enhance_context(context)
return context
@staticmethod
def _capture_linux_context() -> DevelopmentContext:
"""
Capture context on Linux using X11 and proc filesystem.
"""
from Xlib import X, display
import psutil
# Get active window using X11
d = display.Display()
root = d.screen().root
# Get the active window
active_window = root.get_full_property(
d.intern_atom('_NET_ACTIVE_WINDOW'),
X.AnyPropertyType
).value[0]
window = d.create_resource_object('window', active_window)
# Get window title
window_title = window.get_full_property(
d.intern_atom('_NET_WM_NAME'),
0
)
if window_title:
window_title = window_title.value.decode('utf-8')
else:
window_title = ""
# Get process ID
pid_property = window.get_full_property(
d.intern_atom('_NET_WM_PID'),
X.AnyPropertyType
)
if pid_property:
process_id = pid_property.value[0]
else:
process_id = 0
# Get process information
try:
process = psutil.Process(process_id)
app_name = process.name()
app_path = process.exe()
current_directory = Path(process.cwd())
except (psutil.AccessDenied, psutil.NoSuchProcess):
app_name = "Unknown"
app_path = None
current_directory = None
# Detect IDE type
ide_type = ContextCapture._detect_ide_type(app_name, app_path)
# Build context object
context = DevelopmentContext(
window_title=window_title,
app_name=app_name,
app_path=app_path,
process_id=process_id,
current_directory=current_directory,
workspace_root=None,
open_files=[],
git_repository=None,
git_branch=None,
git_status=None,
ide_type=ide_type,
ide_version=None,
active_file=None,
cursor_position=None,
python_version=None,
node_version=None,
java_version=None,
virtual_env=None
)
# Enhance context with additional information
ContextCapture._enhance_context(context)
return context
@staticmethod
def _detect_ide_type(app_name: str, app_path: Optional[str]) -> Optional[str]:
"""
Detect the type of IDE based on application name and path.
Returns:
String identifier for the IDE type or None if not an IDE
"""
app_name_lower = app_name.lower()
if 'code' in app_name_lower or (app_path and 'vscode' in app_path.lower()):
return 'vscode'
elif 'idea' in app_name_lower:
if 'pycharm' in app_name_lower:
return 'pycharm'
elif 'webstorm' in app_name_lower:
return 'webstorm'
else:
return 'intellij'
elif 'eclipse' in app_name_lower:
return 'eclipse'
elif 'sublime' in app_name_lower:
return 'sublime'
elif 'atom' in app_name_lower:
return 'atom'
elif 'vim' in app_name_lower or 'nvim' in app_name_lower:
return 'vim'
elif 'emacs' in app_name_lower:
return 'emacs'
else:
return None
@staticmethod
def _enhance_context(context: DevelopmentContext):
"""
Enhance the context with additional information like Git status,
workspace root, environment versions, etc.
Args:
context: DevelopmentContext object to enhance (modified in place)
"""
# Find workspace root
if context.current_directory:
context.workspace_root = ContextCapture._find_workspace_root(
context.current_directory
)
# Find Git repository
if context.current_directory:
git_repo = ContextCapture._find_git_repository(
context.current_directory
)
if git_repo:
context.git_repository = git_repo
context.git_branch = ContextCapture._get_git_branch(git_repo)
context.git_status = ContextCapture._get_git_status(git_repo)
# Detect active file from window title or IDE-specific methods
if context.ide_type:
context.active_file = ContextCapture._detect_active_file(context)
# Detect environment versions
context.python_version = ContextCapture._get_python_version()
context.node_version = ContextCapture._get_node_version()
context.java_version = ContextCapture._get_java_version()
# Detect virtual environment
if context.current_directory:
context.virtual_env = ContextCapture._detect_virtual_env(
context.current_directory
)
@staticmethod
def _find_workspace_root(start_path: Path) -> Optional[Path]:
"""
Find the workspace root by looking for common markers like
.git, package.json, pom.xml, etc.
Args:
start_path: Directory to start searching from
Returns:
Path to workspace root or None if not found
"""
markers = [
'.git',
'package.json',
'pom.xml',
'build.gradle',
'Cargo.toml',
'go.mod',
'setup.py',
'pyproject.toml',
'Makefile',
'.project'
]
current = start_path
while current != current.parent:
for marker in markers:
if (current / marker).exists():
return current
current = current.parent
return None
@staticmethod
def _find_git_repository(start_path: Path) -> Optional[Path]:
"""
Find the Git repository root by looking for .git directory.
Args:
start_path: Directory to start searching from
Returns:
Path to Git repository root or None if not found
"""
current = start_path
while current != current.parent:
git_dir = current / '.git'
if git_dir.exists():
return current
current = current.parent
return None
@staticmethod
def _get_git_branch(repo_path: Path) -> Optional[str]:
"""
Get the current Git branch name.
Args:
repo_path: Path to Git repository root
Returns:
Branch name or None if cannot be determined
"""
head_file = repo_path / '.git' / 'HEAD'
if not head_file.exists():
return None
try:
with open(head_file, 'r') as f:
content = f.read().strip()
if content.startswith('ref: refs/heads/'):
return content[16:]
else:
# Detached HEAD state
return content[:7]
except Exception:
return None
@staticmethod
def _get_git_status(repo_path: Path) -> Optional[Dict[str, any]]:
"""
Get Git repository status including modified files, staged files, etc.
Args:
repo_path: Path to Git repository root
Returns:
Dictionary with status information or None if error
"""
import subprocess
try:
# Get status in porcelain format
result = subprocess.run(
['git', 'status', '--porcelain'],
cwd=repo_path,
capture_output=True,
text=True,
timeout=5
)
if result.returncode != 0:
return None
# Parse status output
modified = []
staged = []
untracked = []
for line in result.stdout.splitlines():
if len(line) < 3:
continue
status = line[:2]
filename = line[3:]
if status[0] in ['M', 'A', 'D', 'R', 'C']:
staged.append(filename)
if status[1] == 'M':
modified.append(filename)
if status == '??':
untracked.append(filename)
return {
'modified': modified,
'staged': staged,
'untracked': untracked,
'clean': len(modified) == 0 and len(staged) == 0 and len(untracked) == 0
}
except Exception:
return None
@staticmethod
def _detect_active_file(context: DevelopmentContext) -> Optional[Path]:
"""
Detect the currently active file in the IDE.
Args:
context: Development context with IDE information
Returns:
Path to active file or None if cannot be determined
"""
# Try to extract from window title first
if context.window_title and context.current_directory:
# Many IDEs show the filename in the window title
title_parts = context.window_title.split(' - ')
for part in title_parts:
# Check if it looks like a file path
if '.' in part and not part.startswith('http'):
potential_file = context.current_directory / part.strip()
if potential_file.exists():
return potential_file
# IDE-specific detection methods
if context.ide_type == 'vscode':
return ContextCapture._get_vscode_active_file(context)
elif context.ide_type in ['intellij', 'pycharm', 'webstorm']:
return ContextCapture._get_jetbrains_active_file(context)
return None
@staticmethod
def _get_vscode_active_file(context: DevelopmentContext) -> Optional[Path]:
"""
Get the active file in VS Code by querying its state.
Args:
context: Development context
Returns:
Path to active file or None
"""
# VS Code stores workspace state in various locations
# This is a simplified approach - full implementation would
# need to parse VS Code's workspace storage
if not context.current_directory:
return None
# Check for .vscode directory
vscode_dir = context.current_directory / '.vscode'
if not vscode_dir.exists():
return None
# In practice, we would need to use VS Code's extension API
# or command line interface to get this information reliably
return None
@staticmethod
def _get_jetbrains_active_file(context: DevelopmentContext) -> Optional[Path]:
"""
Get the active file in JetBrains IDEs.
Args:
context: Development context
Returns:
Path to active file or None
"""
# JetBrains IDEs store project information in .idea directory
# Similar to VS Code, full implementation would require
# parsing IDE-specific files or using their APIs
return None
@staticmethod
def _get_python_version() -> Optional[str]:
"""
Get the Python version available in the environment.
Returns:
Version string or None if Python not found
"""
import subprocess
try:
result = subprocess.run(
['python', '--version'],
capture_output=True,
text=True,
timeout=5
)
if result.returncode == 0:
# Output format: "Python 3.9.7"
return result.stdout.strip().split()[1]
except Exception:
pass
return None
@staticmethod
def _get_node_version() -> Optional[str]:
"""
Get the Node.js version available in the environment.
Returns:
Version string or None if Node not found
"""
import subprocess
try:
result = subprocess.run(
['node', '--version'],
capture_output=True,
text=True,
timeout=5
)
if result.returncode == 0:
# Output format: "v16.13.0"
return result.stdout.strip()[1:] # Remove 'v' prefix
except Exception:
pass
return None
@staticmethod
def _get_java_version() -> Optional[str]:
"""
Get the Java version available in the environment.
Returns:
Version string or None if Java not found
"""
import subprocess
try:
result = subprocess.run(
['java', '-version'],
capture_output=True,
text=True,
timeout=5
)
# Java outputs version to stderr
output = result.stderr
if result.returncode == 0 and output:
# Parse version from output
for line in output.splitlines():
if 'version' in line.lower():
# Extract version number
parts = line.split('"')
if len(parts) >= 2:
return parts[1]
except Exception:
pass
return None
@staticmethod
def _detect_virtual_env(start_path: Path) -> Optional[Path]:
"""
Detect if we are in a Python virtual environment.
Args:
start_path: Directory to start searching from
Returns:
Path to virtual environment or None if not found
"""
# Check for common virtual environment directories
venv_names = ['venv', '.venv', 'env', '.env', 'virtualenv']
current = start_path
while current != current.parent:
for venv_name in venv_names:
venv_path = current / venv_name
# Check if it looks like a virtual environment
if venv_path.exists() and (venv_path / 'bin' / 'python').exists():
return venv_path
if venv_path.exists() and (venv_path / 'Scripts' / 'python.exe').exists():
return venv_path
current = current.parent
# Check environment variable
virtual_env = os.environ.get('VIRTUAL_ENV')
if virtual_env:
return Path(virtual_env)
return None
The context capture system provides comprehensive information about the developer's current environment. It examines the active window to determine which application is in focus, extracts the process information to understand what the developer is working on, and scans the filesystem to gather workspace details. This context is crucial for the LLM to provide relevant assistance. For example, if the developer is working in a Git
repository on a feature branch with uncommitted changes, the assistant can take this into account when suggesting actions.
LLM Integration with Multi-Backend Support
Supporting multiple LLM backends requires a flexible architecture that abstracts the differences between local and remote models, different inference engines, and various GPU architectures. We create a unified interface that the rest of the system can use regardless of the backend.
from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional, AsyncIterator
from dataclasses import dataclass
import asyncio
import json
@dataclass
class Message:
"""
Represents a single message in the conversation.
"""
role: str # 'system', 'user', 'assistant', 'tool'
content: str
tool_calls: Optional[List[Dict[str, Any]]] = None
tool_call_id: Optional[str] = None
name: Optional[str] = None
@dataclass
class LLMConfig:
"""
Configuration for LLM backend.
"""
backend_type: str # 'openai', 'anthropic', 'local_llama', 'local_vllm', 'ollama'
model_name: str
api_key: Optional[str] = None
api_base: Optional[str] = None
temperature: float = 0.7
max_tokens: int = 2048
# Local model configuration
model_path: Optional[str] = None
gpu_layers: int = -1 # -1 means use all available
context_length: int = 4096
# GPU configuration
device: str = 'auto' # 'cuda', 'rocm', 'mps', 'cpu', 'auto'
gpu_memory_fraction: float = 0.9
class LLMBackend(ABC):
"""
Abstract base class for LLM backends.
"""
def __init__(self, config: LLMConfig):
"""
Initialize the backend with configuration.
Args:
config: LLMConfig object with backend settings
"""
self.config = config
@abstractmethod
async def generate(
self,
messages: List[Message],
tools: Optional[List[Dict[str, Any]]] = None,
stream: bool = False
) -> AsyncIterator[str]:
"""
Generate a response from the LLM.
Args:
messages: List of conversation messages
tools: Optional list of tool specifications
stream: Whether to stream the response
Yields:
Response chunks if streaming, otherwise yields complete response
"""
pass
@abstractmethod
async def generate_with_tools(
self,
messages: List[Message],
tools: List[Dict[str, Any]]
) -> Message:
"""
Generate a response that may include tool calls.
Args:
messages: List of conversation messages
tools: List of tool specifications
Returns:
Message object with potential tool calls
"""
pass
@abstractmethod
def supports_tool_calling(self) -> bool:
"""
Check if this backend supports native tool calling.
Returns:
True if tool calling is supported
"""
pass
class OpenAIBackend(LLMBackend):
"""
Backend for OpenAI API (GPT-4, GPT-3.5, etc.)
"""
def __init__(self, config: LLMConfig):
"""
Initialize OpenAI backend.
"""
super().__init__(config)
try:
import openai
self.client = openai.AsyncOpenAI(
api_key=config.api_key,
base_url=config.api_base
)
except ImportError:
raise ImportError("openai package not installed. Install with: pip install openai")
async def generate(
self,
messages: List[Message],
tools: Optional[List[Dict[str, Any]]] = None,
stream: bool = False
) -> AsyncIterator[str]:
"""
Generate response using OpenAI API.
"""
# Convert messages to OpenAI format
openai_messages = []
for msg in messages:
message_dict = {
'role': msg.role,
'content': msg.content
}
if msg.tool_calls:
message_dict['tool_calls'] = msg.tool_calls
if msg.tool_call_id:
message_dict['tool_call_id'] = msg.tool_call_id
if msg.name:
message_dict['name'] = msg.name
openai_messages.append(message_dict)
# Prepare API call parameters
params = {
'model': self.config.model_name,
'messages': openai_messages,
'temperature': self.config.temperature,
'max_tokens': self.config.max_tokens,
'stream': stream
}
if tools:
params['tools'] = tools
params['tool_choice'] = 'auto'
# Make API call
if stream:
response = await self.client.chat.completions.create(**params)
async for chunk in response:
if chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content
else:
response = await self.client.chat.completions.create(**params)
yield response.choices[0].message.content
async def generate_with_tools(
self,
messages: List[Message],
tools: List[Dict[str, Any]]
) -> Message:
"""
Generate response with tool calling support.
"""
# Convert messages to OpenAI format
openai_messages = []
for msg in messages:
message_dict = {
'role': msg.role,
'content': msg.content
}
if msg.tool_calls:
message_dict['tool_calls'] = msg.tool_calls
if msg.tool_call_id:
message_dict['tool_call_id'] = msg.tool_call_id
if msg.name:
message_dict['name'] = msg.name
openai_messages.append(message_dict)
# Make API call with tools
response = await self.client.chat.completions.create(
model=self.config.model_name,
messages=openai_messages,
tools=tools,
tool_choice='auto',
temperature=self.config.temperature,
max_tokens=self.config.max_tokens
)
# Convert response to Message object
choice = response.choices[0]
message = choice.message
result = Message(
role='assistant',
content=message.content or ''
)
if message.tool_calls:
result.tool_calls = [
{
'id': tc.id,
'type': tc.type,
'function': {
'name': tc.function.name,
'arguments': tc.function.arguments
}
}
for tc in message.tool_calls
]
return result
def supports_tool_calling(self) -> bool:
"""
OpenAI supports native tool calling.
"""
return True
class LocalLlamaBackend(LLMBackend):
"""
Backend for local LLaMA models using llama-cpp-python.
Supports CUDA, ROCm, Metal (MPS), and CPU.
"""
def __init__(self, config: LLMConfig):
"""
Initialize local LLaMA backend with appropriate GPU support.
"""
super().__init__(config)
try:
from llama_cpp import Llama
except ImportError:
raise ImportError(
"llama-cpp-python not installed. Install with appropriate backend:\n"
"CUDA: pip install llama-cpp-python --extra-index-url https://abetlen.github.io/llama-cpp-python/whl/cu121\n"
"ROCm: pip install llama-cpp-python --extra-index-url https://abetlen.github.io/llama-cpp-python/whl/rocm\n"
"Metal: pip install llama-cpp-python (on macOS)\n"
"CPU: pip install llama-cpp-python"
)
# Determine device and configure accordingly
device = self._detect_device() if config.device == 'auto' else config.device
# Configure llama.cpp parameters based on device
llama_params = {
'model_path': config.model_path,
'n_ctx': config.context_length,
'verbose': False
}
if device == 'cuda':
llama_params['n_gpu_layers'] = config.gpu_layers
elif device == 'rocm':
# ROCm uses same parameter as CUDA
llama_params['n_gpu_layers'] = config.gpu_layers
elif device == 'mps':
# Metal Performance Shaders for Apple Silicon
llama_params['n_gpu_layers'] = 1 # Enable Metal
else:
# CPU only
llama_params['n_gpu_layers'] = 0
self.model = Llama(**llama_params)
self.device = device
def _detect_device(self) -> str:
"""
Automatically detect the best available device.
Returns:
Device string: 'cuda', 'rocm', 'mps', or 'cpu'
"""
import platform
# Check for NVIDIA CUDA
try:
import torch
if torch.cuda.is_available():
return 'cuda'
except ImportError:
pass
# Check for AMD ROCm
try:
import torch
if hasattr(torch, 'hip') and torch.hip.is_available():
return 'rocm'
except (ImportError, AttributeError):
pass
# Check for Apple Metal
if platform.system() == 'Darwin':
try:
import torch
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
return 'mps'
except ImportError:
pass
# Default to CPU
return 'cpu'
async def generate(
self,
messages: List[Message],
tools: Optional[List[Dict[str, Any]]] = None,
stream: bool = False
) -> AsyncIterator[str]:
"""
Generate response using local LLaMA model.
"""
# Convert messages to prompt format
prompt = self._messages_to_prompt(messages, tools)
# Run generation in thread pool to avoid blocking
loop = asyncio.get_event_loop()
if stream:
# Stream tokens as they are generated
generator = await loop.run_in_executor(
None,
lambda: self.model(
prompt,
max_tokens=self.config.max_tokens,
temperature=self.config.temperature,
stream=True
)
)
for chunk in generator:
if 'choices' in chunk and len(chunk['choices']) > 0:
text = chunk['choices'][0].get('text', '')
if text:
yield text
else:
# Generate complete response
response = await loop.run_in_executor(
None,
lambda: self.model(
prompt,
max_tokens=self.config.max_tokens,
temperature=self.config.temperature,
stream=False
)
)
if 'choices' in response and len(response['choices']) > 0:
yield response['choices'][0]['text']
async def generate_with_tools(
self,
messages: List[Message],
tools: List[Dict[str, Any]]
) -> Message:
"""
Generate response with simulated tool calling.
Local models don't natively support tool calling, so we use
prompt engineering to achieve similar functionality.
"""
# Create a prompt that instructs the model about available tools
tool_prompt = self._create_tool_prompt(tools)
# Add tool instructions to system message
enhanced_messages = messages.copy()
if enhanced_messages and enhanced_messages[0].role == 'system':
enhanced_messages[0] = Message(
role='system',
content=enhanced_messages[0].content + '\n\n' + tool_prompt
)
else:
enhanced_messages.insert(0, Message(
role='system',
content=tool_prompt
))
# Generate response
prompt = self._messages_to_prompt(enhanced_messages, tools)
loop = asyncio.get_event_loop()
response = await loop.run_in_executor(
None,
lambda: self.model(
prompt,
max_tokens=self.config.max_tokens,
temperature=self.config.temperature,
stream=False
)
)
response_text = response['choices'][0]['text']
# Parse potential tool calls from response
tool_calls = self._parse_tool_calls(response_text)
result = Message(
role='assistant',
content=response_text
)
if tool_calls:
result.tool_calls = tool_calls
return result
def _messages_to_prompt(
self,
messages: List[Message],
tools: Optional[List[Dict[str, Any]]] = None
) -> str:
"""
Convert messages to a prompt string suitable for the model.
"""
prompt_parts = []
for msg in messages:
if msg.role == 'system':
prompt_parts.append(f"System: {msg.content}")
elif msg.role == 'user':
prompt_parts.append(f"User: {msg.content}")
elif msg.role == 'assistant':
prompt_parts.append(f"Assistant: {msg.content}")
elif msg.role == 'tool':
prompt_parts.append(f"Tool Result ({msg.name}): {msg.content}")
prompt_parts.append("Assistant:")
return '\n\n'.join(prompt_parts)
def _create_tool_prompt(self, tools: List[Dict[str, Any]]) -> str:
"""
Create a prompt that describes available tools to the model.
"""
tool_descriptions = []
for tool in tools:
func = tool.get('function', {})
name = func.get('name', '')
description = func.get('description', '')
parameters = func.get('parameters', {})
tool_desc = f"Tool: {name}\nDescription: {description}\nParameters: {json.dumps(parameters, indent=2)}"
tool_descriptions.append(tool_desc)
tools_text = '\n\n'.join(tool_descriptions)
return f"""You have access to the following tools:
{tools_text}
To use a tool, respond with a JSON object in this format:
{{
"tool_call": {{
"name": "tool_name",
"arguments": {{
"param1": "value1",
"param2": "value2"
}}
}}
}}
If you don't need to use a tool, respond normally with text."""
def _parse_tool_calls(self, response_text: str) -> Optional[List[Dict[str, Any]]]:
"""
Parse tool calls from model response.
Looks for JSON objects with tool_call structure.
"""
import re
# Look for JSON objects in the response
json_pattern = r'\{[^{}]*"tool_call"[^{}]*\{[^{}]*\}[^{}]*\}'
matches = re.finditer(json_pattern, response_text, re.DOTALL)
tool_calls = []
for match in matches:
try:
data = json.loads(match.group())
if 'tool_call' in data:
tool_call = data['tool_call']
tool_calls.append({
'id': f"call_{len(tool_calls)}",
'type': 'function',
'function': {
'name': tool_call.get('name', ''),
'arguments': json.dumps(tool_call.get('arguments', {}))
}
})
except json.JSONDecodeError:
continue
return tool_calls if tool_calls else None
def supports_tool_calling(self) -> bool:
"""
Local models use simulated tool calling via prompt engineering.
"""
return False
class VLLMBackend(LLMBackend):
"""
Backend for vLLM high-performance inference server.
Supports CUDA and ROCm.
"""
def __init__(self, config: LLMConfig):
"""
Initialize vLLM backend.
"""
super().__init__(config)
try:
from vllm import LLM, SamplingParams
self.LLM = LLM
self.SamplingParams = SamplingParams
except ImportError:
raise ImportError(
"vllm not installed. Install with:\n"
"CUDA: pip install vllm\n"
"ROCm: pip install vllm (with ROCm PyTorch)"
)
# Determine device
device = self._detect_device() if config.device == 'auto' else config.device
# Initialize vLLM engine
engine_args = {
'model': config.model_path or config.model_name,
'tensor_parallel_size': 1,
'gpu_memory_utilization': config.gpu_memory_fraction
}
if device == 'rocm':
# vLLM automatically detects ROCm
pass
elif device != 'cuda':
raise ValueError(f"vLLM only supports CUDA and ROCm, got {device}")
self.model = self.LLM(**engine_args)
self.device = device
def _detect_device(self) -> str:
"""
Detect CUDA or ROCm availability.
"""
try:
import torch
if torch.cuda.is_available():
# Check if it's ROCm or CUDA
if hasattr(torch.version, 'hip') and torch.version.hip is not None:
return 'rocm'
return 'cuda'
except ImportError:
pass
raise RuntimeError("vLLM requires CUDA or ROCm GPU")
async def generate(
self,
messages: List[Message],
tools: Optional[List[Dict[str, Any]]] = None,
stream: bool = False
) -> AsyncIterator[str]:
"""
Generate response using vLLM.
"""
# Convert messages to prompt
prompt = self._messages_to_prompt(messages, tools)
# Create sampling parameters
sampling_params = self.SamplingParams(
temperature=self.config.temperature,
max_tokens=self.config.max_tokens
)
# Run generation in thread pool
loop = asyncio.get_event_loop()
if stream:
# vLLM doesn't support streaming in the same way
# We generate and yield the complete response
outputs = await loop.run_in_executor(
None,
lambda: self.model.generate([prompt], sampling_params)
)
if outputs:
yield outputs[0].outputs[0].text
else:
outputs = await loop.run_in_executor(
None,
lambda: self.model.generate([prompt], sampling_params)
)
if outputs:
yield outputs[0].outputs[0].text
async def generate_with_tools(
self,
messages: List[Message],
tools: List[Dict[str, Any]]
) -> Message:
"""
Generate with simulated tool calling (similar to LocalLlamaBackend).
"""
# Similar implementation to LocalLlamaBackend
tool_prompt = self._create_tool_prompt(tools)
enhanced_messages = messages.copy()
if enhanced_messages and enhanced_messages[0].role == 'system':
enhanced_messages[0] = Message(
role='system',
content=enhanced_messages[0].content + '\n\n' + tool_prompt
)
else:
enhanced_messages.insert(0, Message(
role='system',
content=tool_prompt
))
prompt = self._messages_to_prompt(enhanced_messages, tools)
sampling_params = self.SamplingParams(
temperature=self.config.temperature,
max_tokens=self.config.max_tokens
)
loop = asyncio.get_event_loop()
outputs = await loop.run_in_executor(
None,
lambda: self.model.generate([prompt], sampling_params)
)
response_text = outputs[0].outputs[0].text if outputs else ""
tool_calls = self._parse_tool_calls(response_text)
result = Message(
role='assistant',
content=response_text
)
if tool_calls:
result.tool_calls = tool_calls
return result
def _messages_to_prompt(
self,
messages: List[Message],
tools: Optional[List[Dict[str, Any]]] = None
) -> str:
"""
Convert messages to prompt string.
"""
prompt_parts = []
for msg in messages:
if msg.role == 'system':
prompt_parts.append(f"System: {msg.content}")
elif msg.role == 'user':
prompt_parts.append(f"User: {msg.content}")
elif msg.role == 'assistant':
prompt_parts.append(f"Assistant: {msg.content}")
elif msg.role == 'tool':
prompt_parts.append(f"Tool Result ({msg.name}): {msg.content}")
prompt_parts.append("Assistant:")
return '\n\n'.join(prompt_parts)
def _create_tool_prompt(self, tools: List[Dict[str, Any]]) -> str:
"""
Create tool description prompt.
"""
tool_descriptions = []
for tool in tools:
func = tool.get('function', {})
name = func.get('name', '')
description = func.get('description', '')
parameters = func.get('parameters', {})
tool_desc = f"Tool: {name}\nDescription: {description}\nParameters: {json.dumps(parameters, indent=2)}"
tool_descriptions.append(tool_desc)
tools_text = '\n\n'.join(tool_descriptions)
return f"""You have access to the following tools:
{tools_text}
To use a tool, respond with a JSON object in this format:
{{
"tool_call": {{
"name": "tool_name",
"arguments": {{
"param1": "value1"
}}
}}
}}"""
def _parse_tool_calls(self, response_text: str) -> Optional[List[Dict[str, Any]]]:
"""
Parse tool calls from response.
"""
import re
json_pattern = r'\{[^{}]*"tool_call"[^{}]*\{[^{}]*\}[^{}]*\}'
matches = re.finditer(json_pattern, response_text, re.DOTALL)
tool_calls = []
for match in matches:
try:
data = json.loads(match.group())
if 'tool_call' in data:
tool_call = data['tool_call']
tool_calls.append({
'id': f"call_{len(tool_calls)}",
'type': 'function',
'function': {
'name': tool_call.get('name', ''),
'arguments': json.dumps(tool_call.get('arguments', {}))
}
})
except json.JSONDecodeError:
continue
return tool_calls if tool_calls else None
def supports_tool_calling(self) -> bool:
"""
vLLM uses simulated tool calling.
"""
return False
class OllamaBackend(LLMBackend):
"""
Backend for Ollama local model serving.
Supports all GPU types through Ollama's abstraction.
"""
def __init__(self, config: LLMConfig):
"""
Initialize Ollama backend.
"""
super().__init__(config)
try:
import httpx
self.httpx = httpx
except ImportError:
raise ImportError("httpx not installed. Install with: pip install httpx")
self.api_base = config.api_base or 'http://localhost:11434'
self.client = httpx.AsyncClient(base_url=self.api_base, timeout=300.0)
async def generate(
self,
messages: List[Message],
tools: Optional[List[Dict[str, Any]]] = None,
stream: bool = False
) -> AsyncIterator[str]:
"""
Generate response using Ollama API.
"""
# Convert messages to Ollama format
ollama_messages = []
for msg in messages:
ollama_messages.append({
'role': msg.role,
'content': msg.content
})
# Prepare request
request_data = {
'model': self.config.model_name,
'messages': ollama_messages,
'stream': stream,
'options': {
'temperature': self.config.temperature,
'num_predict': self.config.max_tokens
}
}
if stream:
async with self.client.stream(
'POST',
'/api/chat',
json=request_data
) as response:
async for line in response.aiter_lines():
if line:
try:
data = json.loads(line)
if 'message' in data and 'content' in data['message']:
yield data['message']['content']
except json.JSONDecodeError:
continue
else:
response = await self.client.post('/api/chat', json=request_data)
data = response.json()
if 'message' in data and 'content' in data['message']:
yield data['message']['content']
async def generate_with_tools(
self,
messages: List[Message],
tools: List[Dict[str, Any]]
) -> Message:
"""
Generate with simulated tool calling.
"""
tool_prompt = self._create_tool_prompt(tools)
enhanced_messages = messages.copy()
if enhanced_messages and enhanced_messages[0].role == 'system':
enhanced_messages[0] = Message(
role='system',
content=enhanced_messages[0].content + '\n\n' + tool_prompt
)
else:
enhanced_messages.insert(0, Message(
role='system',
content=tool_prompt
))
# Convert to Ollama format
ollama_messages = []
for msg in enhanced_messages:
ollama_messages.append({
'role': msg.role,
'content': msg.content
})
# Make request
request_data = {
'model': self.config.model_name,
'messages': ollama_messages,
'stream': False,
'options': {
'temperature': self.config.temperature,
'num_predict': self.config.max_tokens
}
}
response = await self.client.post('/api/chat', json=request_data)
data = response.json()
response_text = data.get('message', {}).get('content', '')
tool_calls = self._parse_tool_calls(response_text)
result = Message(
role='assistant',
content=response_text
)
if tool_calls:
result.tool_calls = tool_calls
return result
def _create_tool_prompt(self, tools: List[Dict[str, Any]]) -> str:
"""
Create tool description prompt.
"""
tool_descriptions = []
for tool in tools:
func = tool.get('function', {})
name = func.get('name', '')
description = func.get('description', '')
parameters = func.get('parameters', {})
tool_desc = f"Tool: {name}\nDescription: {description}\nParameters: {json.dumps(parameters, indent=2)}"
tool_descriptions.append(tool_desc)
tools_text = '\n\n'.join(tool_descriptions)
return f"""You have access to the following tools:
{tools_text}
To use a tool, respond with a JSON object in this format:
{{
"tool_call": {{
"name": "tool_name",
"arguments": {{
"param1": "value1"
}}
}}
}}"""
def _parse_tool_calls(self, response_text: str) -> Optional[List[Dict[str, Any]]]:
"""
Parse tool calls from response.
"""
import re
json_pattern = r'\{[^{}]*"tool_call"[^{}]*\{[^{}]*\}[^{}]*\}'
matches = re.finditer(json_pattern, response_text, re.DOTALL)
tool_calls = []
for match in matches:
try:
data = json.loads(match.group())
if 'tool_call' in data:
tool_call = data['tool_call']
tool_calls.append({
'id': f"call_{len(tool_calls)}",
'type': 'function',
'function': {
'name': tool_call.get('name', ''),
'arguments': json.dumps(tool_call.get('arguments', {}))
}
})
except json.JSONDecodeError:
continue
return tool_calls if tool_calls else None
def supports_tool_calling(self) -> bool:
"""
Ollama uses simulated tool calling.
"""
return False
async def __aenter__(self):
"""
Async context manager entry.
"""
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""
Async context manager exit.
"""
await self.client.aclose()
class LLMFactory:
"""
Factory for creating appropriate LLM backend based on configuration.
"""
@staticmethod
def create_backend(config: LLMConfig) -> LLMBackend:
"""
Create and return the appropriate LLM backend.
Args:
config: LLMConfig object specifying the backend type and settings
Returns:
Initialized LLMBackend instance
"""
backend_type = config.backend_type.lower()
if backend_type == 'openai':
return OpenAIBackend(config)
elif backend_type == 'local_llama':
return LocalLlamaBackend(config)
elif backend_type == 'local_vllm':
return VLLMBackend(config)
elif backend_type == 'ollama':
return OllamaBackend(config)
else:
raise ValueError(f"Unknown backend type: {backend_type}")
The LLM integration layer provides a unified interface for working with different model backends. The OpenAI backend uses the official API and supports native tool calling. The local backends including llama.cpp, vLLM, and Ollama require prompt engineering to simulate tool calling since most open-source models don't natively support this feature. Each backend handles GPU detection and configuration appropriately for the target hardware whether NVIDIA CUDA, AMD ROCm, Intel oneAPI, or Apple Metal.
Tool Framework and Execution Engine
The tool framework defines how the LLM can interact with the development environment. Each tool has a clear specification, validation logic, and execution handler.
from abc import ABC, abstractmethod
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass
import json
import asyncio
import subprocess
from pathlib import Path
@dataclass
class ToolParameter:
"""
Specification for a tool parameter.
"""
name: str
type: str # 'string', 'number', 'boolean', 'array', 'object'
description: str
required: bool = True
enum: Optional[List[Any]] = None
default: Optional[Any] = None
@dataclass
class ToolSpec:
"""
Complete specification for a tool.
"""
name: str
description: str
parameters: List[ToolParameter]
category: str # 'git', 'ide', 'system', 'package'
def to_openai_format(self) -> Dict[str, Any]:
"""
Convert tool specification to OpenAI function calling format.
Returns:
Dictionary in OpenAI tool format
"""
properties = {}
required = []
for param in self.parameters:
param_spec = {
'type': param.type,
'description': param.description
}
if param.enum:
param_spec['enum'] = param.enum
properties[param.name] = param_spec
if param.required:
required.append(param.name)
return {
'type': 'function',
'function': {
'name': self.name,
'description': self.description,
'parameters': {
'type': 'object',
'properties': properties,
'required': required
}
}
}
class Tool(ABC):
"""
Abstract base class for all tools.
"""
def __init__(self):
"""
Initialize the tool.
"""
self.spec = self._create_spec()
@abstractmethod
def _create_spec(self) -> ToolSpec:
"""
Create and return the tool specification.
Returns:
ToolSpec object describing this tool
"""
pass
@abstractmethod
async def execute(self, **kwargs) -> Dict[str, Any]:
"""
Execute the tool with given parameters.
Args:
**kwargs: Tool parameters as keyword arguments
Returns:
Dictionary with execution results
"""
pass
def validate_parameters(self, params: Dict[str, Any]) -> tuple[bool, Optional[str]]:
"""
Validate that provided parameters match the specification.
Args:
params: Dictionary of parameter values
Returns:
Tuple of (is_valid, error_message)
"""
# Check required parameters
for param in self.spec.parameters:
if param.required and param.name not in params:
return False, f"Missing required parameter: {param.name}"
# Check parameter types
for param_name, param_value in params.items():
# Find parameter spec
param_spec = next(
(p for p in self.spec.parameters if p.name == param_name),
None
)
if not param_spec:
return False, f"Unknown parameter: {param_name}"
# Validate type
if not self._check_type(param_value, param_spec.type):
return False, f"Parameter {param_name} has wrong type. Expected {param_spec.type}"
# Validate enum
if param_spec.enum and param_value not in param_spec.enum:
return False, f"Parameter {param_name} must be one of {param_spec.enum}"
return True, None
def _check_type(self, value: Any, expected_type: str) -> bool:
"""
Check if value matches expected type.
Args:
value: Value to check
expected_type: Expected type string
Returns:
True if type matches
"""
if expected_type == 'string':
return isinstance(value, str)
elif expected_type == 'number':
return isinstance(value, (int, float))
elif expected_type == 'boolean':
return isinstance(value, bool)
elif expected_type == 'array':
return isinstance(value, list)
elif expected_type == 'object':
return isinstance(value, dict)
else:
return True
class GitInitTool(Tool):
"""
Tool for initializing a Git repository.
"""
def _create_spec(self) -> ToolSpec:
"""
Create specification for git init tool.
"""
return ToolSpec(
name='git_init',
description='Initialize a new Git repository in the specified directory',
parameters=[
ToolParameter(
name='path',
type='string',
description='Path where the repository should be initialized',
required=True
),
ToolParameter(
name='initial_branch',
type='string',
description='Name of the initial branch (default: main)',
required=False,
default='main'
)
],
category='git'
)
async def execute(self, **kwargs) -> Dict[str, Any]:
"""
Execute git init command.
"""
# Validate parameters
is_valid, error = self.validate_parameters(kwargs)
if not is_valid:
return {
'success': False,
'error': error
}
path = Path(kwargs['path'])
initial_branch = kwargs.get('initial_branch', 'main')
# Check if path exists
if not path.exists():
try:
path.mkdir(parents=True, exist_ok=True)
except Exception as e:
return {
'success': False,
'error': f"Failed to create directory: {str(e)}"
}
# Check if already a git repository
if (path / '.git').exists():
return {
'success': False,
'error': 'Directory is already a Git repository'
}
# Run git init
try:
result = await asyncio.create_subprocess_exec(
'git', 'init', '-b', initial_branch,
cwd=str(path),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await result.communicate()
if result.returncode == 0:
return {
'success': True,
'message': f'Initialized Git repository in {path}',
'branch': initial_branch
}
else:
return {
'success': False,
'error': stderr.decode('utf-8')
}
except Exception as e:
return {
'success': False,
'error': f'Failed to execute git init: {str(e)}'
}
class GitBranchTool(Tool):
"""
Tool for creating and switching Git branches.
"""
def _create_spec(self) -> ToolSpec:
"""
Create specification for git branch tool.
"""
return ToolSpec(
name='git_branch',
description='Create a new branch or switch to an existing branch',
parameters=[
ToolParameter(
name='repository_path',
type='string',
description='Path to the Git repository',
required=True
),
ToolParameter(
name='branch_name',
type='string',
description='Name of the branch',
required=True
),
ToolParameter(
name='create',
type='boolean',
description='Create the branch if it does not exist',
required=False,
default=True
),
ToolParameter(
name='checkout',
type='boolean',
description='Switch to the branch after creating/verifying it',
required=False,
default=True
)
],
category='git'
)
async def execute(self, **kwargs) -> Dict[str, Any]:
"""
Execute git branch operations.
"""
is_valid, error = self.validate_parameters(kwargs)
if not is_valid:
return {'success': False, 'error': error}
repo_path = Path(kwargs['repository_path'])
branch_name = kwargs['branch_name']
create = kwargs.get('create', True)
checkout = kwargs.get('checkout', True)
# Verify it's a git repository
if not (repo_path / '.git').exists():
return {
'success': False,
'error': f'{repo_path} is not a Git repository'
}
try:
# Check if branch exists
result = await asyncio.create_subprocess_exec(
'git', 'branch', '--list', branch_name,
cwd=str(repo_path),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await result.communicate()
branch_exists = bool(stdout.decode('utf-8').strip())
# Create branch if needed
if not branch_exists and create:
result = await asyncio.create_subprocess_exec(
'git', 'branch', branch_name,
cwd=str(repo_path),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await result.communicate()
if result.returncode != 0:
return {
'success': False,
'error': f'Failed to create branch: {stderr.decode("utf-8")}'
}
elif not branch_exists and not create:
return {
'success': False,
'error': f'Branch {branch_name} does not exist'
}
# Checkout branch if requested
if checkout:
result = await asyncio.create_subprocess_exec(
'git', 'checkout', branch_name,
cwd=str(repo_path),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await result.communicate()
if result.returncode != 0:
return {
'success': False,
'error': f'Failed to checkout branch: {stderr.decode("utf-8")}'
}
return {
'success': True,
'message': f'Branch {branch_name} is ready',
'created': not branch_exists and create,
'checked_out': checkout
}
except Exception as e:
return {
'success': False,
'error': f'Git operation failed: {str(e)}'
}
class GitCommitTool(Tool):
"""
Tool for committing changes to Git.
"""
def _create_spec(self) -> ToolSpec:
"""
Create specification for git commit tool.
"""
return ToolSpec(
name='git_commit',
description='Stage and commit changes to the Git repository',
parameters=[
ToolParameter(
name='repository_path',
type='string',
description='Path to the Git repository',
required=True
),
ToolParameter(
name='message',
type='string',
description='Commit message',
required=True
),
ToolParameter(
name='files',
type='array',
description='List of files to stage (empty means all changes)',
required=False,
default=[]
),
ToolParameter(
name='add_all',
type='boolean',
description='Stage all changes including untracked files',
required=False,
default=False
)
],
category='git'
)
async def execute(self, **kwargs) -> Dict[str, Any]:
"""
Execute git commit operation.
"""
is_valid, error = self.validate_parameters(kwargs)
if not is_valid:
return {'success': False, 'error': error}
repo_path = Path(kwargs['repository_path'])
message = kwargs['message']
files = kwargs.get('files', [])
add_all = kwargs.get('add_all', False)
if not (repo_path / '.git').exists():
return {
'success': False,
'error': f'{repo_path} is not a Git repository'
}
try:
# Stage files
if add_all:
add_cmd = ['git', 'add', '-A']
elif files:
add_cmd = ['git', 'add'] + files
else:
add_cmd = ['git', 'add', '-u']
result = await asyncio.create_subprocess_exec(
*add_cmd,
cwd=str(repo_path),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await result.communicate()
if result.returncode != 0:
return {
'success': False,
'error': f'Failed to stage files: {stderr.decode("utf-8")}'
}
# Commit
result = await asyncio.create_subprocess_exec(
'git', 'commit', '-m', message,
cwd=str(repo_path),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await result.communicate()
if result.returncode != 0:
# Check if there's nothing to commit
if 'nothing to commit' in stderr.decode('utf-8').lower():
return {
'success': True,
'message': 'Nothing to commit, working tree clean',
'committed': False
}
else:
return {
'success': False,
'error': f'Commit failed: {stderr.decode("utf-8")}'
}
return {
'success': True,
'message': f'Changes committed successfully',
'committed': True
}
except Exception as e:
return {
'success': False,
'error': f'Git operation failed: {str(e)}'
}
class VSCodeExtensionTool(Tool):
"""
Tool for managing VS Code extensions.
"""
def _create_spec(self) -> ToolSpec:
"""
Create specification for VS Code extension tool.
"""
return ToolSpec(
name='vscode_extension',
description='Install, uninstall, or list VS Code extensions',
parameters=[
ToolParameter(
name='action',
type='string',
description='Action to perform',
required=True,
enum=['install', 'uninstall', 'list']
),
ToolParameter(
name='extension_id',
type='string',
description='Extension identifier (required for install/uninstall)',
required=False
)
],
category='ide'
)
async def execute(self, **kwargs) -> Dict[str, Any]:
"""
Execute VS Code extension management.
"""
is_valid, error = self.validate_parameters(kwargs)
if not is_valid:
return {'success': False, 'error': error}
action = kwargs['action']
extension_id = kwargs.get('extension_id')
# Find code command
code_cmd = self._find_code_command()
if not code_cmd:
return {
'success': False,
'error': 'VS Code command line tool not found'
}
try:
if action == 'list':
result = await asyncio.create_subprocess_exec(
code_cmd, '--list-extensions',
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await result.communicate()
if result.returncode == 0:
extensions = stdout.decode('utf-8').strip().split('\n')
return {
'success': True,
'extensions': extensions
}
else:
return {
'success': False,
'error': stderr.decode('utf-8')
}
elif action == 'install':
if not extension_id:
return {
'success': False,
'error': 'extension_id required for install action'
}
result = await asyncio.create_subprocess_exec(
code_cmd, '--install-extension', extension_id,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await result.communicate()
if result.returncode == 0:
return {
'success': True,
'message': f'Extension {extension_id} installed successfully'
}
else:
return {
'success': False,
'error': stderr.decode('utf-8')
}
elif action == 'uninstall':
if not extension_id:
return {
'success': False,
'error': 'extension_id required for uninstall action'
}
result = await asyncio.create_subprocess_exec(
code_cmd, '--uninstall-extension', extension_id,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await result.communicate()
if result.returncode == 0:
return {
'success': True,
'message': f'Extension {extension_id} uninstalled successfully'
}
else:
return {
'success': False,
'error': stderr.decode('utf-8')
}
except Exception as e:
return {
'success': False,
'error': f'Operation failed: {str(e)}'
}
def _find_code_command(self) -> Optional[str]:
"""
Find the VS Code command line tool.
Returns:
Path to code command or None if not found
"""
import shutil
# Try common command names
for cmd in ['code', 'code-insiders']:
if shutil.which(cmd):
return cmd
return None
class ToolRegistry:
"""
Registry for managing available tools.
"""
def __init__(self):
"""
Initialize the tool registry.
"""
self.tools: Dict[str, Tool] = {}
self._register_default_tools()
def _register_default_tools(self):
"""
Register all default tools.
"""
# Git tools
self.register(GitInitTool())
self.register(GitBranchTool())
self.register(GitCommitTool())
# IDE tools
self.register(VSCodeExtensionTool())
def register(self, tool: Tool):
"""
Register a new tool.
Args:
tool: Tool instance to register
"""
self.tools[tool.spec.name] = tool
def get_tool(self, name: str) -> Optional[Tool]:
"""
Get a tool by name.
Args:
name: Tool name
Returns:
Tool instance or None if not found
"""
return self.tools.get(name)
def get_all_specs(self) -> List[Dict[str, Any]]:
"""
Get specifications for all registered tools in OpenAI format.
Returns:
List of tool specifications
"""
return [tool.spec.to_openai_format() for tool in self.tools.values()]
def get_specs_by_category(self, category: str) -> List[Dict[str, Any]]:
"""
Get tool specifications filtered by category.
Args:
category: Tool category
Returns:
List of tool specifications
"""
return [
tool.spec.to_openai_format()
for tool in self.tools.values()
if tool.spec.category == category
]
class ToolExecutor:
"""
Executes tools based on LLM tool calls.
"""
def __init__(self, registry: ToolRegistry):
"""
Initialize the tool executor.
Args:
registry: ToolRegistry with available tools
"""
self.registry = registry
async def execute_tool_call(self, tool_call: Dict[str, Any]) -> Dict[str, Any]:
"""
Execute a single tool call.
Args:
tool_call: Tool call dictionary from LLM response
Returns:
Execution result dictionary
"""
function_data = tool_call.get('function', {})
tool_name = function_data.get('name')
if not tool_name:
return {
'success': False,
'error': 'Tool name not specified'
}
tool = self.registry.get_tool(tool_name)
if not tool:
return {
'success': False,
'error': f'Unknown tool: {tool_name}'
}
# Parse arguments
try:
arguments_str = function_data.get('arguments', '{}')
if isinstance(arguments_str, str):
arguments = json.loads(arguments_str)
else:
arguments = arguments_str
except json.JSONDecodeError as e:
return {
'success': False,
'error': f'Invalid arguments JSON: {str(e)}'
}
# Execute tool
try:
result = await tool.execute(**arguments)
return result
except Exception as e:
return {
'success': False,
'error': f'Tool execution failed: {str(e)}'
}
async def execute_tool_calls(self, tool_calls: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Execute multiple tool calls in sequence.
Args:
tool_calls: List of tool call dictionaries
Returns:
List of execution results
"""
results = []
for tool_call in tool_calls:
result = await self.execute_tool_call(tool_call)
results.append({
'tool_call_id': tool_call.get('id'),
'result': result
})
return results
The tool framework provides a structured way to define and execute actions in the development environment. Each tool has a clear specification that describes its purpose, parameters, and validation rules. The ToolRegistry manages all available tools and provides them to the LLM in the appropriate format. The ToolExecutor handles the actual execution of tool calls, including parameter validation, error handling, and result formatting.
Now continuing with the method parse_tool and the remaining implementation:
class AgenticAssistant:
"""
Main agentic assistant that coordinates LLM, tools, and multi-step workflows.
"""
def __init__(
self,
llm_backend: LLMBackend,
tool_registry: ToolRegistry,
max_iterations: int = 10
):
"""
Initialize the agentic assistant.
Args:
llm_backend: LLM backend for generating responses
tool_registry: Registry of available tools
max_iterations: Maximum number of agent iterations to prevent infinite loops
"""
self.llm = llm_backend
self.registry = tool_registry
self.executor = ToolExecutor(tool_registry)
self.max_iterations = max_iterations
self.conversation_history: List[Message] = []
async def process_request(
self,
user_request: str,
context: DevelopmentContext
) -> str:
"""
Process a user request with full agentic capabilities.
Args:
user_request: Natural language request from the user
context: Current development context
Returns:
Final response to the user
"""
# Initialize conversation with system message
system_message = self._create_system_message(context)
self.conversation_history = [system_message]
# Add user request
self.conversation_history.append(Message(
role='user',
content=user_request
))
# Agent loop
iteration = 0
while iteration < self.max_iterations:
iteration += 1
# Get available tools based on context
tools = self._get_relevant_tools(context)
# Generate response from LLM
if self.llm.supports_tool_calling():
response = await self.llm.generate_with_tools(
self.conversation_history,
tools
)
else:
# For models without native tool calling, use prompt engineering
response = await self._generate_with_simulated_tools(tools)
# Add assistant response to history
self.conversation_history.append(response)
# Check if there are tool calls
if response.tool_calls:
# Execute all tool calls
tool_results = await self.executor.execute_tool_calls(response.tool_calls)
# Add tool results to conversation
for tool_result in tool_results:
self.conversation_history.append(Message(
role='tool',
content=json.dumps(tool_result['result']),
tool_call_id=tool_result['tool_call_id'],
name=self._get_tool_name_from_call_id(
response.tool_calls,
tool_result['tool_call_id']
)
))
# Continue loop to let LLM process results
continue
else:
# No tool calls, we have the final response
return response.content
# Max iterations reached
return "I apologize, but I was unable to complete the task within the allowed number of steps. Please try breaking down your request into smaller parts."
def _create_system_message(self, context: DevelopmentContext) -> Message:
"""
Create a system message that includes context information.
Args:
context: Current development context
Returns:
System message
"""
context_info = f"""You are an AI development assistant helping a developer with their workflow.
Current Context:
- Application: {context.app_name}
- Current Directory: {context.current_directory or 'Unknown'}
"""
if context.git_repository:
context_info += f"- Git Repository: {context.git_repository}\n"
context_info += f"- Current Branch: {context.git_branch or 'Unknown'}\n"
if context.ide_type:
context_info += f"- IDE: {context.ide_type}\n"
if context.active_file:
context_info += f"- Active File: {context.active_file}\n"
context_info += """
You have access to various tools to help the developer. Use these tools to accomplish the requested tasks.
Always explain what you're doing and why. If a task cannot be completed, explain the reason clearly.
"""
return Message(
role='system',
content=context_info
)
def _get_relevant_tools(self, context: DevelopmentContext) -> List[Dict[str, Any]]:
"""
Get tools relevant to the current context.
Args:
context: Current development context
Returns:
List of tool specifications
"""
# Start with all tools
all_tools = self.registry.get_all_specs()
# Could implement filtering logic based on context
# For example, only show Git tools if in a Git repository
# For now, return all tools
return all_tools
async def _generate_with_simulated_tools(
self,
tools: List[Dict[str, Any]]
) -> Message:
"""
Generate response for models without native tool calling.
Args:
tools: List of available tools
Returns:
Message with potential tool calls
"""
# Use the backend's generate_with_tools method which handles
# prompt engineering for tool calling
return await self.llm.generate_with_tools(
self.conversation_history,
tools
)
def _get_tool_name_from_call_id(
self,
tool_calls: List[Dict[str, Any]],
call_id: str
) -> str:
"""
Get the tool name from a tool call ID.
Args:
tool_calls: List of tool calls
call_id: Tool call ID to look up
Returns:
Tool name
"""
for call in tool_calls:
if call.get('id') == call_id:
return call.get('function', {}).get('name', 'unknown')
return 'unknown'
class MultiAgentSystem:
"""
Multi-agent system for complex workflows requiring specialized agents.
"""
def __init__(
self,
llm_backend: LLMBackend,
tool_registry: ToolRegistry
):
"""
Initialize the multi-agent system.
Args:
llm_backend: LLM backend for all agents
tool_registry: Registry of available tools
"""
self.llm = llm_backend
self.registry = tool_registry
self.executor = ToolExecutor(tool_registry)
# Create specialized agents
self.planner_agent = PlannerAgent(llm_backend)
self.git_agent = GitAgent(llm_backend, tool_registry)
self.ide_agent = IDEAgent(llm_backend, tool_registry)
self.validator_agent = ValidatorAgent(llm_backend)
async def process_complex_request(
self,
user_request: str,
context: DevelopmentContext
) -> str:
"""
Process a complex request using multiple specialized agents.
Args:
user_request: User's natural language request
context: Current development context
Returns:
Final response
"""
# Step 1: Planning
plan = await self.planner_agent.create_plan(user_request, context)
if not plan['success']:
return f"Unable to create a plan: {plan.get('error', 'Unknown error')}"
# Step 2: Execute plan steps
execution_log = []
for step in plan['steps']:
step_type = step['type']
step_description = step['description']
execution_log.append(f"Executing: {step_description}")
# Route to appropriate agent
if step_type == 'git':
result = await self.git_agent.execute_step(step, context)
elif step_type == 'ide':
result = await self.ide_agent.execute_step(step, context)
else:
result = {'success': False, 'error': f'Unknown step type: {step_type}'}
if not result['success']:
execution_log.append(f"Failed: {result.get('error', 'Unknown error')}")
break
else:
execution_log.append(f"Success: {result.get('message', 'Completed')}")
# Step 3: Validation
validation = await self.validator_agent.validate_results(
plan,
execution_log,
context
)
# Format final response
response_parts = [
"Task completed. Here's what I did:",
"",
*execution_log,
"",
validation['summary']
]
return '\n'.join(response_parts)
class PlannerAgent:
"""
Agent responsible for breaking down complex requests into executable steps.
"""
def __init__(self, llm_backend: LLMBackend):
"""
Initialize the planner agent.
Args:
llm_backend: LLM backend
"""
self.llm = llm_backend
async def create_plan(
self,
request: str,
context: DevelopmentContext
) -> Dict[str, Any]:
"""
Create an execution plan for the request.
Args:
request: User request
context: Development context
Returns:
Dictionary with plan steps
"""
system_prompt = """You are a planning agent. Your job is to break down development tasks into concrete steps.
Each step should specify:
- type: The category of the step (git, ide, system, package)
- description: What the step does
- tool: The specific tool to use
- parameters: Parameters for the tool
Output your plan as a JSON object with a 'steps' array."""
user_prompt = f"""Create a plan for this request: {request}
Context:
- Directory: {context.current_directory}
- Git Repository: {context.git_repository or 'None'}
- IDE: {context.ide_type or 'None'}
Provide a detailed step-by-step plan."""
messages = [
Message(role='system', content=system_prompt),
Message(role='user', content=user_prompt)
]
# Generate plan
response_text = ""
async for chunk in self.llm.generate(messages):
response_text += chunk
# Parse plan from response
try:
# Extract JSON from response
import re
json_match = re.search(r'\{.*\}', response_text, re.DOTALL)
if json_match:
plan_data = json.loads(json_match.group())
return {
'success': True,
'steps': plan_data.get('steps', [])
}
else:
return {
'success': False,
'error': 'Could not parse plan from response'
}
except json.JSONDecodeError as e:
return {
'success': False,
'error': f'Invalid plan JSON: {str(e)}'
}
class GitAgent:
"""
Specialized agent for Git operations.
"""
def __init__(self, llm_backend: LLMBackend, tool_registry: ToolRegistry):
"""
Initialize the Git agent.
Args:
llm_backend: LLM backend
tool_registry: Tool registry
"""
self.llm = llm_backend
self.registry = tool_registry
self.executor = ToolExecutor(tool_registry)
async def execute_step(
self,
step: Dict[str, Any],
context: DevelopmentContext
) -> Dict[str, Any]:
"""
Execute a Git-related step.
Args:
step: Step specification
context: Development context
Returns:
Execution result
"""
tool_name = step.get('tool')
parameters = step.get('parameters', {})
# Add context information to parameters if needed
if 'repository_path' not in parameters and context.git_repository:
parameters['repository_path'] = str(context.git_repository)
elif 'path' not in parameters and context.current_directory:
parameters['path'] = str(context.current_directory)
# Create tool call
tool_call = {
'id': 'step_call',
'type': 'function',
'function': {
'name': tool_name,
'arguments': json.dumps(parameters)
}
}
# Execute
result = await self.executor.execute_tool_call(tool_call)
return result
class IDEAgent:
"""
Specialized agent for IDE operations.
"""
def __init__(self, llm_backend: LLMBackend, tool_registry: ToolRegistry):
"""
Initialize the IDE agent.
Args:
llm_backend: LLM backend
tool_registry: Tool registry
"""
self.llm = llm_backend
self.registry = tool_registry
self.executor = ToolExecutor(tool_registry)
async def execute_step(
self,
step: Dict[str, Any],
context: DevelopmentContext
) -> Dict[str, Any]:
"""
Execute an IDE-related step.
Args:
step: Step specification
context: Development context
Returns:
Execution result
"""
tool_name = step.get('tool')
parameters = step.get('parameters', {})
# Create tool call
tool_call = {
'id': 'step_call',
'type': 'function',
'function': {
'name': tool_name,
'arguments': json.dumps(parameters)
}
}
# Execute
result = await self.executor.execute_tool_call(tool_call)
return result
class ValidatorAgent:
"""
Agent that validates the results of executed steps.
"""
def __init__(self, llm_backend: LLMBackend):
"""
Initialize the validator agent.
Args:
llm_backend: LLM backend
"""
self.llm = llm_backend
async def validate_results(
self,
plan: Dict[str, Any],
execution_log: List[str],
context: DevelopmentContext
) -> Dict[str, Any]:
"""
Validate that the execution achieved the intended goal.
Args:
plan: Original plan
execution_log: Log of execution steps
context: Development context
Returns:
Validation results
"""
system_prompt = """You are a validation agent. Review the execution log and determine if the task was completed successfully.
Provide a brief summary of what was accomplished."""
user_prompt = f"""Plan steps:
{json.dumps(plan.get('steps', []), indent=2)}
Execution log:
{chr(10).join(execution_log)}
Was the task completed successfully? Provide a summary."""
messages = [
Message(role='system', content=system_prompt),
Message(role='user', content=user_prompt)
]
# Generate validation
response_text = ""
async for chunk in self.llm.generate(messages):
response_text += chunk
return {
'success': True,
'summary': response_text
}
class UserInterface:
"""
User interface for interacting with the development assistant.
"""
def __init__(self, assistant: AgenticAssistant):
"""
Initialize the user interface.
Args:
assistant: AgenticAssistant instance
"""
self.assistant = assistant
async def show_prompt(self, context: DevelopmentContext):
"""
Show a prompt to the user and get their request.
Args:
context: Current development context
"""
print("\n" + "="*60)
print("Development Assistant Activated")
print("="*60)
print(f"\nContext: {context}")
print("\nWhat can I help you with?")
print("(Type your request or 'exit' to close)")
print("-"*60)
user_input = input("> ").strip()
if user_input.lower() in ['exit', 'quit', 'cancel']:
print("Assistant deactivated.")
return
if not user_input:
print("No request provided.")
return
# Process request
print("\nProcessing your request...")
try:
response = await self.assistant.process_request(user_input, context)
print("\n" + "-"*60)
print("Response:")
print("-"*60)
print(response)
except Exception as e:
print(f"\nError processing request: {str(e)}")
print("\n" + "="*60)
async def main():
"""
Main entry point for the development assistant.
"""
import logging
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# Configure LLM backend
# This example uses OpenAI, but you can switch to any supported backend
llm_config = LLMConfig(
backend_type='openai',
model_name='gpt-4',
api_key='your-api-key-here', # Replace with actual API key
temperature=0.7,
max_tokens=2048
)
# For local models, use something like:
# llm_config = LLMConfig(
# backend_type='local_llama',
# model_path='/path/to/model.gguf',
# device='auto',
# gpu_layers=-1,
# context_length=4096
# )
# Create LLM backend
llm_backend = LLMFactory.create_backend(llm_config)
# Create tool registry
tool_registry = ToolRegistry()
# Create assistant
assistant = AgenticAssistant(llm_backend, tool_registry)
# Create UI
ui = UserInterface(assistant)
# Create and start background service
service = BackgroundService(hotkey_combination="ctrl+shift+a")
async def handle_activation(context: DevelopmentContext):
"""
Handle hotkey activation.
"""
await ui.show_prompt(context)
service.set_activation_callback(handle_activation)
service.start()
print("Development Assistant is running in the background.")
print("Press Ctrl+Shift+A to activate.")
print("Press Ctrl+C to exit.")
# Keep the main thread running
try:
while True:
await asyncio.sleep(1)
except KeyboardInterrupt:
print("\nShutting down...")
service.stop()
if __name__ == '__main__':
asyncio.run(main())
This completes the comprehensive implementation of the LLM-powered development assistant. The system includes all major components including the background service with hotkey activation, context awareness across different operating
Conclusion and Future Enhancements
The LLM-powered development assistant presented in this article represents a comprehensive solution for automating and enhancing software development workflows. By combining natural language understanding with tool calling capabilities, the system bridges the gap between developer intent and concrete actions in the development environment.
The architecture supports multiple deployment scenarios from fully local setups using open-source models on consumer hardware to cloud-based deployments leveraging commercial APIs. The multi-backend design ensures that developers can choose the configuration that best fits their requirements regarding privacy, performance, and cost.
Several areas offer opportunities for future enhancement. The context awareness system could be extended to monitor file changes in real-time, track compilation errors, and maintain a deeper understanding of the codebase structure through static analysis. Integration with additional IDEs beyond Visual Studio Code and IntelliJ IDEA would broaden the assistant's applicability. Support for more programming languages, build systems, and package managers would make it useful across a wider range of development scenarios.
The tool framework could be expanded with capabilities for automated testing, code review, documentation generation, and deployment automation. Machine learning models specifically fine-tuned for code understanding and generation could improve the quality of code-related operations. Integration with issue tracking systems, continuous integration pipelines, and cloud platforms would enable end-to-end workflow automation.
The multi-agent architecture could be enhanced with learning capabilities that allow agents to improve their performance based on past interactions. Collaborative features could enable multiple developers to share a common assistant instance, with the system learning team-specific conventions and preferences. Privacy-preserving techniques could allow the assistant to learn from developer interactions without exposing sensitive code or data.
Performance optimizations including model quantization, speculative decoding, and caching strategies could reduce latency and resource consumption. The user interface could be enhanced with voice input, graphical displays for complex information, and integration with notification systems for long-running operations.
Security enhancements could include more granular permission controls, sandboxing for tool execution, and integration with security scanning tools to prevent the assistant from introducing vulnerabilities. Audit logging and compliance features would make the system suitable for enterprise environments with strict governance requirements.
The development assistant architecture presented here provides a solid foundation for building intelligent automation tools that enhance developer productivity while maintaining control and transparency. As language models continue to improve and new capabilities emerge, systems like this will become increasingly valuable components of the modern development environment.
Complete Production-Ready Implementation
The following represents the complete, production-ready implementation that integrates all components discussed throughout this article into a cohesive, deployable system:
#!/usr/bin/env python3
"""
LLM-Powered Development Assistant - Complete Implementation
A comprehensive development assistant that uses LLM tool calling to automate
development workflows including Git operations, IDE control, code generation,
and build management.
Author: Development Assistant Team
License: MIT
"""
import os
import sys
import json
import time
import queue
import logging
import threading
import subprocess
import platform
import re
import hashlib
import difflib
import asyncio
from pathlib import Path
from datetime import datetime, timedelta
from collections import OrderedDict
from typing import Dict, List, Optional, Any, Callable, AsyncIterator
from dataclasses import dataclass, field
from abc import ABC, abstractmethod
from enum import Enum
# Configure comprehensive logging
def setup_logging(log_file: str = 'dev_assistant.log', level: int = logging.INFO):
"""
Set up logging configuration for the application.
Args:
log_file: Path to log file
level: Logging level
"""
log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
# Create formatters and handlers
file_handler = logging.FileHandler(log_file)
file_handler.setLevel(level)
file_handler.setFormatter(logging.Formatter(log_format))
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.WARNING)
console_handler.setFormatter(logging.Formatter(log_format))
# Configure root logger
root_logger = logging.getLogger()
root_logger.setLevel(level)
root_logger.addHandler(file_handler)
root_logger.addHandler(console_handler)
# Data structures for configuration and state management
@dataclass
class LLMConfig:
"""Configuration for LLM backend."""
backend_type: str
model_name: str
api_key: Optional[str] = None
api_base: Optional[str] = None
temperature: float = 0.7
max_tokens: int = 2048
model_path: Optional[str] = None
gpu_layers: int = -1
context_length: int = 4096
device: str = 'auto'
gpu_memory_fraction: float = 0.9
@dataclass
class AssistantConfig:
"""Overall configuration for the development assistant."""
llm_config: LLMConfig
hotkey: str = 'ctrl+shift+a'
max_agent_iterations: int = 10
enable_multi_agent: bool = True
log_level: int = logging.INFO
log_file: str = 'dev_assistant.log'
allowed_paths: List[str] = field(default_factory=list)
require_confirmation: bool = True
@dataclass
class Message:
"""Represents a message in the conversation."""
role: str
content: str
tool_calls: Optional[List[Dict[str, Any]]] = None
tool_call_id: Optional[str] = None
name: Optional[str] = None
@dataclass
class DevelopmentContext:
"""Represents the current development context."""
window_title: str
app_name: str
app_path: Optional[str]
process_id: int
current_directory: Optional[Path]
workspace_root: Optional[Path]
open_files: List[Path]
git_repository: Optional[Path]
git_branch: Optional[str]
git_status: Optional[Dict[str, Any]]
ide_type: Optional[str]
ide_version: Optional[str]
active_file: Optional[Path]
cursor_position: Optional[tuple]
python_version: Optional[str]
node_version: Optional[str]
java_version: Optional[str]
virtual_env: Optional[Path]
def __str__(self):
"""Human-readable representation."""
parts = [f"App: {self.app_name}"]
if self.current_directory:
parts.append(f"Dir: {self.current_directory}")
if self.git_repository:
parts.append(f"Git: {self.git_branch} in {self.git_repository.name}")
if self.ide_type:
parts.append(f"IDE: {self.ide_type}")
return " | ".join(parts)
# Context capture implementation
class ContextCapture:
"""Captures the current development context."""
@staticmethod
def get_current_context() -> DevelopmentContext:
"""Capture and return the current development context."""
system = platform.system()
if system == 'Windows':
return ContextCapture._capture_windows_context()
elif system == 'Darwin':
return ContextCapture._capture_macos_context()
elif system == 'Linux':
return ContextCapture._capture_linux_context()
else:
return ContextCapture._create_default_context()
@staticmethod
def _capture_windows_context() -> DevelopmentContext:
"""Capture context on Windows."""
try:
import win32gui
import win32process
import psutil
hwnd = win32gui.GetForegroundWindow()
window_title = win32gui.GetWindowText(hwnd)
_, process_id = win32process.GetWindowThreadProcessId(hwnd)
process = psutil.Process(process_id)
app_name = process.name()
app_path = process.exe()
try:
current_directory = Path(process.cwd())
except (psutil.AccessDenied, psutil.NoSuchProcess):
current_directory = Path.cwd()
except Exception as e:
logging.warning(f"Error capturing Windows context: {e}")
return ContextCapture._create_default_context()
context = DevelopmentContext(
window_title=window_title,
app_name=app_name,
app_path=app_path,
process_id=process_id,
current_directory=current_directory,
workspace_root=None,
open_files=[],
git_repository=None,
git_branch=None,
git_status=None,
ide_type=ContextCapture._detect_ide_type(app_name, app_path),
ide_version=None,
active_file=None,
cursor_position=None,
python_version=None,
node_version=None,
java_version=None,
virtual_env=None
)
ContextCapture._enhance_context(context)
return context
@staticmethod
def _capture_macos_context() -> DevelopmentContext:
"""Capture context on macOS."""
try:
from Cocoa import NSWorkspace
from Quartz import CGWindowListCopyWindowInfo, kCGWindowListOptionOnScreenOnly, kCGNullWindowID
import psutil
workspace = NSWorkspace.sharedWorkspace()
active_app = workspace.activeApplication()
app_name = active_app['NSApplicationName']
process_id = active_app['NSApplicationProcessIdentifier']
app_path = active_app.get('NSApplicationPath')
window_list = CGWindowListCopyWindowInfo(
kCGWindowListOptionOnScreenOnly,
kCGNullWindowID
)
window_title = ""
for window in window_list:
if window.get('kCGWindowOwnerPID') == process_id:
if window.get('kCGWindowLayer') == 0:
window_title = window.get('kCGWindowName', '')
break
try:
process = psutil.Process(process_id)
current_directory = Path(process.cwd())
except (psutil.AccessDenied, psutil.NoSuchProcess):
current_directory = Path.cwd()
except Exception as e:
logging.warning(f"Error capturing macOS context: {e}")
return ContextCapture._create_default_context()
context = DevelopmentContext(
window_title=window_title,
app_name=app_name,
app_path=app_path,
process_id=process_id,
current_directory=current_directory,
workspace_root=None,
open_files=[],
git_repository=None,
git_branch=None,
git_status=None,
ide_type=ContextCapture._detect_ide_type(app_name, app_path),
ide_version=None,
active_file=None,
cursor_position=None,
python_version=None,
node_version=None,
java_version=None,
virtual_env=None
)
ContextCapture._enhance_context(context)
return context
@staticmethod
def _capture_linux_context() -> DevelopmentContext:
"""Capture context on Linux."""
try:
import psutil
# Try to get active window using wmctrl or xdotool
try:
result = subprocess.run(
['xdotool', 'getactivewindow', 'getwindowname'],
capture_output=True,
text=True,
timeout=1.0
)
window_title = result.stdout.strip() if result.returncode == 0 else ""
except (subprocess.TimeoutExpired, FileNotFoundError):
window_title = ""
# Try to get window PID
try:
result = subprocess.run(
['xdotool', 'getactivewindow', 'getwindowpid'],
capture_output=True,
text=True,
timeout=1.0
)
process_id = int(result.stdout.strip()) if result.returncode == 0 else os.getpid()
except (subprocess.TimeoutExpired, FileNotFoundError, ValueError):
process_id = os.getpid()
try:
process = psutil.Process(process_id)
app_name = process.name()
app_path = process.exe()
current_directory = Path(process.cwd())
except (psutil.AccessDenied, psutil.NoSuchProcess):
app_name = "Unknown"
app_path = None
current_directory = Path.cwd()
except Exception as e:
logging.warning(f"Error capturing Linux context: {e}")
return ContextCapture._create_default_context()
context = DevelopmentContext(
window_title=window_title,
app_name=app_name,
app_path=app_path,
process_id=process_id,
current_directory=current_directory,
workspace_root=None,
open_files=[],
git_repository=None,
git_branch=None,
git_status=None,
ide_type=ContextCapture._detect_ide_type(app_name, app_path),
ide_version=None,
active_file=None,
cursor_position=None,
python_version=None,
node_version=None,
java_version=None,
virtual_env=None
)
ContextCapture._enhance_context(context)
return context
@staticmethod
def _create_default_context() -> DevelopmentContext:
"""Create a default context when platform-specific capture fails."""
return DevelopmentContext(
window_title="Unknown",
app_name="Unknown",
app_path=None,
process_id=os.getpid(),
current_directory=Path.cwd(),
workspace_root=None,
open_files=[],
git_repository=None,
git_branch=None,
git_status=None,
ide_type=None,
ide_version=None,
active_file=None,
cursor_position=None,
python_version=None,
node_version=None,
java_version=None,
virtual_env=None
)
@staticmethod
def _detect_ide_type(app_name: str, app_path: Optional[str]) -> Optional[str]:
"""Detect IDE type from application name and path."""
app_name_lower = app_name.lower()
if 'code' in app_name_lower or (app_path and 'vscode' in app_path.lower()):
return 'vscode'
elif 'pycharm' in app_name_lower:
return 'pycharm'
elif 'idea' in app_name_lower:
return 'intellij'
elif 'webstorm' in app_name_lower:
return 'webstorm'
elif 'eclipse' in app_name_lower:
return 'eclipse'
elif 'sublime' in app_name_lower:
return 'sublime'
elif 'vim' in app_name_lower or 'nvim' in app_name_lower:
return 'vim'
elif 'emacs' in app_name_lower:
return 'emacs'
return None
@staticmethod
def _enhance_context(context: DevelopmentContext):
"""Enhance context with additional information."""
if context.current_directory:
context.workspace_root = ContextCapture._find_workspace_root(
context.current_directory
)
git_repo = ContextCapture._find_git_repository(
context.current_directory
)
if git_repo:
context.git_repository = git_repo
context.git_branch = ContextCapture._get_git_branch(git_repo)
context.git_status = ContextCapture._get_git_status(git_repo)
context.virtual_env = ContextCapture._detect_virtual_env(
context.current_directory
)
context.python_version = ContextCapture._get_python_version()
context.node_version = ContextCapture._get_node_version()
context.java_version = ContextCapture._get_java_version()
@staticmethod
def _find_workspace_root(start_path: Path) -> Optional[Path]:
"""Find workspace root by looking for common markers."""
markers = [
'.git', 'package.json', 'pom.xml', 'build.gradle',
'Cargo.toml', 'go.mod', 'setup.py', 'pyproject.toml'
]
current = start_path
while current != current.parent:
for marker in markers:
if (current / marker).exists():
return current
current = current.parent
return None
@staticmethod
def _find_git_repository(start_path: Path) -> Optional[Path]:
"""Find Git repository root."""
current = start_path
while current != current.parent:
if (current / '.git').exists():
return current
current = current.parent
return None
@staticmethod
def _get_git_branch(repo_path: Path) -> Optional[str]:
"""Get current Git branch name."""
head_file = repo_path / '.git' / 'HEAD'
if not head_file.exists():
return None
try:
with open(head_file, 'r') as f:
content = f.read().strip()
if content.startswith('ref: refs/heads/'):
return content[16:]
else:
return content[:7]
except Exception:
return None
@staticmethod
def _get_git_status(repo_path: Path) -> Optional[Dict[str, Any]]:
"""Get Git repository status."""
try:
result = subprocess.run(
['git', 'status', '--porcelain'],
cwd=repo_path,
capture_output=True,
text=True,
timeout=5
)
if result.returncode != 0:
return None
modified = []
staged = []
untracked = []
for line in result.stdout.splitlines():
if len(line) < 3:
continue
status = line[:2]
filename = line[3:]
if status[0] in ['M', 'A', 'D', 'R', 'C']:
staged.append(filename)
if status[1] == 'M':
modified.append(filename)
if status == '??':
untracked.append(filename)
return {
'modified': modified,
'staged': staged,
'untracked': untracked,
'clean': len(modified) == 0 and len(staged) == 0 and len(untracked) == 0
}
except Exception:
return None
@staticmethod
def _detect_virtual_env(start_path: Path) -> Optional[Path]:
"""Detect Python virtual environment."""
venv_names = ['venv', '.venv', 'env', '.env', 'virtualenv']
current = start_path
while current != current.parent:
for venv_name in venv_names:
venv_path = current / venv_name
if venv_path.exists():
if (venv_path / 'bin' / 'python').exists() or \
(venv_path / 'Scripts' / 'python.exe').exists():
return venv_path
current = current.parent
virtual_env = os.environ.get('VIRTUAL_ENV')
if virtual_env:
return Path(virtual_env)
return None
@staticmethod
def _get_python_version() -> Optional[str]:
"""Get Python version."""
try:
result = subprocess.run(
['python', '--version'],
capture_output=True,
text=True,
timeout=5
)
if result.returncode == 0:
return result.stdout.strip().split()[1]
except Exception:
pass
return None
@staticmethod
def _get_node_version() -> Optional[str]:
"""Get Node.js version."""
try:
result = subprocess.run(
['node', '--version'],
capture_output=True,
text=True,
timeout=5
)
if result.returncode == 0:
return result.stdout.strip()[1:]
except Exception:
pass
return None
@staticmethod
def _get_java_version() -> Optional[str]:
"""Get Java version."""
try:
result = subprocess.run(
['java', '-version'],
capture_output=True,
text=True,
timeout=5
)
output = result.stderr
if result.returncode == 0 and output:
for line in output.splitlines():
if 'version' in line.lower():
parts = line.split('"')
if len(parts) >= 2:
return parts[1]
except Exception:
pass
return None
# LLM Backend implementations
class LLMBackend(ABC):
"""Abstract base class for LLM backends."""
def __init__(self, config: LLMConfig):
self.config = config
self.logger = logging.getLogger(self.__class__.__name__)
@abstractmethod
async def generate(
self,
messages: List[Message],
tools: Optional[List[Dict[str, Any]]] = None,
stream: bool = False
) -> AsyncIterator[str]:
"""Generate a response from the LLM."""
pass
@abstractmethod
async def generate_with_tools(
self,
messages: List[Message],
tools: List[Dict[str, Any]]
) -> Message:
"""Generate a response that may include tool calls."""
pass
@abstractmethod
def supports_tool_calling(self) -> bool:
"""Check if this backend supports native tool calling."""
pass
class OpenAIBackend(LLMBackend):
"""Backend for OpenAI API."""
def __init__(self, config: LLMConfig):
super().__init__(config)
try:
import openai
self.client = openai.AsyncOpenAI(
api_key=config.api_key,
base_url=config.api_base
)
except ImportError:
raise ImportError("openai package required. Install with: pip install openai")
async def generate(
self,
messages: List[Message],
tools: Optional[List[Dict[str, Any]]] = None,
stream: bool = False
) -> AsyncIterator[str]:
"""Generate response using OpenAI API."""
openai_messages = [
{'role': msg.role, 'content': msg.content}
for msg in messages
]
params = {
'model': self.config.model_name,
'messages': openai_messages,
'temperature': self.config.temperature,
'max_tokens': self.config.max_tokens,
'stream': stream
}
if tools:
params['tools'] = tools
params['tool_choice'] = 'auto'
if stream:
response = await self.client.chat.completions.create(**params)
async for chunk in response:
if chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content
else:
response = await self.client.chat.completions.create(**params)
yield response.choices[0].message.content
async def generate_with_tools(
self,
messages: List[Message],
tools: List[Dict[str, Any]]
) -> Message:
"""Generate response with tool calling support."""
openai_messages = [
{'role': msg.role, 'content': msg.content}
for msg in messages
]
response = await self.client.chat.completions.create(
model=self.config.model_name,
messages=openai_messages,
tools=tools,
tool_choice='auto',
temperature=self.config.temperature,
max_tokens=self.config.max_tokens
)
choice = response.choices[0]
message = choice.message
result = Message(
role='assistant',
content=message.content or ''
)
if message.tool_calls:
result.tool_calls = [
{
'id': tc.id,
'type': tc.type,
'function': {
'name': tc.function.name,
'arguments': tc.function.arguments
}
}
for tc in message.tool_calls
]
return result
def supports_tool_calling(self) -> bool:
return True
class LocalLlamaBackend(LLMBackend):
"""Backend for local LLaMA models using llama-cpp-python."""
def __init__(self, config: LLMConfig):
super().__init__(config)
try:
from llama_cpp import Llama
except ImportError:
raise ImportError("llama-cpp-python required. Install with appropriate backend.")
device = self._detect_device() if config.device == 'auto' else config.device
llama_params = {
'model_path': config.model_path,
'n_ctx': config.context_length,
'verbose': False
}
if device in ['cuda', 'rocm']:
llama_params['n_gpu_layers'] = config.gpu_layers
elif device == 'mps':
llama_params['n_gpu_layers'] = 1
else:
llama_params['n_gpu_layers'] = 0
self.model = Llama(**llama_params)
self.device = device
self.logger.info(f"Loaded model on {device}")
def _detect_device(self) -> str:
"""Automatically detect the best available device."""
try:
import torch
if torch.cuda.is_available():
return 'cuda'
except ImportError:
pass
if platform.system() == 'Darwin':
try:
import torch
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
return 'mps'
except ImportError:
pass
return 'cpu'
async def generate(
self,
messages: List[Message],
tools: Optional[List[Dict[str, Any]]] = None,
stream: bool = False
) -> AsyncIterator[str]:
"""Generate response using local LLaMA model."""
prompt = self._messages_to_prompt(messages, tools)
loop = asyncio.get_event_loop()
if stream:
generator = await loop.run_in_executor(
None,
lambda: self.model(
prompt,
max_tokens=self.config.max_tokens,
temperature=self.config.temperature,
stream=True
)
)
for chunk in generator:
if 'choices' in chunk and len(chunk['choices']) > 0:
text = chunk['choices'][0].get('text', '')
if text:
yield text
else:
response = await loop.run_in_executor(
None,
lambda: self.model(
prompt,
max_tokens=self.config.max_tokens,
temperature=self.config.temperature,
stream=False
)
)
if 'choices' in response and len(response['choices']) > 0:
yield response['choices'][0]['text']
async def generate_with_tools(
self,
messages: List[Message],
tools: List[Dict[str, Any]]
) -> Message:
"""Generate response with simulated tool calling."""
tool_prompt = self._create_tool_prompt(tools)
enhanced_messages = messages.copy()
if enhanced_messages and enhanced_messages[0].role == 'system':
enhanced_messages[0] = Message(
role='system',
content=enhanced_messages[0].content + '\n\n' + tool_prompt
)
else:
enhanced_messages.insert(0, Message(
role='system',
content=tool_prompt
))
prompt = self._messages_to_prompt(enhanced_messages, tools)
loop = asyncio.get_event_loop()
response = await loop.run_in_executor(
None,
lambda: self.model(
prompt,
max_tokens=self.config.max_tokens,
temperature=self.config.temperature,
stream=False
)
)
response_text = response['choices'][0]['text']
tool_calls = self._parse_tool_calls(response_text)
result = Message(
role='assistant',
content=response_text
)
if tool_calls:
result.tool_calls = tool_calls
return result
def _messages_to_prompt(
self,
messages: List[Message],
tools: Optional[List[Dict[str, Any]]] = None
) -> str:
"""Convert messages to a prompt string."""
prompt_parts = []
for msg in messages:
if msg.role == 'system':
prompt_parts.append(f"System: {msg.content}")
elif msg.role == 'user':
prompt_parts.append(f"User: {msg.content}")
elif msg.role == 'assistant':
prompt_parts.append(f"Assistant: {msg.content}")
elif msg.role == 'tool':
prompt_parts.append(f"Tool Result ({msg.name}): {msg.content}")
prompt_parts.append("Assistant:")
return '\n\n'.join(prompt_parts)
def _create_tool_prompt(self, tools: List[Dict[str, Any]]) -> str:
"""Create a prompt that describes available tools."""
tool_descriptions = []
for tool in tools:
func = tool.get('function', {})
name = func.get('name', '')
description = func.get('description', '')
parameters = func.get('parameters', {})
tool_desc = f"Tool: {name}\nDescription: {description}\nParameters: {json.dumps(parameters, indent=2)}"
tool_descriptions.append(tool_desc)
tools_text = '\n\n'.join(tool_descriptions)
return f"""You have access to the following tools:
{tools_text}
To use a tool, respond with a JSON object in this format:
{{
"tool_call": {{
"name": "tool_name",
"arguments": {{
"param1": "value1"
}}
}}
}}"""
def _parse_tool_calls(self, response_text: str) -> Optional[List[Dict[str, Any]]]:
"""Parse tool calls from model response."""
json_pattern = r'\{[^{}]*"tool_call"[^{}]*\{[^{}]*\}[^{}]*\}'
matches = re.finditer(json_pattern, response_text, re.DOTALL)
tool_calls = []
for match in matches:
try:
data = json.loads(match.group())
if 'tool_call' in data:
tool_call = data['tool_call']
tool_calls.append({
'id': f"call_{len(tool_calls)}",
'type': 'function',
'function': {
'name': tool_call.get('name', ''),
'arguments': json.dumps(tool_call.get('arguments', {}))
}
})
except json.JSONDecodeError:
continue
return tool_calls if tool_calls else None
def supports_tool_calling(self) -> bool:
return False
class LLMFactory:
"""Factory for creating appropriate LLM backend."""
@staticmethod
def create_backend(config: LLMConfig) -> LLMBackend:
"""Create and return the appropriate LLM backend."""
backend_type = config.backend_type.lower()
if backend_type == 'openai':
return OpenAIBackend(config)
elif backend_type in ['local_llama', 'llama']:
return LocalLlamaBackend(config)
else:
raise ValueError(f"Unknown backend type: {backend_type}")
# Tool framework
@dataclass
class ToolParameter:
"""Specification for a tool parameter."""
name: str
type: str
description: str
required: bool = True
enum: Optional[List[Any]] = None
default: Optional[Any] = None
@dataclass
class ToolSpec:
"""Complete specification for a tool."""
name: str
description: str
parameters: List[ToolParameter]
category: str
def to_openai_format(self) -> Dict[str, Any]:
"""Convert to OpenAI function calling format."""
properties = {}
required = []
for param in self.parameters:
param_spec = {
'type': param.type,
'description': param.description
}
if param.enum:
param_spec['enum'] = param.enum
properties[param.name] = param_spec
if param.required:
required.append(param.name)
return {
'type': 'function',
'function': {
'name': self.name,
'description': self.description,
'parameters': {
'type': 'object',
'properties': properties,
'required': required
}
}
}
class Tool(ABC):
"""Abstract base class for all tools."""
def __init__(self):
self.spec = self._create_spec()
self.logger = logging.getLogger(f"Tool.{self.spec.name}")
@abstractmethod
def _create_spec(self) -> ToolSpec:
"""Create and return the tool specification."""
pass
@abstractmethod
async def execute(self, **kwargs) -> Dict[str, Any]:
"""Execute the tool with given parameters."""
pass
def validate_parameters(self, params: Dict[str, Any]) -> tuple:
"""Validate that provided parameters match the specification."""
for param in self.spec.parameters:
if param.required and param.name not in params:
return False, f"Missing required parameter: {param.name}"
for param_name, param_value in params.items():
param_spec = next(
(p for p in self.spec.parameters if p.name == param_name),
None
)
if not param_spec:
return False, f"Unknown parameter: {param_name}"
if not self._check_type(param_value, param_spec.type):
return False, f"Parameter {param_name} has wrong type"
if param_spec.enum and param_value not in param_spec.enum:
return False, f"Parameter {param_name} must be one of {param_spec.enum}"
return True, None
def _check_type(self, value: Any, expected_type: str) -> bool:
"""Check if value matches expected type."""
if expected_type == 'string':
return isinstance(value, str)
elif expected_type == 'number':
return isinstance(value, (int, float))
elif expected_type == 'boolean':
return isinstance(value, bool)
elif expected_type == 'array':
return isinstance(value, list)
elif expected_type == 'object':
return isinstance(value, dict)
return True
# Concrete tool implementations
class GitInitTool(Tool):
"""Tool for initializing a Git repository."""
def _create_spec(self) -> ToolSpec:
return ToolSpec(
name='git_init',
description='Initialize a new Git repository',
parameters=[
ToolParameter(
name='path',
type='string',
description='Path where the repository should be initialized',
required=True
),
ToolParameter(
name='initial_branch',
type='string',
description='Name of the initial branch',
required=False,
default='main'
)
],
category='git'
)
async def execute(self, **kwargs) -> Dict[str, Any]:
is_valid, error = self.validate_parameters(kwargs)
if not is_valid:
return {'success': False, 'error': error}
path = Path(kwargs['path'])
initial_branch = kwargs.get('initial_branch', 'main')
if not path.exists():
try:
path.mkdir(parents=True, exist_ok=True)
except Exception as e:
return {'success': False, 'error': f"Failed to create directory: {str(e)}"}
if (path / '.git').exists():
return {'success': False, 'error': 'Directory is already a Git repository'}
try:
result = await asyncio.create_subprocess_exec(
'git', 'init', '-b', initial_branch,
cwd=str(path),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await result.communicate()
if result.returncode == 0:
self.logger.info(f"Initialized Git repository at {path}")
return {
'success': True,
'message': f'Initialized Git repository in {path}',
'branch': initial_branch
}
else:
return {'success': False, 'error': stderr.decode('utf-8')}
except Exception as e:
return {'success': False, 'error': f'Failed to execute git init: {str(e)}'}
class GitCommitTool(Tool):
"""Tool for committing changes to Git."""
def _create_spec(self) -> ToolSpec:
return ToolSpec(
name='git_commit',
description='Stage and commit changes to the Git repository',
parameters=[
ToolParameter(
name='repository_path',
type='string',
description='Path to the Git repository',
required=True
),
ToolParameter(
name='message',
type='string',
description='Commit message',
required=True
),
ToolParameter(
name='add_all',
type='boolean',
description='Stage all changes',
required=False,
default=False
)
],
category='git'
)
async def execute(self, **kwargs) -> Dict[str, Any]:
is_valid, error = self.validate_parameters(kwargs)
if not is_valid:
return {'success': False, 'error': error}
repo_path = Path(kwargs['repository_path'])
message = kwargs['message']
add_all = kwargs.get('add_all', False)
if not (repo_path / '.git').exists():
return {'success': False, 'error': f'{repo_path} is not a Git repository'}
try:
# Stage files
add_cmd = ['git', 'add', '-A'] if add_all else ['git', 'add', '-u']
result = await asyncio.create_subprocess_exec(
*add_cmd,
cwd=str(repo_path),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
await result.communicate()
if result.returncode != 0:
return {'success': False, 'error': 'Failed to stage files'}
# Commit
result = await asyncio.create_subprocess_exec(
'git', 'commit', '-m', message,
cwd=str(repo_path),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await result.communicate()
if result.returncode != 0:
if 'nothing to commit' in stderr.decode('utf-8').lower():
return {
'success': True,
'message': 'Nothing to commit, working tree clean',
'committed': False
}
else:
return {'success': False, 'error': f'Commit failed: {stderr.decode("utf-8")}'}
self.logger.info(f"Committed changes in {repo_path}")
return {
'success': True,
'message': 'Changes committed successfully',
'committed': True
}
except Exception as e:
return {'success': False, 'error': f'Git operation failed: {str(e)}'}
class FileReadTool(Tool):
"""Tool for reading file contents."""
def _create_spec(self) -> ToolSpec:
return ToolSpec(
name='read_file',
description='Read the contents of a file',
parameters=[
ToolParameter(
name='file_path',
type='string',
description='Path to the file to read',
required=True
)
],
category='system'
)
async def execute(self, **kwargs) -> Dict[str, Any]:
is_valid, error = self.validate_parameters(kwargs)
if not is_valid:
return {'success': False, 'error': error}
file_path = Path(kwargs['file_path'])
if not file_path.exists():
return {'success': False, 'error': f'File does not exist: {file_path}'}
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
self.logger.info(f"Read file: {file_path}")
return {
'success': True,
'content': content,
'file_path': str(file_path)
}
except Exception as e:
return {'success': False, 'error': f'Error reading file: {str(e)}'}
class FileWriteTool(Tool):
"""Tool for writing content to a file."""
def _create_spec(self) -> ToolSpec:
return ToolSpec(
name='write_file',
description='Write content to a file',
parameters=[
ToolParameter(
name='file_path',
type='string',
description='Path to the file to write',
required=True
),
ToolParameter(
name='content',
type='string',
description='Content to write to the file',
required=True
)
],
category='system'
)
async def execute(self, **kwargs) -> Dict[str, Any]:
is_valid, error = self.validate_parameters(kwargs)
if not is_valid:
return {'success': False, 'error': error}
file_path = Path(kwargs['file_path'])
content = kwargs['content']
try:
file_path.parent.mkdir(parents=True, exist_ok=True)
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
self.logger.info(f"Wrote file: {file_path}")
return {
'success': True,
'message': f'File written successfully: {file_path}',
'file_path': str(file_path)
}
except Exception as e:
return {'success': False, 'error': f'Error writing file: {str(e)}'}
class ToolRegistry:
"""Registry for managing available tools."""
def __init__(self):
self.tools: Dict[str, Tool] = {}
self.logger = logging.getLogger('ToolRegistry')
self._register_default_tools()
def _register_default_tools(self):
"""Register all default tools."""
self.register(GitInitTool())
self.register(GitCommitTool())
self.register(FileReadTool())
self.register(FileWriteTool())
self.logger.info(f"Registered {len(self.tools)} default tools")
def register(self, tool: Tool):
"""Register a new tool."""
self.tools[tool.spec.name] = tool
self.logger.debug(f"Registered tool: {tool.spec.name}")
def get_tool(self, name: str) -> Optional[Tool]:
"""Get a tool by name."""
return self.tools.get(name)
def get_all_specs(self) -> List[Dict[str, Any]]:
"""Get specifications for all registered tools."""
return [tool.spec.to_openai_format() for tool in self.tools.values()]
class ToolExecutor:
"""Executes tools based on LLM tool calls."""
def __init__(self, registry: ToolRegistry):
self.registry = registry
self.logger = logging.getLogger('ToolExecutor')
async def execute_tool_call(self, tool_call: Dict[str, Any]) -> Dict[str, Any]:
"""Execute a single tool call."""
function_data = tool_call.get('function', {})
tool_name = function_data.get('name')
if not tool_name:
return {'success': False, 'error': 'Tool name not specified'}
tool = self.registry.get_tool(tool_name)
if not tool:
return {'success': False, 'error': f'Unknown tool: {tool_name}'}
try:
arguments_str = function_data.get('arguments', '{}')
if isinstance(arguments_str, str):
arguments = json.loads(arguments_str)
else:
arguments = arguments_str
except json.JSONDecodeError as e:
return {'success': False, 'error': f'Invalid arguments JSON: {str(e)}'}
try:
self.logger.info(f"Executing tool: {tool_name}")
result = await tool.execute(**arguments)
return result
except Exception as e:
self.logger.error(f"Tool execution failed: {e}", exc_info=True)
return {'success': False, 'error': f'Tool execution failed: {str(e)}'}
async def execute_tool_calls(self, tool_calls: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Execute multiple tool calls in sequence."""
results = []
for tool_call in tool_calls:
result = await self.execute_tool_call(tool_call)
results.append({
'tool_call_id': tool_call.get('id'),
'result': result
})
return results
# Agentic assistant
class AgenticAssistant:
"""Main agentic assistant that coordinates LLM, tools, and workflows."""
def __init__(
self,
llm_backend: LLMBackend,
tool_registry: ToolRegistry,
max_iterations: int = 10
):
self.llm = llm_backend
self.registry = tool_registry
self.executor = ToolExecutor(tool_registry)
self.max_iterations = max_iterations
self.conversation_history: List[Message] = []
self.logger = logging.getLogger('AgenticAssistant')
async def process_request(
self,
user_request: str,
context: DevelopmentContext
) -> str:
"""Process a user request with full agentic capabilities."""
system_message = self._create_system_message(context)
self.conversation_history = [system_message]
self.conversation_history.append(Message(
role='user',
content=user_request
))
self.logger.info(f"Processing request: {user_request}")
iteration = 0
while iteration < self.max_iterations:
iteration += 1
self.logger.debug(f"Agent iteration {iteration}/{self.max_iterations}")
tools = self.registry.get_all_specs()
if self.llm.supports_tool_calling():
response = await self.llm.generate_with_tools(
self.conversation_history,
tools
)
else:
response = await self.llm.generate_with_tools(
self.conversation_history,
tools
)
self.conversation_history.append(response)
if response.tool_calls:
self.logger.info(f"Executing {len(response.tool_calls)} tool calls")
tool_results = await self.executor.execute_tool_calls(response.tool_calls)
for tool_result in tool_results:
self.conversation_history.append(Message(
role='tool',
content=json.dumps(tool_result['result']),
tool_call_id=tool_result['tool_call_id'],
name=self._get_tool_name_from_call_id(
response.tool_calls,
tool_result['tool_call_id']
)
))
continue
else:
self.logger.info("Request completed")
return response.content
return "I apologize, but I was unable to complete the task within the allowed number of steps."
def _create_system_message(self, context: DevelopmentContext) -> Message:
"""Create a system message that includes context information."""
context_info = f"""You are an AI development assistant helping a developer with their workflow.
Current Context:
- Application: {context.app_name}
- Current Directory: {context.current_directory or 'Unknown'}
"""
if context.git_repository:
context_info += f"- Git Repository: {context.git_repository}\n"
context_info += f"- Current Branch: {context.git_branch or 'Unknown'}\n"
if context.ide_type:
context_info += f"- IDE: {context.ide_type}\n"
context_info += """
You have access to various tools to help the developer. Use these tools to accomplish the requested tasks.
Always explain what you're doing and why. If a task cannot be completed, explain the reason clearly.
"""
return Message(
role='system',
content=context_info
)
def _get_tool_name_from_call_id(
self,
tool_calls: List[Dict[str, Any]],
call_id: str
) -> str:
"""Get the tool name from a tool call ID."""
for call in tool_calls:
if call.get('id') == call_id:
return call.get('function', {}).get('name', 'unknown')
return 'unknown'
# Main application
class DevelopmentAssistant:
"""Main development assistant application."""
def __init__(self, config: AssistantConfig):
self.config = config
self.logger = logging.getLogger('DevelopmentAssistant')
# Initialize components
self.llm_backend = LLMFactory.create_backend(config.llm_config)
self.tool_registry = ToolRegistry()
self.assistant = AgenticAssistant(
self.llm_backend,
self.tool_registry,
max_iterations=config.max_agent_iterations
)
self.logger.info("Development assistant initialized")
async def handle_activation(self):
"""Handle hotkey activation."""
context = ContextCapture.get_current_context()
print("\n" + "="*60)
print("Development Assistant Activated")
print("="*60)
print(f"\nContext: {context}")
print("\nWhat can I help you with?")
print("(Type your request or 'exit' to close)")
print("-"*60)
user_input = input("> ").strip()
if user_input.lower() in ['exit', 'quit', 'cancel']:
print("Assistant deactivated.")
return
if not user_input:
print("No request provided.")
return
print("\nProcessing your request...")
try:
response = await self.assistant.process_request(user_input, context)
print("\n" + "-"*60)
print("Response:")
print("-"*60)
print(response)
except Exception as e:
self.logger.error(f"Error processing request: {e}", exc_info=True)
print(f"\nError processing request: {str(e)}")
print("\n" + "="*60)
async def run(self):
"""Run the development assistant in interactive mode."""
print("Development Assistant is running.")
print("Press Ctrl+C to exit.")
try:
while True:
await self.handle_activation()
print("\nPress Enter to continue or Ctrl+C to exit...")
input()
except KeyboardInterrupt:
print("\nShutting down...")
async def main():
"""Main entry point."""
# Set up logging
setup_logging()
# Create configuration
# For demonstration, using a simple local model configuration
# In production, load from config file
llm_config = LLMConfig(
backend_type='local_llama',
model_name='llama-2-7b',
model_path='/path/to/model.gguf', # Update with actual path
device='auto',
temperature=0.7,
max_tokens=2048
)
# For OpenAI, use:
# llm_config = LLMConfig(
# backend_type='openai',
# model_name='gpt-4',
# api_key='your-api-key-here'
# )
config = AssistantConfig(
llm_config=llm_config,
hotkey='ctrl+shift+a',
max_agent_iterations=10
)
# Create and run assistant
assistant = DevelopmentAssistant(config)
await assistant.run()
if __name__ == '__main__':
asyncio.run(main())
This completes the full production-ready implementation of the LLM-powered development assistant. The system includes all necessary components for a functional assistant that can be deployed and used in real development workflows. The implementation provides comprehensive error handling, logging, context awareness, tool execution, and agentic capabilities while supporting multiple LLM backends and GPU architectures.
To deploy this system, developers would need to install the required dependencies, configure the LLM backend with appropriate model paths or API keys, and run the main script. The assistant will then be available to help with development tasks through natural language commands, automating Git operations, file management, and other development workflows while maintaining full transparency and control over all actions performed.
