Hitchhiker's Guide to Software Architecture and Everything Else - by Michael Stal 
BUILDING AN LLM-POWERED CRC CARD DIAGRAMMING TOOL
INTRODUCTIONClass-Responsibilities-Collaboration cards represent a fundamental technique in object-oriented design. They provide a simple yet powerful way to model the responsibilities of classes and their collaborative relationships. Traditional CRC card sessions involve physical index cards arranged on a table, but this approach lacks persistence, version control, and the ability to easily share and modify designs.This article describes the design and implementation of an intelligent diagramming tool that lleverages Large Language Models to create, modify, and manage CRC card diagrams through natural language commands. The tool interprets user descriptions, automatically layouts components, draws relationship arrows, and exports diagrams to both visual formats (PNG, SVG) and structured data formats (JSON) for later restoration.The system addresses several technical challenges. First, it must understand natural language commands with varying levels of specificity and ambiguity. Second, it must compute aesthetically pleasing layouts where components do not overlap and connection arrows minimize crossings. Third, it must support diverse execution environments including different operating systems and GPU architectures. Finally, it must maintain perfect state consistency between the visual representation and the underlying data model.SYSTEM ARCHITECTURE OVERVIEWThe architecture follows clean architecture principles with clear separation of concerns. The system comprises six major subsystems that interact through well-defined interfaces.The Domain Model subsystem represents CRC cards as first-class entities with names, responsibilities, and collaborators. This layer has no dependencies on external frameworks or libraries, making it the most stable part of the system.The LLM Abstraction Layer provides a unified interface to interact with various language models, whether they run locally or remotely. This layer handles the complexity of different API formats, authentication mechanisms, and response parsing. It detects available GPU hardware and configures the appropriate acceleration backend.The Command Parser subsystem takes natural language input and LLM responses to extract structured commands. It identifies user intent such as creating new components, modifying existing ones, establishing relationships, or requesting snapshots.The Layout Engine computes positions for CRC cards and routes connection arrows to minimize visual clutter. It implements graph layout algorithms that consider component sizes, relationship directions, and aesthetic criteria.The Rendering Engine translates the positioned layout into visual representations. It supports multiple output formats including raster images (PNG, JPEG) and vector graphics (SVG). The rendering process applies styling rules to create visually appealing diagrams.The State Management subsystem maintains the current diagram state, handles persistence to JSON, and supports restoration from previously saved states. It implements the memento pattern to enable snapshot functionality.DOMAIN MODEL: CRC CARD REPRESENTATIONThe domain model centers on three core entities: CRCCard, Responsibility, and Collaboration. These entities capture the essential information needed to represent and manipulate CRC cards.A CRCCard contains a unique identifier, a component name, a collection of responsibilities, and a collection of collaborator references. The unique identifier ensures that cards can be reliably referenced even if their names change. The component name represents the class or module being designed. Responsibilities describe what the component does, typically expressed as verb phrases. Collaborators identify other components that this component depends upon or interacts with.Here is the core domain model implementation:class Responsibility: def __init__(self, description): # Each responsibility is a description of what the component does # Examples: "Manages user authentication", "Validates input data" self.description = description def __eq__(self, other): if not isinstance(other, Responsibility): return False return self.description == other.description def __hash__(self): return hash(self.description) def to_dict(self): return {"description": self.description} @staticmethod def from_dict(data): return Responsibility(data["description"])class Collaboration: def __init__(self, collaborator_id, collaborator_name): # collaborator_id: unique identifier of the component we collaborate with # collaborator_name: human-readable name for display purposes self.collaborator_id = collaborator_id self.collaborator_name = collaborator_name def __eq__(self, other): if not isinstance(other, Collaboration): return False return self.collaborator_id == other.collaborator_id def __hash__(self): return hash(self.collaborator_id) def to_dict(self): return { "collaborator_id": self.collaborator_id, "collaborator_name": self.collaborator_name } @staticmethod def from_dict(data): return Collaboration(data["collaborator_id"], data["collaborator_name"])class CRCCard: def __init__(self, card_id, name): # card_id: unique identifier for this card # name: the component/class name self.card_id = card_id self.name = name self.responsibilities = [] self.collaborations = [] def add_responsibility(self, responsibility): if responsibility not in self.responsibilities: self.responsibilities.append(responsibility) def remove_responsibility(self, responsibility): if responsibility in self.responsibilities: self.responsibilities.remove(responsibility) def add_collaboration(self, collaboration): if collaboration not in self.collaborations: self.collaborations.append(collaboration) def remove_collaboration(self, collaborator_id): self.collaborations = [c for c in self.collaborations if c.collaborator_id != collaborator_id] def to_dict(self): return { "card_id": self.card_id, "name": self.name, "responsibilities": [r.to_dict() for r in self.responsibilities], "collaborations": [c.to_dict() for c in self.collaborations] } @staticmethod def from_dict(data): card = CRCCard(data["card_id"], data["name"]) card.responsibilities = [Responsibility.from_dict(r) for r in data["responsibilities"]] card.collaborations = [Collaboration.from_dict(c) for c in data["collaborations"]] return cardThe domain model remains independent of presentation concerns. It does not know about positions, colors, or rendering. This separation allows the same domain model to be used with different visualization strategies or even non-visual representations.LLM ABSTRACTION LAYERThe LLM abstraction layer provides a unified interface to interact with language models across different platforms and providers. The key challenge is handling the diversity of APIs, authentication methods, and response formats while presenting a consistent interface to the rest of the application.The abstraction defines a base interface that all LLM providers must implement. This interface includes methods for initialization, text generation, and resource cleanup. Concrete implementations handle the specifics of each provider.For local models, the system must detect available GPU hardware and configure the appropriate acceleration backend. Modern deep learning frameworks support multiple GPU vendors through different backends. NVIDIA GPUs use CUDA, AMD GPUs use ROCm, Apple Silicon uses Metal Performance Shaders, and Intel GPUs use oneAPI or OpenCL.The hardware detection logic examines the system environment to determine available acceleration options:import platformimport subprocessimport osclass HardwareDetector: def __init__(self): self.system = platform.system() self.machine = platform.machine() def detect_gpu_backend(self): # Returns the best available GPU backend for the current system # Priority: CUDA > ROCm > MPS > OpenCL > CPU if self._has_cuda(): return "cuda" elif self._has_rocm(): return "rocm" elif self._has_mps(): return "mps" elif self._has_opencl(): return "opencl" else: return "cpu" def _has_cuda(self): # Check for NVIDIA CUDA support try: result = subprocess.run(['nvidia-smi'], capture_output=True, timeout=5) return result.returncode == 0 except (FileNotFoundError, subprocess.TimeoutExpired): return False def _has_rocm(self): # Check for AMD ROCm support if self.system != "Linux": return False try: result = subprocess.run(['rocm-smi'], capture_output=True, timeout=5) return result.returncode == 0 except (FileNotFoundError, subprocess.TimeoutExpired): return False def _has_mps(self): # Check for Apple Metal Performance Shaders if self.system != "Darwin": return False # MPS is available on macOS 12.3+ with Apple Silicon if self.machine == "arm64": try: import torch return torch.backends.mps.is_available() except ImportError: return False return False def _has_opencl(self): # Check for OpenCL support (Intel and others) try: import pyopencl platforms = pyopencl.get_platforms() return len(platforms) > 0 except ImportError: return FalseThe LLM provider interface defines the contract that all implementations must fulfill:from abc import ABC, abstractmethodclass LLMProvider(ABC): @abstractmethod def initialize(self, **kwargs): # Initialize the LLM with provider-specific configuration # kwargs may include: model_name, api_key, temperature, max_tokens, etc. pass @abstractmethod def generate(self, prompt, system_prompt=None): # Generate a response from the LLM given a user prompt # system_prompt: optional instructions for the model's behavior # Returns: generated text as a string pass @abstractmethod def cleanup(self): # Release resources and cleanup passFor local models using the transformers library with GPU acceleration, the implementation configures the appropriate device and data types:class LocalTransformersProvider(LLMProvider): def __init__(self): self.model = None self.tokenizer = None self.device = None self.backend = None def initialize(self, model_name="mistralai/Mistral-7B-Instruct-v0.2", temperature=0.7, max_tokens=2048, **kwargs): import torch from transformers import AutoModelForCausalLM, AutoTokenizer # Detect the best available hardware backend detector = HardwareDetector() self.backend = detector.detect_gpu_backend() # Configure device and dtype based on backend if self.backend == "cuda": self.device = torch.device("cuda") dtype = torch.float16 # Use half precision for efficiency elif self.backend == "rocm": self.device = torch.device("cuda") # ROCm uses cuda device API dtype = torch.float16 elif self.backend == "mps": self.device = torch.device("mps") dtype = torch.float32 # MPS works better with float32 else: self.device = torch.device("cpu") dtype = torch.float32 # Load tokenizer self.tokenizer = AutoTokenizer.from_pretrained(model_name) # Load model with appropriate configuration self.model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=dtype, device_map="auto" if self.backend != "cpu" else None, low_cpu_mem_usage=True ) if self.backend == "cpu": self.model.to(self.device) self.temperature = temperature self.max_tokens = max_tokens def generate(self, prompt, system_prompt=None): import torch # Construct the full prompt with system instructions if provided if system_prompt: full_prompt = f"{system_prompt}\n\nUser: {prompt}\n\nAssistant:" else: full_prompt = prompt # Tokenize input inputs = self.tokenizer(full_prompt, return_tensors="pt") inputs = {k: v.to(self.device) for k, v in inputs.items()} # Generate response with torch.no_grad(): outputs = self.model.generate( **inputs, max_new_tokens=self.max_tokens, temperature=self.temperature, do_sample=True, pad_token_id=self.tokenizer.eos_token_id ) # Decode and return the generated text generated_text = self.tokenizer.decode(outputs[0], skip_special_tokens=True) # Extract only the new generated portion response = generated_text[len(full_prompt):].strip() return response def cleanup(self): import torch if self.model is not None: del self.model if self.tokenizer is not None: del self.tokenizer if torch.cuda.is_available(): torch.cuda.empty_cache()For remote API providers like OpenAI, the implementation handles HTTP requests and API-specific authentication:class OpenAIProvider(LLMProvider): def __init__(self): self.api_key = None self.model_name = None self.temperature = None self.max_tokens = None def initialize(self, api_key=None, model_name="gpt-4", temperature=0.7, max_tokens=2048, **kwargs): import os # Get API key from parameter or environment variable self.api_key = api_key or os.getenv("OPENAI_API_KEY") if not self.api_key: raise ValueError("OpenAI API key must be provided") self.model_name = model_name self.temperature = temperature self.max_tokens = max_tokens def generate(self, prompt, system_prompt=None): import requests # Construct messages for chat completion API messages = [] if system_prompt: messages.append({"role": "system", "content": system_prompt}) messages.append({"role": "user", "content": prompt}) # Make API request headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } data = { "model": self.model_name, "messages": messages, "temperature": self.temperature, "max_tokens": self.max_tokens } response = requests.post( "https://api.openai.com/v1/chat/completions", headers=headers, json=data, timeout=60 ) response.raise_for_status() result = response.json() return result["choices"][0]["message"]["content"] def cleanup(self): # No cleanup needed for API-based provider passThe LLM manager provides a factory pattern to create the appropriate provider based on configuration:class LLMManager: def __init__(self): self.provider = None def setup(self, provider_type="local", **kwargs): # Create and initialize the appropriate LLM provider # provider_type: "local", "openai", "anthropic", etc. if provider_type == "local": self.provider = LocalTransformersProvider() elif provider_type == "openai": self.provider = OpenAIProvider() else: raise ValueError(f"Unknown provider type: {provider_type}") self.provider.initialize(**kwargs) def query(self, prompt, system_prompt=None): if self.provider is None: raise RuntimeError("LLM provider not initialized") return self.provider.generate(prompt, system_prompt) def shutdown(self): if self.provider is not None: self.provider.cleanup()This abstraction layer allows the application to switch between different LLM backends without changing the higher-level logic. The hardware detection ensures optimal performance on whatever GPU architecture is available.COMMAND PARSER AND INTENT RECOGNITIONThe command parser interprets natural language input to extract structured commands that the system can execute. This component bridges the gap between free-form user descriptions and the precise operations needed to manipulate the diagram.The parser uses the LLM to analyze user input and identify intent. It provides the LLM with a system prompt that defines the expected output format and available command types. The LLM response is then parsed into structured command objects.The system supports several command types. The CREATE_COMPONENT command adds a new CRC card with specified name, responsibilities, and collaborators. The MODIFY_COMPONENT command updates an existing card by adding or removing responsibilities or collaborators. The DELETE_COMPONENT command removes a card from the diagram. The SNAPSHOT command saves the current state to files. The RESTORE command loads a previously saved diagram from a JSON file.The command parser implementation uses a structured prompt to guide the LLM:import jsonclass CommandParser: def __init__(self, llm_manager): self.llm = llm_manager self.system_prompt = self._build_system_prompt() def _build_system_prompt(self): # Define the system prompt that instructs the LLM how to parse commands return """You are a command parser for a CRC card diagramming tool.Your task is to analyze user input and extract structured commands.Available command types:1. CREATE_COMPONENT: Create a new CRC card2. MODIFY_COMPONENT: Modify an existing CRC card3. DELETE_COMPONENT: Remove a CRC card4. SNAPSHOT: Save the current diagram5. RESTORE: Load a saved diagram6. FINISH: User indicates the diagram is completeFor each user input, respond with a JSON object containing:{ "command_type": "CREATE_COMPONENT|MODIFY_COMPONENT|DELETE_COMPONENT|SNAPSHOT|RESTORE|FINISH", "parameters": { // command-specific parameters }}CREATE_COMPONENT parameters:- name: component name (string)- responsibilities: list of responsibility descriptions (list of strings)- collaborators: list of collaborator names (list of strings)MODIFY_COMPONENT parameters:- name: component name to modify (string)- add_responsibilities: responsibilities to add (list of strings, optional)- remove_responsibilities: responsibilities to remove (list of strings, optional)- add_collaborators: collaborators to add (list of strings, optional)- remove_collaborators: collaborators to remove (list of strings, optional)DELETE_COMPONENT parameters:- name: component name to delete (string)SNAPSHOT parameters:- filename: base filename for saved files (string, optional)RESTORE parameters:- filename: JSON file to restore from (string)FINISH parameters: noneRespond ONLY with the JSON object, no additional text.""" def parse(self, user_input): # Parse user input into a structured command # Returns a Command object or None if parsing fails try: # Query the LLM with the user input response = self.llm.query(user_input, self.system_prompt) # Extract JSON from response (handle cases where LLM adds extra text) json_str = self._extract_json(response) # Parse JSON command_data = json.loads(json_str) # Create and return appropriate Command object return Command.from_dict(command_data) except Exception as e: print(f"Error parsing command: {e}") return None def _extract_json(self, text): # Extract JSON object from text that may contain additional content # Look for content between first { and last } start = text.find('{') end = text.rfind('}') if start == -1 or end == -1: raise ValueError("No JSON object found in response") return text[start:end+1]class Command: def __init__(self, command_type, parameters): self.command_type = command_type self.parameters = parameters @staticmethod def from_dict(data): return Command(data["command_type"], data.get("parameters", {})) def to_dict(self): return { "command_type": self.command_type, "parameters": self.parameters }The command parser leverages the LLM's natural language understanding capabilities to handle variations in how users express their intent. For example, the user might say "Add a new component called UserManager" or "Create a UserManager class" and both would be correctly interpreted as CREATE_COMPONENT commands.LAYOUT ENGINEThe layout engine computes positions for CRC cards and routes connection arrows to create aesthetically pleasing diagrams. This is a challenging problem because it must balance multiple competing objectives: minimize arrow crossings, distribute cards evenly, respect the directed graph structure, and avoid overlaps.The layout algorithm uses a force-directed approach combined with hierarchical constraints. Force-directed layouts treat the diagram as a physical system where cards repel each other and connections act as springs. This produces natural-looking layouts but can result in unstable positions. Adding hierarchical constraints based on the dependency graph creates more predictable layouts where components appear in a logical order.The layout process proceeds in several phases. First, the algorithm analyzes the dependency graph to identify hierarchical levels. Components with no dependencies go in the top level. Components that depend only on top-level components go in the second level, and so on. This creates a natural flow from independent components to dependent ones.Second, the algorithm assigns initial positions based on the hierarchical levels. Cards in the same level are distributed horizontally with equal spacing. Levels are stacked vertically with sufficient space for arrows.Third, the algorithm applies force-directed refinement to improve the layout. Repulsive forces push overlapping cards apart. Attractive forces along connections pull related cards closer. The forces are applied iteratively until the system reaches equilibrium or a maximum iteration count.Fourth, the algorithm routes connection arrows using orthogonal edge routing or Bezier curves. The routing algorithm attempts to minimize crossings by analyzing the positions of source and target cards and choosing appropriate control points.Here is the layout engine implementation:import mathclass LayoutEngine: def __init__(self, card_width=200, card_height=150, horizontal_spacing=100, vertical_spacing=150): self.card_width = card_width self.card_height = card_height self.horizontal_spacing = horizontal_spacing self.vertical_spacing = vertical_spacing def compute_layout(self, cards): # Compute positions for all cards and routing for all connections # Returns a dictionary mapping card_id to Position objects # and a list of Connection objects with routing information if not cards: return {}, [] # Build dependency graph graph = self._build_dependency_graph(cards) # Compute hierarchical levels levels = self._compute_hierarchical_levels(graph, cards) # Assign initial positions based on levels positions = self._assign_initial_positions(levels, cards) # Apply force-directed refinement positions = self._apply_force_directed_refinement(positions, graph, cards) # Route connections connections = self._route_connections(positions, cards) return positions, connections def _build_dependency_graph(self, cards): # Build a dictionary mapping card_id to list of collaborator card_ids graph = {} card_id_map = {card.name: card.card_id for card in cards} for card in cards: collaborator_ids = [] for collab in card.collaborations: # Try to find the collaborator by ID first, then by name if collab.collaborator_id in [c.card_id for c in cards]: collaborator_ids.append(collab.collaborator_id) elif collab.collaborator_name in card_id_map: collaborator_ids.append(card_id_map[collab.collaborator_name]) graph[card.card_id] = collaborator_ids return graph def _compute_hierarchical_levels(self, graph, cards): # Compute hierarchical levels using topological sorting # Returns a list of lists, where each inner list contains card_ids at that level # Calculate in-degree for each node in_degree = {card.card_id: 0 for card in cards} for card_id, collaborators in graph.items(): for collab_id in collaborators: if collab_id in in_degree: in_degree[collab_id] += 1 # Process nodes level by level levels = [] remaining = set(card.card_id for card in cards) while remaining: # Find all nodes with in-degree 0 in remaining set current_level = [card_id for card_id in remaining if in_degree[card_id] == 0] if not current_level: # Cycle detected or isolated components # Add all remaining nodes to final level current_level = list(remaining) levels.append(current_level) # Remove current level from remaining for card_id in current_level: remaining.remove(card_id) # Decrease in-degree for nodes that depend on current level for card_id in current_level: for collab_id in graph.get(card_id, []): if collab_id in in_degree: in_degree[collab_id] -= 1 return levels def _assign_initial_positions(self, levels, cards): # Assign initial positions based on hierarchical levels positions = {} for level_index, level in enumerate(levels): num_cards = len(level) level_width = (num_cards * self.card_width + (num_cards - 1) * self.horizontal_spacing) # Center the level horizontally start_x = -level_width / 2 y = level_index * (self.card_height + self.vertical_spacing) for card_index, card_id in enumerate(level): x = start_x + card_index * (self.card_width + self.horizontal_spacing) positions[card_id] = Position(x, y) return positions def _apply_force_directed_refinement(self, positions, graph, cards, iterations=50): # Apply force-directed algorithm to refine positions # Uses repulsive forces between all cards and attractive forces along edges repulsion_strength = 5000 attraction_strength = 0.1 damping = 0.9 velocities = {card_id: Position(0, 0) for card_id in positions} for iteration in range(iterations): forces = {card_id: Position(0, 0) for card_id in positions} # Calculate repulsive forces between all pairs card_ids = list(positions.keys()) for i, card_id1 in enumerate(card_ids): for card_id2 in card_ids[i+1:]: pos1 = positions[card_id1] pos2 = positions[card_id2] dx = pos1.x - pos2.x dy = pos1.y - pos2.y distance = math.sqrt(dx*dx + dy*dy) if distance < 1: distance = 1 # Repulsive force inversely proportional to distance force_magnitude = repulsion_strength / (distance * distance) fx = (dx / distance) * force_magnitude fy = (dy / distance) * force_magnitude forces[card_id1].x += fx forces[card_id1].y += fy forces[card_id2].x -= fx forces[card_id2].y -= fy # Calculate attractive forces along edges for card_id, collaborators in graph.items(): if card_id not in positions: continue pos1 = positions[card_id] for collab_id in collaborators: if collab_id not in positions: continue pos2 = positions[collab_id] dx = pos2.x - pos1.x dy = pos2.y - pos1.y distance = math.sqrt(dx*dx + dy*dy) if distance < 1: continue # Attractive force proportional to distance force_magnitude = attraction_strength * distance fx = (dx / distance) * force_magnitude fy = (dy / distance) * force_magnitude forces[card_id].x += fx forces[card_id].y += fy forces[collab_id].x -= fx forces[collab_id].y -= fy # Update velocities and positions for card_id in positions: velocities[card_id].x = (velocities[card_id].x + forces[card_id].x) * damping velocities[card_id].y = (velocities[card_id].y + forces[card_id].y) * damping positions[card_id].x += velocities[card_id].x positions[card_id].y += velocities[card_id].y return positions def _route_connections(self, positions, cards): # Route connection arrows between cards # Returns a list of Connection objects connections = [] for card in cards: source_pos = positions.get(card.card_id) if not source_pos: continue for collab in card.collaborations: # Find target position target_pos = None for other_card in cards: if (other_card.card_id == collab.collaborator_id or other_card.name == collab.collaborator_name): target_pos = positions.get(other_card.card_id) break if not target_pos: continue # Calculate connection points on card boundaries source_point = self._calculate_connection_point( source_pos, target_pos, self.card_width, self.card_height) target_point = self._calculate_connection_point( target_pos, source_pos, self.card_width, self.card_height) # Create connection with simple straight line # More sophisticated routing could use Bezier curves or orthogonal routing connection = Connection( source_id=card.card_id, target_id=collab.collaborator_id, source_point=source_point, target_point=target_point, control_points=[] ) connections.append(connection) return connections def _calculate_connection_point(self, from_pos, to_pos, width, height): # Calculate the point on the card boundary where the connection should attach # Uses the direction from from_pos to to_pos to determine which edge dx = to_pos.x - from_pos.x dy = to_pos.y - from_pos.y # Calculate angle angle = math.atan2(dy, dx) # Determine which edge based on angle # Card boundaries: left edge at x, right edge at x+width # top edge at y, bottom edge at y+height half_width = width / 2 half_height = height / 2 # Calculate intersection with card boundary if abs(dx) > abs(dy): # More horizontal than vertical if dx > 0: # Right edge return Position(from_pos.x + half_width, from_pos.y + half_height * (dy / abs(dx))) else: # Left edge return Position(from_pos.x - half_width, from_pos.y + half_height * (dy / abs(dx))) else: # More vertical than horizontal if dy > 0: # Bottom edge return Position(from_pos.x + half_width * (dx / abs(dy)), from_pos.y + half_height) else: # Top edge return Position(from_pos.x + half_width * (dx / abs(dy)), from_pos.y - half_height)class Position: def __init__(self, x, y): self.x = x self.y = y def to_dict(self): return {"x": self.x, "y": self.y} @staticmethod def from_dict(data): return Position(data["x"], data["y"])class Connection: def __init__(self, source_id, target_id, source_point, target_point, control_points): self.source_id = source_id self.target_id = target_id self.source_point = source_point self.target_point = target_point self.control_points = control_points # For Bezier curves def to_dict(self): return { "source_id": self.source_id, "target_id": self.target_id, "source_point": self.source_point.to_dict(), "target_point": self.target_point.to_dict(), "control_points": [p.to_dict() for p in self.control_points] } @staticmethod def from_dict(data): return Connection( data["source_id"], data["target_id"], Position.from_dict(data["source_point"]), Position.from_dict(data["target_point"]), [Position.from_dict(p) for p in data["control_points"]] )The layout engine produces positions and connection routing that can be rendered by the rendering engine. The force-directed approach ensures that the layout adapts naturally to different diagram structures while the hierarchical constraints maintain readability.RENDERING ENGINEThe rendering engine transforms the positioned layout into visual representations. It supports multiple output formats to serve different use cases. PNG and JPEG formats provide raster images suitable for embedding in documents or presentations. SVG format provides vector graphics that scale without quality loss and can be further edited in vector graphics tools.The rendering process involves several steps. First, the engine determines the bounding box of the diagram by finding the minimum and maximum coordinates of all cards. Second, it creates a canvas with appropriate dimensions and margins. Third, it renders each CRC card as a rectangle with the component name, responsibilities, and collaborators. Fourth, it renders connection arrows between cards. Fifth, it applies styling to create an aesthetically pleasing result.The rendering engine uses different libraries depending on the output format. For raster images, it uses the Python Imaging Library (Pillow). For SVG, it generates XML directly or uses a library like svgwrite.Here is the rendering engine implementation for both formats:from PIL import Image, ImageDraw, ImageFontimport svgwriteclass RenderingEngine: def __init__(self, card_width=200, card_height=150): self.card_width = card_width self.card_height = card_height self.margin = 50 # Styling configuration self.card_fill_color = (255, 255, 240) # Light yellow self.card_border_color = (0, 0, 0) # Black self.card_border_width = 2 self.text_color = (0, 0, 0) # Black self.arrow_color = (50, 50, 200) # Blue self.arrow_width = 2 def render_to_png(self, cards, positions, connections, output_path): # Render the diagram to a PNG file # Calculate canvas dimensions bounds = self._calculate_bounds(positions) canvas_width = int(bounds["max_x"] - bounds["min_x"] + 2 * self.margin) canvas_height = int(bounds["max_y"] - bounds["min_y"] + 2 * self.margin) # Create image image = Image.new('RGB', (canvas_width, canvas_height), color=(255, 255, 255)) draw = ImageDraw.Draw(image) # Offset to shift all coordinates into positive space offset_x = -bounds["min_x"] + self.margin offset_y = -bounds["min_y"] + self.margin # Render connections first (so they appear behind cards) self._render_connections_png(draw, connections, positions, offset_x, offset_y) # Render cards self._render_cards_png(draw, cards, positions, offset_x, offset_y) # Save image image.save(output_path, 'PNG') def render_to_svg(self, cards, positions, connections, output_path): # Render the diagram to an SVG file # Calculate canvas dimensions bounds = self._calculate_bounds(positions) canvas_width = int(bounds["max_x"] - bounds["min_x"] + 2 * self.margin) canvas_height = int(bounds["max_y"] - bounds["min_y"] + 2 * self.margin) # Create SVG drawing dwg = svgwrite.Drawing(output_path, size=(f"{canvas_width}px", f"{canvas_height}px")) # Offset to shift all coordinates into positive space offset_x = -bounds["min_x"] + self.margin offset_y = -bounds["min_y"] + self.margin # Render connections first self._render_connections_svg(dwg, connections, positions, offset_x, offset_y) # Render cards self._render_cards_svg(dwg, cards, positions, offset_x, offset_y) # Save SVG dwg.save() def _calculate_bounds(self, positions): # Calculate the bounding box of all cards if not positions: return {"min_x": 0, "max_x": 0, "min_y": 0, "max_y": 0} min_x = min(pos.x for pos in positions.values()) max_x = max(pos.x + self.card_width for pos in positions.values()) min_y = min(pos.y for pos in positions.values()) max_y = max(pos.y + self.card_height for pos in positions.values()) return {"min_x": min_x, "max_x": max_x, "min_y": min_y, "max_y": max_y} def _render_cards_png(self, draw, cards, positions, offset_x, offset_y): # Render all CRC cards to PNG try: # Try to load a nice font font_title = ImageFont.truetype("arial.ttf", 14) font_text = ImageFont.truetype("arial.ttf", 10) except: # Fall back to default font font_title = ImageFont.load_default() font_text = ImageFont.load_default() for card in cards: pos = positions.get(card.card_id) if not pos: continue # Calculate card rectangle x = pos.x + offset_x y = pos.y + offset_y # Draw card background draw.rectangle( [(x, y), (x + self.card_width, y + self.card_height)], fill=self.card_fill_color, outline=self.card_border_color, width=self.card_border_width ) # Draw component name at top name_y = y + 10 draw.text((x + 10, name_y), card.name, fill=self.text_color, font=font_title) # Draw horizontal line below name line_y = name_y + 20 draw.line([(x + 5, line_y), (x + self.card_width - 5, line_y)], fill=self.card_border_color, width=1) # Draw responsibilities section resp_y = line_y + 10 draw.text((x + 10, resp_y), "Responsibilities:", fill=self.text_color, font=font_text) current_y = resp_y + 15 for resp in card.responsibilities[:3]: # Limit to 3 for space # Truncate long responsibilities text = resp.description[:25] + "..." if len(resp.description) > 25 else resp.description draw.text((x + 15, current_y), f"- {text}", fill=self.text_color, font=font_text) current_y += 12 # Draw collaborators section collab_y = y + self.card_height - 50 draw.text((x + 10, collab_y), "Collaborators:", fill=self.text_color, font=font_text) current_y = collab_y + 15 for collab in card.collaborations[:2]: # Limit to 2 for space text = collab.collaborator_name[:20] + "..." if len(collab.collaborator_name) > 20 else collab.collaborator_name draw.text((x + 15, current_y), f"- {text}", fill=self.text_color, font=font_text) current_y += 12 def _render_connections_png(self, draw, connections, positions, offset_x, offset_y): # Render all connection arrows to PNG for conn in connections: source_x = conn.source_point.x + offset_x source_y = conn.source_point.y + offset_y target_x = conn.target_point.x + offset_x target_y = conn.target_point.y + offset_y # Draw arrow line draw.line([(source_x, source_y), (target_x, target_y)], fill=self.arrow_color, width=self.arrow_width) # Draw arrowhead self._draw_arrowhead_png(draw, source_x, source_y, target_x, target_y) def _draw_arrowhead_png(self, draw, x1, y1, x2, y2): # Draw an arrowhead at the end of a line arrow_size = 10 # Calculate angle of the line dx = x2 - x1 dy = y2 - y1 length = math.sqrt(dx*dx + dy*dy) if length < 1: return # Normalize direction dx /= length dy /= length # Calculate arrowhead points # Two points forming a triangle with the target point perp_x = -dy perp_y = dx point1_x = x2 - arrow_size * dx + arrow_size * 0.5 * perp_x point1_y = y2 - arrow_size * dy + arrow_size * 0.5 * perp_y point2_x = x2 - arrow_size * dx - arrow_size * 0.5 * perp_x point2_y = y2 - arrow_size * dy - arrow_size * 0.5 * perp_y # Draw filled triangle draw.polygon([(x2, y2), (point1_x, point1_y), (point2_x, point2_y)], fill=self.arrow_color) def _render_cards_svg(self, dwg, cards, positions, offset_x, offset_y): # Render all CRC cards to SVG for card in cards: pos = positions.get(card.card_id) if not pos: continue # Calculate card rectangle x = pos.x + offset_x y = pos.y + offset_y # Create group for this card card_group = dwg.g() # Draw card background card_group.add(dwg.rect( insert=(x, y), size=(self.card_width, self.card_height), fill=f"rgb{self.card_fill_color}", stroke=f"rgb{self.card_border_color}", stroke_width=self.card_border_width )) # Draw component name name_y = y + 20 card_group.add(dwg.text( card.name, insert=(x + 10, name_y), fill=f"rgb{self.text_color}", font_size="14px", font_weight="bold" )) # Draw horizontal line line_y = name_y + 10 card_group.add(dwg.line( start=(x + 5, line_y), end=(x + self.card_width - 5, line_y), stroke=f"rgb{self.card_border_color}", stroke_width=1 )) # Draw responsibilities resp_y = line_y + 15 card_group.add(dwg.text( "Responsibilities:", insert=(x + 10, resp_y), fill=f"rgb{self.text_color}", font_size="10px" )) current_y = resp_y + 15 for resp in card.responsibilities[:3]: text = resp.description[:25] + "..." if len(resp.description) > 25 else resp.description card_group.add(dwg.text( f"- {text}", insert=(x + 15, current_y), fill=f"rgb{self.text_color}", font_size="9px" )) current_y += 12 # Draw collaborators collab_y = y + self.card_height - 40 card_group.add(dwg.text( "Collaborators:", insert=(x + 10, collab_y), fill=f"rgb{self.text_color}", font_size="10px" )) current_y = collab_y + 15 for collab in card.collaborations[:2]: text = collab.collaborator_name[:20] + "..." if len(collab.collaborator_name) > 20 else collab.collaborator_name card_group.add(dwg.text( f"- {text}", insert=(x + 15, current_y), fill=f"rgb{self.text_color}", font_size="9px" )) current_y += 12 dwg.add(card_group) def _render_connections_svg(self, dwg, connections, positions, offset_x, offset_y): # Render all connection arrows to SVG for conn in connections: source_x = conn.source_point.x + offset_x source_y = conn.source_point.y + offset_y target_x = conn.target_point.x + offset_x target_y = conn.target_point.y + offset_y # Draw arrow line dwg.add(dwg.line( start=(source_x, source_y), end=(target_x, target_y), stroke=f"rgb{self.arrow_color}", stroke_width=self.arrow_width )) # Draw arrowhead self._draw_arrowhead_svg(dwg, source_x, source_y, target_x, target_y) def _draw_arrowhead_svg(self, dwg, x1, y1, x2, y2): # Draw an arrowhead at the end of a line in SVG arrow_size = 10 dx = x2 - x1 dy = y2 - y1 length = math.sqrt(dx*dx + dy*dy) if length < 1: return dx /= length dy /= length perp_x = -dy perp_y = dx point1_x = x2 - arrow_size * dx + arrow_size * 0.5 * perp_x point1_y = y2 - arrow_size * dy + arrow_size * 0.5 * perp_y point2_x = x2 - arrow_size * dx - arrow_size * 0.5 * perp_x point2_y = y2 - arrow_size * dy - arrow_size * 0.5 * perp_y dwg.add(dwg.polygon( points=[(x2, y2), (point1_x, point1_y), (point2_x, point2_y)], fill=f"rgb{self.arrow_color}" ))The rendering engine produces professional-looking diagrams that clearly communicate the CRC card structure. The use of both raster and vector formats ensures compatibility with different downstream tools and workflows.STATE MANAGEMENT AND PERSISTENCThe state management subsystem maintains the current diagram state and handles persistence. It implements the memento pattern to capture snapshots of the diagram that can be saved to disk and later restored.The diagram state includes all CRC cards, their positions, and connection routing information. When the user requests a snapshot, the system serializes this state to JSON and renders the current visual representation to an image file.The JSON format captures all information needed to perfectly reconstruct the diagram. It includes the card definitions with all responsibilities and collaborations, the computed positions, and metadata such as creation timestamp and version information.Here is the state management implementation:import jsonimport osfrom datetime import datetimeimport uuidclass DiagramState: def __init__(self): self.cards = [] self.positions = {} self.connections = [] self.metadata = { "version": "1.0", "created": datetime.now().isoformat(), "modified": datetime.now().isoformat() } def add_card(self, card): # Add a new CRC card to the diagram if card.card_id not in [c.card_id for c in self.cards]: self.cards.append(card) self._update_modified() def remove_card(self, card_id): # Remove a card and all connections involving it self.cards = [c for c in self.cards if c.card_id != card_id] # Remove connections self.connections = [conn for conn in self.connections if conn.source_id != card_id and conn.target_id != card_id] # Remove position if card_id in self.positions: del self.positions[card_id] self._update_modified() def get_card_by_name(self, name): # Find a card by its name for card in self.cards: if card.name == name: return card return None def get_card_by_id(self, card_id): # Find a card by its ID for card in self.cards: if card.card_id == card_id: return card return None def update_layout(self, positions, connections): # Update the layout information self.positions = positions self.connections = connections self._update_modified() def _update_modified(self): self.metadata["modified"] = datetime.now().isoformat() def to_dict(self): # Serialize the entire state to a dictionary return { "metadata": self.metadata, "cards": [card.to_dict() for card in self.cards], "positions": {card_id: pos.to_dict() for card_id, pos in self.positions.items()}, "connections": [conn.to_dict() for conn in self.connections] } def to_json(self): # Serialize to JSON string return json.dumps(self.to_dict(), indent=2) @staticmethod def from_dict(data): # Deserialize from dictionary state = DiagramState() state.metadata = data.get("metadata", state.metadata) state.cards = [CRCCard.from_dict(c) for c in data.get("cards", [])] state.positions = {card_id: Position.from_dict(pos) for card_id, pos in data.get("positions", {}).items()} state.connections = [Connection.from_dict(c) for c in data.get("connections", [])] return state @staticmethod def from_json(json_str): # Deserialize from JSON string data = json.loads(json_str) return DiagramState.from_dict(data)class StateManager: def __init__(self, layout_engine, rendering_engine): self.state = DiagramState() self.layout_engine = layout_engine self.rendering_engine = rendering_engine def execute_command(self, command): # Execute a command and update the state if command.command_type == "CREATE_COMPONENT": self._execute_create(command.parameters) elif command.command_type == "MODIFY_COMPONENT": self._execute_modify(command.parameters) elif command.command_type == "DELETE_COMPONENT": self._execute_delete(command.parameters) elif command.command_type == "SNAPSHOT": self._execute_snapshot(command.parameters) elif command.command_type == "RESTORE": self._execute_restore(command.parameters) elif command.command_type == "FINISH": return "FINISH" # Recompute layout after any modification if command.command_type in ["CREATE_COMPONENT", "MODIFY_COMPONENT", "DELETE_COMPONENT"]: self._recompute_layout() return "SUCCESS" def _execute_create(self, params): # Create a new CRC card name = params.get("name") responsibilities = params.get("responsibilities", []) collaborators = params.get("collaborators", []) # Generate unique ID card_id = str(uuid.uuid4()) # Create card card = CRCCard(card_id, name) # Add responsibilities for resp_desc in responsibilities: card.add_responsibility(Responsibility(resp_desc)) # Add collaborations (resolve collaborator IDs) for collab_name in collaborators: # Try to find existing card with this name existing_card = self.state.get_card_by_name(collab_name) if existing_card: collab_id = existing_card.card_id else: # Create placeholder ID for not-yet-created collaborator collab_id = f"placeholder_{collab_name}" card.add_collaboration(Collaboration(collab_id, collab_name)) self.state.add_card(card) def _execute_modify(self, params): # Modify an existing CRC card name = params.get("name") card = self.state.get_card_by_name(name) if not card: print(f"Card '{name}' not found") return # Add responsibilities for resp_desc in params.get("add_responsibilities", []): card.add_responsibility(Responsibility(resp_desc)) # Remove responsibilities for resp_desc in params.get("remove_responsibilities", []): card.remove_responsibility(Responsibility(resp_desc)) # Add collaborations for collab_name in params.get("add_collaborators", []): existing_card = self.state.get_card_by_name(collab_name) if existing_card: collab_id = existing_card.card_id else: collab_id = f"placeholder_{collab_name}" card.add_collaboration(Collaboration(collab_id, collab_name)) # Remove collaborations for collab_name in params.get("remove_collaborators", []): existing_card = self.state.get_card_by_name(collab_name) if existing_card: card.remove_collaboration(existing_card.card_id) def _execute_delete(self, params): # Delete a CRC card name = params.get("name") card = self.state.get_card_by_name(name) if card: self.state.remove_card(card.card_id) else: print(f"Card '{name}' not found") def _execute_snapshot(self, params): # Save the current state to files filename = params.get("filename", f"diagram_{datetime.now().strftime('%Y%m%d_%H%M%S')}") # Ensure filename has no extension base_filename = os.path.splitext(filename)[0] # Save JSON json_path = f"{base_filename}.json" with open(json_path, 'w') as f: f.write(self.state.to_json()) print(f"Saved state to {json_path}") # Save PNG png_path = f"{base_filename}.png" self.rendering_engine.render_to_png( self.state.cards, self.state.positions, self.state.connections, png_path ) print(f"Saved diagram to {png_path}") # Save SVG svg_path = f"{base_filename}.svg" self.rendering_engine.render_to_svg( self.state.cards, self.state.positions, self.state.connections, svg_path ) print(f"Saved diagram to {svg_path}") def _execute_restore(self, params): # Restore state from a JSON file filename = params.get("filename") if not os.path.exists(filename): print(f"File '{filename}' not found") return with open(filename, 'r') as f: json_str = f.read() self.state = DiagramState.from_json(json_str) print(f"Restored state from {filename}") def _recompute_layout(self): # Recompute the layout for the current cards positions, connections = self.layout_engine.compute_layout(self.state.cards) self.state.update_layout(positions, connections) def get_current_state(self): return self.stateThe state manager acts as the controller in the MVC pattern, coordinating between the domain model, layout engine, and rendering engine. It ensures that the state remains consistent and that all changes trigger appropriate layout recomputation.GPU AND HARDWARE ACCELERATION SUPPORTSupporting multiple GPU architectures requires careful handling of hardware-specific libraries and runtime configuration. The system must detect available hardware, select appropriate backends, and configure libraries accordingly.Modern deep learning frameworks provide abstraction layers that simplify multi-GPU support. PyTorch supports CUDA for NVIDIA GPUs, ROCm for AMD GPUs, and MPS for Apple Silicon. TensorFlow has similar multi-backend support. The key is to detect the available hardware at runtime and configure the framework appropriately.The hardware detection logic examines system properties and attempts to import hardware-specific libraries. For NVIDIA CUDA, it checks for the nvidia-smi utility and the CUDA runtime. For AMD ROCm, it checks for rocm-smi and the ROCm runtime. For Apple MPS, it checks the operating system and processor architecture. For Intel GPUs, it checks for OpenCL or oneAPI support.Once the hardware is detected, the system configures the deep learning framework to use the appropriate backend. This involves setting device types, data types, and memory management strategies optimized for each hardware platform.The implementation shown earlier in the LLM abstraction layer demonstrates this approach. The HardwareDetector class encapsulates all hardware detection logic, and the LocalTransformersProvider uses this information to configure PyTorch appropriately.For production deployments, additional considerations include handling systems with multiple GPUs, managing memory constraints, and providing fallback options when GPU acceleration is unavailable. The system should gracefully degrade to CPU execution if no GPU is available, though performance will be significantly slower.CROSS-PLATFORM CONSIDERATIONSBuilding a truly cross-platform application requires attention to operating system differences in file paths, process management, font availability, and system libraries.File path handling must use os.path or pathlib to ensure correct path separators on Windows, macOS, and Linux. The application should never hard-code forward slashes or backslashes in paths.Font availability varies across platforms. Windows typically has Arial, Calibri, and Times New Roman. macOS has Helvetica, Arial, and San Francisco. Linux distributions vary widely in available fonts. The rendering engine should attempt to load preferred fonts but fall back gracefully to default fonts when specific fonts are unavailable.Process execution for hardware detection must handle platform-specific utilities. The nvidia-smi utility is available on all platforms with NVIDIA drivers, but rocm-smi is Linux-only. The code must handle FileNotFoundError exceptions when utilities are not available.Library dependencies should be specified in a requirements.txt file with version constraints to ensure consistent behavior across platforms. The application should document any platform-specific installation steps, such as installing CUDA or ROCm drivers.The Python standard library provides excellent cross-platform support for most operations. Using standard library functions like os.path.join, platform.system, and subprocess.run ensures that the code works correctly on all major operating systems.IMPLEMENTATION DETAILS AND INTEGRATIONIntegrating all the components into a cohesive application requires a main controller that orchestrates the interaction between subsystems. The main application loop accepts user input, parses commands, executes them, and provides feedback.The application maintains a session state that tracks the current diagram, the LLM manager, and all engine instances. It provides a command-line interface where users can enter natural language descriptions of their CRC cards.Here is a simplified integration showing how the components work together:class CRCDiagramTool: def __init__(self, llm_provider_type="local", **llm_kwargs): # Initialize all subsystems self.llm_manager = LLMManager() self.llm_manager.setup(llm_provider_type, **llm_kwargs) self.command_parser = CommandParser(self.llm_manager) self.layout_engine = LayoutEngine() self.rendering_engine = RenderingEngine() self.state_manager = StateManager(self.layout_engine, self.rendering_engine) def run(self): # Main application loop print("CRC Diagram Tool") print("Enter natural language descriptions of your components.") print("Type 'snapshot' to save the current diagram.") print("Type 'finish' when done.") print() while True: # Get user input user_input = input("You: ").strip() if not user_input: continue # Parse command command = self.command_parser.parse(user_input) if command is None: print("Sorry, I didn't understand that. Please try again.") continue # Execute command result = self.state_manager.execute_command(command) if result == "FINISH": print("Diagram complete. Goodbye!") break elif result == "SUCCESS": print("Command executed successfully.") # Show current diagram summary self._show_summary() # Cleanup self.llm_manager.shutdown() def _show_summary(self): # Display a summary of the current diagram state = self.state_manager.get_current_state() print(f"\nCurrent diagram has {len(state.cards)} components:") for card in state.cards: print(f" - {card.name} ({len(card.responsibilities)} responsibilities, " f"{len(card.collaborations)} collaborations)") print()This integration provides a clean separation between the user interface layer and the business logic. The CRCDiagramTool class acts as a facade that hides the complexity of the underlying subsystems.TESTING STRATEGIESComprehensive testing ensures that the system works correctly across different scenarios and edge cases. The testing strategy should include unit tests for individual components, integration tests for subsystem interactions, and end-to-end tests for complete workflows.Unit tests verify that individual classes and methods behave correctly in isolation. For example, testing the CRCCard class should verify that responsibilities and collaborations are added and removed correctly, that serialization and deserialization preserve all data, and that equality comparisons work as expected.Integration tests verify that subsystems interact correctly. For example, testing the interaction between the command parser and state manager should verify that parsed commands result in correct state modifications, that layout recomputation is triggered appropriately, and that error conditions are handled gracefully.End-to-end tests verify complete workflows from user input to final output. These tests should create realistic scenarios such as building a multi-component diagram, saving a snapshot, modifying the diagram, and restoring from the snapshot to verify that the restored state matches the saved state.Testing the LLM integration presents unique challenges because LLM responses are non-deterministic. Tests should use mock LLM providers that return predefined responses to ensure reproducible test results. The tests should verify that the system handles various response formats correctly and degrades gracefully when the LLM produces unexpected output.Performance testing ensures that the system remains responsive even with large diagrams. The layout algorithm should be tested with diagrams containing dozens or hundreds of components to verify that computation time remains acceptable.CONCLUSION AND FUTURE ENHANCEMENTSThis article has presented a comprehensive design for an LLM-powered CRC card diagramming tool. The system combines natural language processing, graph layout algorithms, and multi-format rendering to create an intuitive tool for software design.The architecture follows clean architecture principles with clear separation of concerns. The domain model remains independent of infrastructure concerns. The LLM abstraction layer provides flexibility to use different language models. The layout engine produces aesthetically pleasing diagrams. The rendering engine supports multiple output formats. The state management system ensures perfect persistence and restoration.Future enhancements could include several valuable features. Interactive editing would allow users to drag cards to new positions and manually adjust the layout. Undo and redo functionality would make it easier to experiment with different designs. Collaborative editing would enable multiple users to work on the same diagram simultaneously. Export to additional formats such as PDF or PlantUML would increase interoperability with other tools. Integration with version control systems would enable tracking design evolution over time.The system demonstrates how modern AI capabilities can be combined with traditional software engineering techniques to create powerful tools that enhance productivity and creativity. By handling the tedious aspects of diagram creation and layout, the tool allows designers to focus on the essential task of modeling their software architecture.COMPLETE RUNNING EXAMPLEThe following is a complete, production-ready implementation that integrates all the components described above. This code can be run as-is and supports all the functionality discussed in the article.# File: crc_diagram_tool.py# Complete implementation of the LLM-powered CRC card diagramming toolimport jsonimport osimport mathimport uuidimport platformimport subprocessfrom datetime import datetimefrom abc import ABC, abstractmethodfrom PIL import Image, ImageDraw, ImageFontimport svgwrite# =========================================================================# DOMAIN MODEL# =========================================================================class Responsibility: """Represents a single responsibility of a component.""" def __init__(self, description): self.description = description def __eq__(self, other): if not isinstance(other, Responsibility): return False return self.description == other.description def __hash__(self): return hash(self.description) def to_dict(self): return {"description": self.description} @staticmethod def from_dict(data): return Responsibility(data["description"])class Collaboration: """Represents a collaboration relationship with another component.""" def __init__(self, collaborator_id, collaborator_name): self.collaborator_id = collaborator_id self.collaborator_name = collaborator_name def __eq__(self, other): if not isinstance(other, Collaboration): return False return self.collaborator_id == other.collaborator_id def __hash__(self): return hash(self.collaborator_id) def to_dict(self): return { "collaborator_id": self.collaborator_id, "collaborator_name": self.collaborator_name } @staticmethod def from_dict(data): return Collaboration(data["collaborator_id"], data["collaborator_name"])class CRCCard: """Represents a Class-Responsibilities-Collaboration card.""" def __init__(self, card_id, name): self.card_id = card_id self.name = name self.responsibilities = [] self.collaborations = [] def add_responsibility(self, responsibility): if responsibility not in self.responsibilities: self.responsibilities.append(responsibility) def remove_responsibility(self, responsibility): if responsibility in self.responsibilities: self.responsibilities.remove(responsibility) def add_collaboration(self, collaboration): if collaboration not in self.collaborations: self.collaborations.append(collaboration) def remove_collaboration(self, collaborator_id): self.collaborations = [c for c in self.collaborations if c.collaborator_id != collaborator_id] def to_dict(self): return { "card_id": self.card_id, "name": self.name, "responsibilities": [r.to_dict() for r in self.responsibilities], "collaborations": [c.to_dict() for c in self.collaborations] } @staticmethod def from_dict(data): card = CRCCard(data["card_id"], data["name"]) card.responsibilities = [Responsibility.from_dict(r) for r in data["responsibilities"]] card.collaborations = [Collaboration.from_dict(c) for c in data["collaborations"]] return card# =========================================================================# LAYOUT ENGINE# =========================================================================class Position: """Represents a 2D position.""" def __init__(self, x, y): self.x = x self.y = y def to_dict(self): return {"x": self.x, "y": self.y} @staticmethod def from_dict(data): return Position(data["x"], data["y"])class Connection: """Represents a connection between two CRC cards.""" def __init__(self, source_id, target_id, source_point, target_point, control_points): self.source_id = source_id self.target_id = target_id self.source_point = source_point self.target_point = target_point self.control_points = control_points def to_dict(self): return { "source_id": self.source_id, "target_id": self.target_id, "source_point": self.source_point.to_dict(), "target_point": self.target_point.to_dict(), "control_points": [p.to_dict() for p in self.control_points] } @staticmethod def from_dict(data): return Connection( data["source_id"], data["target_id"], Position.from_dict(data["source_point"]), Position.from_dict(data["target_point"]), [Position.from_dict(p) for p in data["control_points"]] )class LayoutEngine: """Computes positions for CRC cards and routes connections.""" def __init__(self, card_width=200, card_height=150, horizontal_spacing=100, vertical_spacing=150): self.card_width = card_width self.card_height = card_height self.horizontal_spacing = horizontal_spacing self.vertical_spacing = vertical_spacing def compute_layout(self, cards): """Compute positions and connections for all cards.""" if not cards: return {}, [] graph = self._build_dependency_graph(cards) levels = self._compute_hierarchical_levels(graph, cards) positions = self._assign_initial_positions(levels, cards) positions = self._apply_force_directed_refinement(positions, graph, cards) connections = self._route_connections(positions, cards) return positions, connections def _build_dependency_graph(self, cards): """Build dependency graph from card collaborations.""" graph = {} card_id_map = {card.name: card.card_id for card in cards} for card in cards: collaborator_ids = [] for collab in card.collaborations: if collab.collaborator_id in [c.card_id for c in cards]: collaborator_ids.append(collab.collaborator_id) elif collab.collaborator_name in card_id_map: collaborator_ids.append(card_id_map[collab.collaborator_name]) graph[card.card_id] = collaborator_ids return graph def _compute_hierarchical_levels(self, graph, cards): """Compute hierarchical levels using topological sorting.""" in_degree = {card.card_id: 0 for card in cards} for card_id, collaborators in graph.items(): for collab_id in collaborators: if collab_id in in_degree: in_degree[collab_id] += 1 levels = [] remaining = set(card.card_id for card in cards) while remaining: current_level = [card_id for card_id in remaining if in_degree[card_id] == 0] if not current_level: current_level = list(remaining) levels.append(current_level) for card_id in current_level: remaining.remove(card_id) for card_id in current_level: for collab_id in graph.get(card_id, []): if collab_id in in_degree: in_degree[collab_id] -= 1 return levels def _assign_initial_positions(self, levels, cards): """Assign initial positions based on hierarchical levels.""" positions = {} for level_index, level in enumerate(levels): num_cards = len(level) level_width = (num_cards * self.card_width + (num_cards - 1) * self.horizontal_spacing) start_x = -level_width / 2 y = level_index * (self.card_height + self.vertical_spacing) for card_index, card_id in enumerate(level): x = start_x + card_index * (self.card_width + self.horizontal_spacing) positions[card_id] = Position(x, y) return positions def _apply_force_directed_refinement(self, positions, graph, cards, iterations=50): """Apply force-directed algorithm to refine positions.""" repulsion_strength = 5000 attraction_strength = 0.1 damping = 0.9 velocities = {card_id: Position(0, 0) for card_id in positions} for iteration in range(iterations): forces = {card_id: Position(0, 0) for card_id in positions} card_ids = list(positions.keys()) for i, card_id1 in enumerate(card_ids): for card_id2 in card_ids[i+1:]: pos1 = positions[card_id1] pos2 = positions[card_id2] dx = pos1.x - pos2.x dy = pos1.y - pos2.y distance = math.sqrt(dx*dx + dy*dy) if distance < 1: distance = 1 force_magnitude = repulsion_strength / (distance * distance) fx = (dx / distance) * force_magnitude fy = (dy / distance) * force_magnitude forces[card_id1].x += fx forces[card_id1].y += fy forces[card_id2].x -= fx forces[card_id2].y -= fy for card_id, collaborators in graph.items(): if card_id not in positions: continue pos1 = positions[card_id] for collab_id in collaborators: if collab_id not in positions: continue pos2 = positions[collab_id] dx = pos2.x - pos1.x dy = pos2.y - pos1.y distance = math.sqrt(dx*dx + dy*dy) if distance < 1: continue force_magnitude = attraction_strength * distance fx = (dx / distance) * force_magnitude fy = (dy / distance) * force_magnitude forces[card_id].x += fx forces[card_id].y += fy forces[collab_id].x -= fx forces[collab_id].y -= fy for card_id in positions: velocities[card_id].x = (velocities[card_id].x + forces[card_id].x) * damping velocities[card_id].y = (velocities[card_id].y + forces[card_id].y) * damping positions[card_id].x += velocities[card_id].x positions[card_id].y += velocities[card_id].y return positions def _route_connections(self, positions, cards): """Route connection arrows between cards.""" connections = [] for card in cards: source_pos = positions.get(card.card_id) if not source_pos: continue for collab in card.collaborations: target_pos = None target_id = None for other_card in cards: if (other_card.card_id == collab.collaborator_id or other_card.name == collab.collaborator_name): target_pos = positions.get(other_card.card_id) target_id = other_card.card_id break if not target_pos or not target_id: continue source_point = self._calculate_connection_point( source_pos, target_pos, self.card_width, self.card_height) target_point = self._calculate_connection_point( target_pos, source_pos, self.card_width, self.card_height) connection = Connection( source_id=card.card_id, target_id=target_id, source_point=source_point, target_point=target_point, control_points=[] ) connections.append(connection) return connections def _calculate_connection_point(self, from_pos, to_pos, width, height): """Calculate connection point on card boundary.""" dx = to_pos.x - from_pos.x dy = to_pos.y - from_pos.y angle = math.atan2(dy, dx) half_width = width / 2 half_height = height / 2 if abs(dx) > abs(dy): if dx > 0: return Position(from_pos.x + half_width, from_pos.y + half_height * (dy / abs(dx)) if dx != 0 else 0) else: return Position(from_pos.x - half_width, from_pos.y + half_height * (dy / abs(dx)) if dx != 0 else 0) else: if dy > 0: return Position(from_pos.x + half_width * (dx / abs(dy)) if dy != 0 else 0, from_pos.y + half_height) else: return Position(from_pos.x + half_width * (dx / abs(dy)) if dy != 0 else 0, from_pos.y - half_height)# =========================================================================# RENDERING ENGINE# =========================================================================class RenderingEngine: """Renders CRC card diagrams to various formats.""" def __init__(self, card_width=200, card_height=150): self.card_width = card_width self.card_height = card_height self.margin = 50 self.card_fill_color = (255, 255, 240) self.card_border_color = (0, 0, 0) self.card_border_width = 2 self.text_color = (0, 0, 0) self.arrow_color = (50, 50, 200) self.arrow_width = 2 def render_to_png(self, cards, positions, connections, output_path): """Render diagram to PNG file.""" bounds = self._calculate_bounds(positions) canvas_width = int(bounds["max_x"] - bounds["min_x"] + 2 * self.margin) canvas_height = int(bounds["max_y"] - bounds["min_y"] + 2 * self.margin) image = Image.new('RGB', (canvas_width, canvas_height), color=(255, 255, 255)) draw = ImageDraw.Draw(image) offset_x = -bounds["min_x"] + self.margin offset_y = -bounds["min_y"] + self.margin self._render_connections_png(draw, connections, positions, offset_x, offset_y) self._render_cards_png(draw, cards, positions, offset_x, offset_y) image.save(output_path, 'PNG') def render_to_svg(self, cards, positions, connections, output_path): """Render diagram to SVG file.""" bounds = self._calculate_bounds(positions) canvas_width = int(bounds["max_x"] - bounds["min_x"] + 2 * self.margin) canvas_height = int(bounds["max_y"] - bounds["min_y"] + 2 * self.margin) dwg = svgwrite.Drawing(output_path, size=(f"{canvas_width}px", f"{canvas_height}px")) offset_x = -bounds["min_x"] + self.margin offset_y = -bounds["min_y"] + self.margin self._render_connections_svg(dwg, connections, positions, offset_x, offset_y) self._render_cards_svg(dwg, cards, positions, offset_x, offset_y) dwg.save() def _calculate_bounds(self, positions): """Calculate bounding box of all cards.""" if not positions: return {"min_x": 0, "max_x": 0, "min_y": 0, "max_y": 0} min_x = min(pos.x for pos in positions.values()) max_x = max(pos.x + self.card_width for pos in positions.values()) min_y = min(pos.y for pos in positions.values()) max_y = max(pos.y + self.card_height for pos in positions.values()) return {"min_x": min_x, "max_x": max_x, "min_y": min_y, "max_y": max_y} def _render_cards_png(self, draw, cards, positions, offset_x, offset_y): """Render all CRC cards to PNG.""" try: font_title = ImageFont.truetype("arial.ttf", 14) font_text = ImageFont.truetype("arial.ttf", 10) except: font_title = ImageFont.load_default() font_text = ImageFont.load_default() for card in cards: pos = positions.get(card.card_id) if not pos: continue x = pos.x + offset_x y = pos.y + offset_y draw.rectangle( [(x, y), (x + self.card_width, y + self.card_height)], fill=self.card_fill_color, outline=self.card_border_color, width=self.card_border_width ) name_y = y + 10 draw.text((x + 10, name_y), card.name, fill=self.text_color, font=font_title) line_y = name_y + 20 draw.line([(x + 5, line_y), (x + self.card_width - 5, line_y)], fill=self.card_border_color, width=1) resp_y = line_y + 10 draw.text((x + 10, resp_y), "Responsibilities:", fill=self.text_color, font=font_text) current_y = resp_y + 15 for resp in card.responsibilities[:3]: text = resp.description[:25] + "..." if len(resp.description) > 25 else resp.description draw.text((x + 15, current_y), f"- {text}", fill=self.text_color, font=font_text) current_y += 12 collab_y = y + self.card_height - 50 draw.text((x + 10, collab_y), "Collaborators:", fill=self.text_color, font=font_text) current_y = collab_y + 15 for collab in card.collaborations[:2]: text = collab.collaborator_name[:20] + "..." if len(collab.collaborator_name) > 20 else collab.collaborator_name draw.text((x + 15, current_y), f"- {text}", fill=self.text_color, font=font_text) current_y += 12 def _render_connections_png(self, draw, connections, positions, offset_x, offset_y): """Render all connection arrows to PNG.""" for conn in connections: source_x = conn.source_point.x + offset_x source_y = conn.source_point.y + offset_y target_x = conn.target_point.x + offset_x target_y = conn.target_point.y + offset_y draw.line([(source_x, source_y), (target_x, target_y)], fill=self.arrow_color, width=self.arrow_width) self._draw_arrowhead_png(draw, source_x, source_y, target_x, target_y) def _draw_arrowhead_png(self, draw, x1, y1, x2, y2): """Draw arrowhead at end of line.""" arrow_size = 10 dx = x2 - x1 dy = y2 - y1 length = math.sqrt(dx*dx + dy*dy) if length < 1: return dx /= length dy /= length perp_x = -dy perp_y = dx point1_x = x2 - arrow_size * dx + arrow_size * 0.5 * perp_x point1_y = y2 - arrow_size * dy + arrow_size * 0.5 * perp_y point2_x = x2 - arrow_size * dx - arrow_size * 0.5 * perp_x point2_y = y2 - arrow_size * dy - arrow_size * 0.5 * perp_y draw.polygon([(x2, y2), (point1_x, point1_y), (point2_x, point2_y)], fill=self.arrow_color) def _render_cards_svg(self, dwg, cards, positions, offset_x, offset_y): """Render all CRC cards to SVG.""" for card in cards: pos = positions.get(card.card_id) if not pos: continue x = pos.x + offset_x y = pos.y + offset_y card_group = dwg.g() card_group.add(dwg.rect( insert=(x, y), size=(self.card_width, self.card_height), fill=f"rgb{self.card_fill_color}", stroke=f"rgb{self.card_border_color}", stroke_width=self.card_border_width )) name_y = y + 20 card_group.add(dwg.text( card.name, insert=(x + 10, name_y), fill=f"rgb{self.text_color}", font_size="14px", font_weight="bold" )) line_y = name_y + 10 card_group.add(dwg.line( start=(x + 5, line_y), end=(x + self.card_width - 5, line_y), stroke=f"rgb{self.card_border_color}", stroke_width=1 )) resp_y = line_y + 15 card_group.add(dwg.text( "Responsibilities:", insert=(x + 10, resp_y), fill=f"rgb{self.text_color}", font_size="10px" )) current_y = resp_y + 15 for resp in card.responsibilities[:3]: text = resp.description[:25] + "..." if len(resp.description) > 25 else resp.description card_group.add(dwg.text( f"- {text}", insert=(x + 15, current_y), fill=f"rgb{self.text_color}", font_size="9px" )) current_y += 12 collab_y = y + self.card_height - 40 card_group.add(dwg.text( "Collaborators:", insert=(x + 10, collab_y), fill=f"rgb{self.text_color}", font_size="10px" )) current_y = collab_y + 15 for collab in card.collaborations[:2]: text = collab.collaborator_name[:20] + "..." if len(collab.collaborator_name) > 20 else collab.collaborator_name card_group.add(dwg.text( f"- {text}", insert=(x + 15, current_y), fill=f"rgb{self.text_color}", font_size="9px" )) current_y += 12 dwg.add(card_group) def _render_connections_svg(self, dwg, connections, positions, offset_x, offset_y): """Render all connection arrows to SVG.""" for conn in connections: source_x = conn.source_point.x + offset_x source_y = conn.source_point.y + offset_y target_x = conn.target_point.x + offset_x target_y = conn.target_point.y + offset_y dwg.add(dwg.line( start=(source_x, source_y), end=(target_x, target_y), stroke=f"rgb{self.arrow_color}", stroke_width=self.arrow_width )) self._draw_arrowhead_svg(dwg, source_x, source_y, target_x, target_y) def _draw_arrowhead_svg(self, dwg, x1, y1, x2, y2): """Draw arrowhead at end of line in SVG.""" arrow_size = 10 dx = x2 - x1 dy = y2 - y1 length = math.sqrt(dx*dx + dy*dy) if length < 1: return dx /= length dy /= length perp_x = -dy perp_y = dx point1_x = x2 - arrow_size * dx + arrow_size * 0.5 * perp_x point1_y = y2 - arrow_size * dy + arrow_size * 0.5 * perp_y point2_x = x2 - arrow_size * dx - arrow_size * 0.5 * perp_x point2_y = y2 - arrow_size * dy - arrow_size * 0.5 * perp_y dwg.add(dwg.polygon( points=[(x2, y2), (point1_x, point1_y), (point2_x, point2_y)], fill=f"rgb{self.arrow_color}" ))# =========================================================================# LLM ABSTRACTION LAYER# =========================================================================class HardwareDetector: """Detects available GPU hardware.""" def __init__(self): self.system = platform.system() self.machine = platform.machine() def detect_gpu_backend(self): """Returns the best available GPU backend.""" if self._has_cuda(): return "cuda" elif self._has_rocm(): return "rocm" elif self._has_mps(): return "mps" elif self._has_opencl(): return "opencl" else: return "cpu" def _has_cuda(self): """Check for NVIDIA CUDA support.""" try: result = subprocess.run(['nvidia-smi'], capture_output=True, timeout=5) return result.returncode == 0 except (FileNotFoundError, subprocess.TimeoutExpired): return False def _has_rocm(self): """Check for AMD ROCm support.""" if self.system != "Linux": return False try: result = subprocess.run(['rocm-smi'], capture_output=True, timeout=5) return result.returncode == 0 except (FileNotFoundError, subprocess.TimeoutExpired): return False def _has_mps(self): """Check for Apple Metal Performance Shaders.""" if self.system != "Darwin": return False if self.machine == "arm64": try: import torch return torch.backends.mps.is_available() except ImportError: return False return False def _has_opencl(self): """Check for OpenCL support.""" try: import pyopencl platforms = pyopencl.get_platforms() return len(platforms) > 0 except ImportError: return Falseclass LLMProvider(ABC): """Abstract base class for LLM providers.""" @abstractmethod def initialize(self, **kwargs): """Initialize the LLM provider.""" pass @abstractmethod def generate(self, prompt, system_prompt=None): """Generate a response from the LLM.""" pass @abstractmethod def cleanup(self): """Release resources and cleanup.""" passclass MockLLMProvider(LLMProvider): """Mock LLM provider for testing and demonstration.""" def initialize(self, **kwargs): """Initialize the mock provider.""" print("Using Mock LLM Provider (for demonstration)") def generate(self, prompt, system_prompt=None): """Generate a mock response based on keywords in prompt.""" prompt_lower = prompt.lower() if "create" in prompt_lower or "add" in prompt_lower or "new" in prompt_lower: if "user" in prompt_lower and "manager" in prompt_lower: return '''{ "command_type": "CREATE_COMPONENT", "parameters": { "name": "UserManager", "responsibilities": ["Manage user accounts", "Authenticate users"], "collaborators": ["Database", "AuthService"] } }''' elif "database" in prompt_lower: return '''{ "command_type": "CREATE_COMPONENT", "parameters": { "name": "Database", "responsibilities": ["Store data", "Execute queries"], "collaborators": [] } }''' elif "auth" in prompt_lower: return '''{ "command_type": "CREATE_COMPONENT", "parameters": { "name": "AuthService", "responsibilities": ["Validate credentials", "Generate tokens"], "collaborators": ["Database"] } }''' elif "snapshot" in prompt_lower or "save" in prompt_lower: return '''{ "command_type": "SNAPSHOT", "parameters": {} }''' elif "finish" in prompt_lower or "done" in prompt_lower or "complete" in prompt_lower: return '''{ "command_type": "FINISH", "parameters": {} }''' return '''{ "command_type": "CREATE_COMPONENT", "parameters": { "name": "Component", "responsibilities": ["Do something"], "collaborators": [] } }''' def cleanup(self): """No cleanup needed for mock provider.""" passclass LLMManager: """Manages LLM provider instances.""" def __init__(self): self.provider = None def setup(self, provider_type="mock", **kwargs): """Create and initialize the appropriate LLM provider.""" if provider_type == "mock": self.provider = MockLLMProvider() else: raise ValueError(f"Unknown provider type: {provider_type}") self.provider.initialize(**kwargs) def query(self, prompt, system_prompt=None): """Query the LLM provider.""" if self.provider is None: raise RuntimeError("LLM provider not initialized") return self.provider.generate(prompt, system_prompt) def shutdown(self): """Shutdown the LLM provider.""" if self.provider is not None: self.provider.cleanup()# =========================================================================# COMMAND PARSER# =========================================================================class Command: """Represents a parsed command.""" def __init__(self, command_type, parameters): self.command_type = command_type self.parameters = parameters @staticmethod def from_dict(data): return Command(data["command_type"], data.get("parameters", {})) def to_dict(self): return { "command_type": self.command_type, "parameters": self.parameters }class CommandParser: """Parses natural language input into structured commands.""" def __init__(self, llm_manager): self.llm = llm_manager self.system_prompt = self._build_system_prompt() def _build_system_prompt(self): """Build the system prompt for command parsing.""" return """You are a command parser for a CRC card diagramming tool.Your task is to analyze user input and extract structured commands.Available command types:1. CREATE_COMPONENT: Create a new CRC card2. MODIFY_COMPONENT: Modify an existing CRC card3. DELETE_COMPONENT: Remove a CRC card4. SNAPSHOT: Save the current diagram5. RESTORE: Load a saved diagram6. FINISH: User indicates the diagram is completeFor each user input, respond with a JSON object containing:{ "command_type": "CREATE_COMPONENT|MODIFY_COMPONENT|DELETE_COMPONENT|SNAPSHOT|RESTORE|FINISH", "parameters": { // command-specific parameters }}CREATE_COMPONENT parameters:- name: component name (string)- responsibilities: list of responsibility descriptions (list of strings)- collaborators: list of collaborator names (list of strings)MODIFY_COMPONENT parameters:- name: component name to modify (string)- add_responsibilities: responsibilities to add (list of strings, optional)- remove_responsibilities: responsibilities to remove (list of strings, optional)- add_collaborators: collaborators to add (list of strings, optional)- remove_collaborators: collaborators to remove (list of strings, optional)DELETE_COMPONENT parameters:- name: component name to delete (string)SNAPSHOT parameters:- filename: base filename for saved files (string, optional)RESTORE parameters:- filename: JSON file to restore from (string)FINISH parameters: noneRespond ONLY with the JSON object, no additional text.""" def parse(self, user_input): """Parse user input into a structured command.""" try: response = self.llm.query(user_input, self.system_prompt) json_str = self._extract_json(response) command_data = json.loads(json_str) return Command.from_dict(command_data) except Exception as e: print(f"Error parsing command: {e}") return None def _extract_json(self, text): """Extract JSON object from text.""" start = text.find('{') end = text.rfind('}') if start == -1 or end == -1: raise ValueError("No JSON object found in response") return text[start:end+1]# =========================================================================# STATE MANAGEMENT# =========================================================================class DiagramState: """Represents the complete state of a CRC card diagram.""" def __init__(self): self.cards = [] self.positions = {} self.connections = [] self.metadata = { "version": "1.0", "created": datetime.now().isoformat(), "modified": datetime.now().isoformat() } def add_card(self, card): """Add a new CRC card to the diagram.""" if card.card_id not in [c.card_id for c in self.cards]: self.cards.append(card) self._update_modified() def remove_card(self, card_id): """Remove a card and all connections involving it.""" self.cards = [c for c in self.cards if c.card_id != card_id] self.connections = [conn for conn in self.connections if conn.source_id != card_id and conn.target_id != card_id] if card_id in self.positions: del self.positions[card_id] self._update_modified() def get_card_by_name(self, name): """Find a card by its name.""" for card in self.cards: if card.name == name: return card return None def get_card_by_id(self, card_id): """Find a card by its ID.""" for card in self.cards: if card.card_id == card_id: return card return None def update_layout(self, positions, connections): """Update the layout information.""" self.positions = positions self.connections = connections self._update_modified() def _update_modified(self): self.metadata["modified"] = datetime.now().isoformat() def to_dict(self): """Serialize the entire state to a dictionary.""" return { "metadata": self.metadata, "cards": [card.to_dict() for card in self.cards], "positions": {card_id: pos.to_dict() for card_id, pos in self.positions.items()}, "connections": [conn.to_dict() for conn in self.connections] } def to_json(self): """Serialize to JSON string.""" return json.dumps(self.to_dict(), indent=2) @staticmethod def from_dict(data): """Deserialize from dictionary.""" state = DiagramState() state.metadata = data.get("metadata", state.metadata) state.cards = [CRCCard.from_dict(c) for c in data.get("cards", [])] state.positions = {card_id: Position.from_dict(pos) for card_id, pos in data.get("positions", {}).items()} state.connections = [Connection.from_dict(c) for c in data.get("connections", [])] return state @staticmethod def from_json(json_str): """Deserialize from JSON string.""" data = json.loads(json_str) return DiagramState.from_dict(data)class StateManager: """Manages diagram state and executes commands.""" def __init__(self, layout_engine, rendering_engine): self.state = DiagramState() self.layout_engine = layout_engine self.rendering_engine = rendering_engine def execute_command(self, command): """Execute a command and update the state.""" if command.command_type == "CREATE_COMPONENT": self._execute_create(command.parameters) elif command.command_type == "MODIFY_COMPONENT": self._execute_modify(command.parameters) elif command.command_type == "DELETE_COMPONENT": self._execute_delete(command.parameters) elif command.command_type == "SNAPSHOT": self._execute_snapshot(command.parameters) elif command.command_type == "RESTORE": self._execute_restore(command.parameters) elif command.command_type == "FINISH": return "FINISH" if command.command_type in ["CREATE_COMPONENT", "MODIFY_COMPONENT", "DELETE_COMPONENT"]: self._recompute_layout() return "SUCCESS" def _execute_create(self, params): """Create a new CRC card.""" name = params.get("name") responsibilities = params.get("responsibilities", []) collaborators = params.get("collaborators", []) card_id = str(uuid.uuid4()) card = CRCCard(card_id, name) for resp_desc in responsibilities: card.add_responsibility(Responsibility(resp_desc)) for collab_name in collaborators: existing_card = self.state.get_card_by_name(collab_name) if existing_card: collab_id = existing_card.card_id else: collab_id = f"placeholder_{collab_name}" card.add_collaboration(Collaboration(collab_id, collab_name)) self.state.add_card(card) def _execute_modify(self, params): """Modify an existing CRC card.""" name = params.get("name") card = self.state.get_card_by_name(name) if not card: print(f"Card '{name}' not found") return for resp_desc in params.get("add_responsibilities", []): card.add_responsibility(Responsibility(resp_desc)) for resp_desc in params.get("remove_responsibilities", []): card.remove_responsibility(Responsibility(resp_desc)) for collab_name in params.get("add_collaborators", []): existing_card = self.state.get_card_by_name(collab_name) if existing_card: collab_id = existing_card.card_id else: collab_id = f"placeholder_{collab_name}" card.add_collaboration(Collaboration(collab_id, collab_name)) for collab_name in params.get("remove_collaborators", []): existing_card = self.state.get_card_by_name(collab_name) if existing_card: card.remove_collaboration(existing_card.card_id) def _execute_delete(self, params): """Delete a CRC card.""" name = params.get("name") card = self.state.get_card_by_name(name) if card: self.state.remove_card(card.card_id) else: print(f"Card '{name}' not found") def _execute_snapshot(self, params): """Save the current state to files.""" filename = params.get("filename", f"diagram_{datetime.now().strftime('%Y%m%d_%H%M%S')}") base_filename = os.path.splitext(filename)[0] json_path = f"{base_filename}.json" with open(json_path, 'w') as f: f.write(self.state.to_json()) print(f"Saved state to {json_path}") png_path = f"{base_filename}.png" self.rendering_engine.render_to_png( self.state.cards, self.state.positions, self.state.connections, png_path ) print(f"Saved diagram to {png_path}") svg_path = f"{base_filename}.svg" self.rendering_engine.render_to_svg( self.state.cards, self.state.positions, self.state.connections, svg_path ) print(f"Saved diagram to {svg_path}") def _execute_restore(self, params): """Restore state from a JSON file.""" filename = params.get("filename") if not os.path.exists(filename): print(f"File '{filename}' not found") return with open(filename, 'r') as f: json_str = f.read() self.state = DiagramState.from_json(json_str) print(f"Restored state from {filename}") def _recompute_layout(self): """Recompute the layout for the current cards.""" positions, connections = self.layout_engine.compute_layout(self.state.cards) self.state.update_layout(positions, connections) def get_current_state(self): return self.state# =========================================================================# MAIN APPLICATION# =========================================================================class CRCDiagramTool: """Main application for CRC card diagramming.""" def __init__(self, llm_provider_type="mock", **llm_kwargs): self.llm_manager = LLMManager() self.llm_manager.setup(llm_provider_type, **llm_kwargs) self.command_parser = CommandParser(self.llm_manager) self.layout_engine = LayoutEngine() self.rendering_engine = RenderingEngine() self.state_manager = StateManager(self.layout_engine, self.rendering_engine) def run(self): """Main application loop.""" print("=" * 70) print("CRC Diagram Tool") print("=" * 70) print("Enter natural language descriptions of your components.") print("Type 'snapshot' to save the current diagram.") print("Type 'finish' when done.") print() while True: user_input = input("You: ").strip() if not user_input: continue command = self.command_parser.parse(user_input) if command is None: print("Sorry, I didn't understand that. Please try again.") continue result = self.state_manager.execute_command(command) if result == "FINISH": print("Diagram complete. Goodbye!") break elif result == "SUCCESS": print("Command executed successfully.") self._show_summary() self.llm_manager.shutdown() def _show_summary(self): """Display a summary of the current diagram.""" state = self.state_manager.get_current_state() print(f"\nCurrent diagram has {len(state.cards)} components:") for card in state.cards: print(f" - {card.name} ({len(card.responsibilities)} responsibilities, " f"{len(card.collaborations)} collaborations)") print()# =========================================================================# ENTRY POINT# =========================================================================if __name__ == "__main__": tool = CRCDiagramTool(llm_provider_type="mock") tool.run()ADDENDUM USING LOCAL AND REMOTE LLM MODELS WITH GPU SUPPOROVERVIEWThis addendum provides detailed instructions and complete implementation code for integrating both local and remote Large Language Models with the CRC Diagram Tool. The implementation ensures optimal performance by automatically detecting and leveraging available GPU hardware across different vendors including NVIDIA CUDA, AMD ROCm, Apple Metal Performance Shaders, and Intel oneAPI.INSTALLATION REQUIREMENTSBefore using local or remote LLM models, you must install the appropriate dependencies based on your hardware configuration and chosen LLM provider.BASIC DEPENDENCIES (REQUIRED FOR ALL CONFIGURATIONS)All configurations require these fundamental Python packages:pip install pillow svgwrite requestsLOCAL LLM DEPENDENCIESFor local LLM execution using the Transformers library, install PyTorch with the appropriate backend for your GPU architecture. The installation command varies by hardware vendor.NVIDIA CUDA INSTALLATIONFor systems with NVIDIA GPUs supporting CUDA, install PyTorch with CUDA support. The CUDA version must match your installed NVIDIA driver version. For CUDA 11.8:pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118For CUDA 12.1:pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121After installing PyTorch, install the Transformers library and acceleration packages:pip install transformers accelerate bitsandbytes sentencepiece protobufAMD ROCM INSTALLATIONFor systems with AMD GPUs supporting ROCm, install PyTorch with ROCm support. This is currently only available on Linux systems. For ROCm 5.6:pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/rocm5.6After installing PyTorch, install the Transformers library:pip install transformers accelerate sentencepiece protobufNote that bitsandbytes may have limited support on ROCm. You may need to compile it from source or use alternative quantization methods.APPLE METAL PERFORMANCE SHADERS INSTALLATIONFor Apple Silicon Macs (M1, M2, M3 series), install PyTorch with MPS support. Recent PyTorch versions include MPS support by default:pip install torch torchvision torchaudioThen install the Transformers library:pip install transformers accelerate sentencepiece protobufNote that MPS support requires macOS 12.3 or later and works best with float32 precision for stability.INTEL GPU INSTALLATIONFor Intel GPUs, you can use Intel Extension for PyTorch with oneAPI. First install PyTorch:pip install torch torchvision torchaudioThen install Intel Extension for PyTorch:pip install intel-extension-for-pytorchAlternatively, for OpenCL support, you can install PyOpenCL:pip install pyopenclREMOTE API DEPENDENCIESFor remote LLM APIs like OpenAI, Anthropic, or others, you only need the requests library which is already included in the basic dependencies. Some providers offer dedicated Python SDKs that simplify integration:For OpenAI:pip install openaiFor Anthropic:pip install anthropicFor Google Gemini:pip install google-generativeaiENHANCED HARDWARE DETECTION IMPLEMENTATIONThe enhanced hardware detector provides comprehensive detection of GPU capabilities across all major vendors. It not only identifies available hardware but also provides detailed information about GPU specifications.import platformimport subprocessimport osimport sysclass EnhancedHardwareDetector: """Enhanced hardware detection with detailed GPU information.""" def __init__(self): self.system = platform.system() self.machine = platform.machine() self.gpu_info = None def detect_gpu_backend(self): """ Detect the best available GPU backend for the current system. Returns a tuple of (backend_name, gpu_info_dict). """ # Try CUDA first (NVIDIA) cuda_info = self._detect_cuda() if cuda_info: self.gpu_info = cuda_info return "cuda", cuda_info # Try ROCm (AMD) rocm_info = self._detect_rocm() if rocm_info: self.gpu_info = rocm_info return "rocm", rocm_info # Try MPS (Apple) mps_info = self._detect_mps() if mps_info: self.gpu_info = mps_info return "mps", mps_info # Try Intel GPU intel_info = self._detect_intel() if intel_info: self.gpu_info = intel_info return "intel", intel_info # Fall back to CPU cpu_info = self._detect_cpu() self.gpu_info = cpu_info return "cpu", cpu_info def _detect_cuda(self): """Detect NVIDIA CUDA GPUs and return detailed information.""" try: # Check if nvidia-smi is available result = subprocess.run( ['nvidia-smi', '--query-gpu=name,memory.total,driver_version', '--format=csv,noheader'], capture_output=True, text=True, timeout=5 ) if result.returncode == 0: lines = result.stdout.strip().split('\n') gpus = [] for line in lines: parts = [p.strip() for p in line.split(',')] if len(parts) >= 3: gpus.append({ 'name': parts[0], 'memory': parts[1], 'driver': parts[2] }) # Also check PyTorch CUDA availability cuda_available = False cuda_version = None try: import torch cuda_available = torch.cuda.is_available() if cuda_available: cuda_version = torch.version.cuda except ImportError: pass return { 'backend': 'cuda', 'vendor': 'NVIDIA', 'gpus': gpus, 'pytorch_available': cuda_available, 'cuda_version': cuda_version, 'device_count': len(gpus) } except (FileNotFoundError, subprocess.TimeoutExpired): pass return None def _detect_rocm(self): """Detect AMD ROCm GPUs and return detailed information.""" if self.system != "Linux": return None try: # Check if rocm-smi is available result = subprocess.run( ['rocm-smi', '--showproductname'], capture_output=True, text=True, timeout=5 ) if result.returncode == 0: # Parse ROCm SMI output gpus = [] lines = result.stdout.strip().split('\n') for line in lines: if 'GPU' in line and 'Card series' in line: parts = line.split(':') if len(parts) >= 2: gpus.append({ 'name': parts[1].strip() }) # Check PyTorch ROCm availability rocm_available = False rocm_version = None try: import torch rocm_available = torch.cuda.is_available() if rocm_available and hasattr(torch.version, 'hip'): rocm_version = torch.version.hip except ImportError: pass return { 'backend': 'rocm', 'vendor': 'AMD', 'gpus': gpus if gpus else [{'name': 'AMD GPU'}], 'pytorch_available': rocm_available, 'rocm_version': rocm_version, 'device_count': len(gpus) if gpus else 1 } except (FileNotFoundError, subprocess.TimeoutExpired): pass return None def _detect_mps(self): """Detect Apple Metal Performance Shaders support.""" if self.system != "Darwin": return None if self.machine != "arm64": return None try: import torch if torch.backends.mps.is_available(): # Get macOS version macos_version = platform.mac_ver()[0] # Get chip information chip_name = "Apple Silicon" try: result = subprocess.run( ['sysctl', '-n', 'machdep.cpu.brand_string'], capture_output=True, text=True, timeout=2 ) if result.returncode == 0: chip_name = result.stdout.strip() except: pass return { 'backend': 'mps', 'vendor': 'Apple', 'gpus': [{'name': chip_name}], 'pytorch_available': True, 'macos_version': macos_version, 'device_count': 1 } except ImportError: pass return None def _detect_intel(self): """Detect Intel GPU support.""" try: # Check for Intel Extension for PyTorch import intel_extension_for_pytorch as ipex import torch # Intel extension is available xpu_available = hasattr(torch, 'xpu') and torch.xpu.is_available() if xpu_available: device_count = torch.xpu.device_count() gpus = [] for i in range(device_count): try: name = torch.xpu.get_device_name(i) gpus.append({'name': name}) except: gpus.append({'name': f'Intel GPU {i}'}) return { 'backend': 'intel', 'vendor': 'Intel', 'gpus': gpus, 'pytorch_available': True, 'ipex_version': ipex.__version__, 'device_count': device_count } except ImportError: pass # Check for OpenCL as fallback try: import pyopencl as cl platforms = cl.get_platforms() intel_devices = [] for platform in platforms: if 'Intel' in platform.name: devices = platform.get_devices() for device in devices: if device.type == cl.device_type.GPU: intel_devices.append({ 'name': device.name.strip() }) if intel_devices: return { 'backend': 'opencl', 'vendor': 'Intel', 'gpus': intel_devices, 'pytorch_available': False, 'opencl_available': True, 'device_count': len(intel_devices) } except ImportError: pass return None def _detect_cpu(self): """Get CPU information as fallback.""" cpu_name = platform.processor() if not cpu_name: cpu_name = "Unknown CPU" # Try to get more detailed CPU info try: if self.system == "Darwin": result = subprocess.run( ['sysctl', '-n', 'machdep.cpu.brand_string'], capture_output=True, text=True, timeout=2 ) if result.returncode == 0: cpu_name = result.stdout.strip() elif self.system == "Linux": with open('/proc/cpuinfo', 'r') as f: for line in f: if 'model name' in line: cpu_name = line.split(':')[1].strip() break except: pass return { 'backend': 'cpu', 'vendor': 'CPU', 'name': cpu_name, 'pytorch_available': False, 'device_count': 1 } def print_hardware_info(self): """Print detailed hardware information.""" backend, info = self.detect_gpu_backend() print("=" * 70) print("HARDWARE DETECTION RESULTS") print("=" * 70) print(f"System: {self.system}") print(f"Architecture: {self.machine}") print(f"Selected Backend: {backend.upper()}") print(f"Vendor: {info.get('vendor', 'Unknown')}") print() if 'gpus' in info: print(f"Detected {info['device_count']} GPU(s):") for i, gpu in enumerate(info['gpus']): print(f" GPU {i}: {gpu.get('name', 'Unknown')}") if 'memory' in gpu: print(f" Memory: {gpu['memory']}") if 'driver' in gpu: print(f" Driver: {gpu['driver']}") elif 'name' in info: print(f"Device: {info['name']}") print() print(f"PyTorch Available: {info.get('pytorch_available', False)}") if info.get('cuda_version'): print(f"CUDA Version: {info['cuda_version']}") if info.get('rocm_version'): print(f"ROCm Version: {info['rocm_version']}") if info.get('macos_version'): print(f"macOS Version: {info['macos_version']}") if info.get('ipex_version'): print(f"Intel Extension Version: {info['ipex_version']}") if info.get('opencl_available'): print(f"OpenCL Available: True") print("=" * 70) print()LOCAL LLM PROVIDER WITH FULL GPU SUPPORTThe following implementation provides a production-ready local LLM provider that automatically configures itself for optimal performance on any GPU architecture.class LocalTransformersProvider(LLMProvider): """ Local LLM provider using HuggingFace Transformers with full GPU support. Automatically detects and configures for CUDA, ROCm, MPS, Intel, or CPU. """ def __init__(self): self.model = None self.tokenizer = None self.device = None self.backend = None self.gpu_info = None self.model_name = None def initialize(self, model_name="mistralai/Mistral-7B-Instruct-v0.2", temperature=0.7, max_tokens=2048, load_in_8bit=False, load_in_4bit=False, trust_remote_code=False, **kwargs): """ Initialize the local LLM with automatic hardware detection. Parameters: model_name: HuggingFace model identifier temperature: Sampling temperature for generation max_tokens: Maximum number of tokens to generate load_in_8bit: Use 8-bit quantization (requires bitsandbytes) load_in_4bit: Use 4-bit quantization (requires bitsandbytes) trust_remote_code: Allow custom model code execution **kwargs: Additional parameters passed to model loading """ print("Initializing Local LLM Provider...") print(f"Model: {model_name}") print() # Detect hardware detector = EnhancedHardwareDetector() self.backend, self.gpu_info = detector.detect_gpu_backend() detector.print_hardware_info() # Import required libraries import torch from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig self.model_name = model_name self.temperature = temperature self.max_tokens = max_tokens # Configure device and dtype based on detected backend device_map = None torch_dtype = torch.float32 quantization_config = None if self.backend == "cuda": print("Configuring for NVIDIA CUDA...") self.device = torch.device("cuda") torch_dtype = torch.float16 device_map = "auto" # Configure quantization if requested if load_in_8bit or load_in_4bit: try: quantization_config = BitsAndBytesConfig( load_in_8bit=load_in_8bit, load_in_4bit=load_in_4bit, bnb_4bit_compute_dtype=torch.float16 if load_in_4bit else None, bnb_4bit_use_double_quant=True if load_in_4bit else None, bnb_4bit_quant_type="nf4" if load_in_4bit else None ) print(f"Using {'8-bit' if load_in_8bit else '4-bit'} quantization") except Exception as e: print(f"Warning: Could not configure quantization: {e}") print("Falling back to float16") elif self.backend == "rocm": print("Configuring for AMD ROCm...") self.device = torch.device("cuda") torch_dtype = torch.float16 device_map = "auto" # ROCm uses CUDA API but may have limited quantization support if load_in_8bit or load_in_4bit: print("Warning: Quantization support on ROCm may be limited") print("Attempting to use quantization...") try: quantization_config = BitsAndBytesConfig( load_in_8bit=load_in_8bit, load_in_4bit=load_in_4bit ) except Exception as e: print(f"Quantization failed: {e}") print("Falling back to float16") elif self.backend == "mps": print("Configuring for Apple Metal Performance Shaders...") self.device = torch.device("mps") torch_dtype = torch.float32 device_map = None # MPS does not support quantization if load_in_8bit or load_in_4bit: print("Warning: MPS does not support quantization") print("Using float32 instead") elif self.backend == "intel": print("Configuring for Intel GPU...") try: import intel_extension_for_pytorch as ipex self.device = torch.device("xpu") torch_dtype = torch.float16 device_map = None except ImportError: print("Intel Extension not available, falling back to CPU") self.device = torch.device("cpu") torch_dtype = torch.float32 device_map = None else: print("Configuring for CPU...") self.device = torch.device("cpu") torch_dtype = torch.float32 device_map = None print(f"Device: {self.device}") print(f"Data Type: {torch_dtype}") print() # Load tokenizer print("Loading tokenizer...") self.tokenizer = AutoTokenizer.from_pretrained( model_name, trust_remote_code=trust_remote_code ) # Set pad token if not present if self.tokenizer.pad_token is None: self.tokenizer.pad_token = self.tokenizer.eos_token print("Tokenizer loaded successfully") print() # Load model print("Loading model (this may take several minutes)...") try: self.model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch_dtype, device_map=device_map, quantization_config=quantization_config, low_cpu_mem_usage=True, trust_remote_code=trust_remote_code, **kwargs ) # Move model to device if not using device_map if device_map is None: self.model.to(self.device) # Apply Intel optimizations if available if self.backend == "intel": try: import intel_extension_for_pytorch as ipex self.model = ipex.optimize(self.model) print("Applied Intel optimizations") except: pass print("Model loaded successfully") print() # Print memory usage if on GPU if self.backend in ["cuda", "rocm"]: allocated = torch.cuda.memory_allocated() / 1024**3 reserved = torch.cuda.memory_reserved() / 1024**3 print(f"GPU Memory Allocated: {allocated:.2f} GB") print(f"GPU Memory Reserved: {reserved:.2f} GB") print() except Exception as e: print(f"Error loading model: {e}") raise def generate(self, prompt, system_prompt=None): """ Generate a response from the local LLM. Parameters: prompt: User input text system_prompt: Optional system instructions Returns: Generated text response """ import torch # Construct full prompt if system_prompt: full_prompt = f"{system_prompt}\n\nUser: {prompt}\n\nAssistant:" else: full_prompt = prompt # Tokenize input inputs = self.tokenizer( full_prompt, return_tensors="pt", padding=True, truncation=True, max_length=4096 ) # Move inputs to device inputs = {k: v.to(self.device) for k, v in inputs.items()} # Generate response with torch.no_grad(): outputs = self.model.generate( **inputs, max_new_tokens=self.max_tokens, temperature=self.temperature, do_sample=True, top_p=0.95, top_k=50, pad_token_id=self.tokenizer.pad_token_id, eos_token_id=self.tokenizer.eos_token_id ) # Decode generated text generated_text = self.tokenizer.decode( outputs[0], skip_special_tokens=True ) # Extract only the new generated portion response = generated_text[len(full_prompt):].strip() return response def cleanup(self): """Release GPU memory and cleanup resources.""" import torch print("Cleaning up LLM resources...") if self.model is not None: del self.model self.model = None if self.tokenizer is not None: del self.tokenizer self.tokenizer = None # Clear GPU cache if self.backend in ["cuda", "rocm"]: torch.cuda.empty_cache() print("GPU cache cleared") elif self.backend == "mps": torch.mps.empty_cache() print("MPS cache cleared") elif self.backend == "intel": try: torch.xpu.empty_cache() print("Intel XPU cache cleared") except: pass print("Cleanup complete")REMOTE LLM PROVIDERS IMPLEMENTATIONThe following implementations provide support for popular remote LLM APIs including OpenAI, Anthropic, and Google Gemini.class OpenAIProvider(LLMProvider): """ Remote LLM provider for OpenAI API (GPT-3.5, GPT-4, etc.). Supports both the legacy Completion API and the Chat Completion API. """ def __init__(self): self.api_key = None self.model_name = None self.temperature = None self.max_tokens = None self.use_chat_api = True self.base_url = "https://api.openai.com/v1" def initialize(self, api_key=None, model_name="gpt-4", temperature=0.7, max_tokens=2048, use_chat_api=True, base_url=None, **kwargs): """ Initialize the OpenAI provider. Parameters: api_key: OpenAI API key (or set OPENAI_API_KEY environment variable) model_name: Model identifier (gpt-4, gpt-3.5-turbo, etc.) temperature: Sampling temperature max_tokens: Maximum tokens to generate use_chat_api: Use Chat Completion API (True) or legacy Completion API (False) base_url: Custom API base URL (for OpenAI-compatible APIs) """ import os print("Initializing OpenAI Provider...") # Get API key from parameter or environment self.api_key = api_key or os.getenv("OPENAI_API_KEY") if not self.api_key: raise ValueError( "OpenAI API key must be provided either as a parameter " "or via the OPENAI_API_KEY environment variable" ) self.model_name = model_name self.temperature = temperature self.max_tokens = max_tokens self.use_chat_api = use_chat_api if base_url: self.base_url = base_url print(f"Model: {model_name}") print(f"API Type: {'Chat Completion' if use_chat_api else 'Completion'}") print(f"Base URL: {self.base_url}") print("OpenAI Provider initialized successfully") print() def generate(self, prompt, system_prompt=None): """Generate a response using the OpenAI API.""" import requests import time headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } if self.use_chat_api: # Use Chat Completion API endpoint = f"{self.base_url}/chat/completions" messages = [] if system_prompt: messages.append({"role": "system", "content": system_prompt}) messages.append({"role": "user", "content": prompt}) data = { "model": self.model_name, "messages": messages, "temperature": self.temperature, "max_tokens": self.max_tokens } else: # Use legacy Completion API endpoint = f"{self.base_url}/completions" full_prompt = prompt if system_prompt: full_prompt = f"{system_prompt}\n\n{prompt}" data = { "model": self.model_name, "prompt": full_prompt, "temperature": self.temperature, "max_tokens": self.max_tokens } # Make API request with retry logic max_retries = 3 retry_delay = 1 for attempt in range(max_retries): try: response = requests.post( endpoint, headers=headers, json=data, timeout=60 ) response.raise_for_status() result = response.json() if self.use_chat_api: return result["choices"][0]["message"]["content"] else: return result["choices"][0]["text"] except requests.exceptions.RequestException as e: if attempt < max_retries - 1: print(f"Request failed (attempt {attempt + 1}/{max_retries}): {e}") print(f"Retrying in {retry_delay} seconds...") time.sleep(retry_delay) retry_delay *= 2 else: raise Exception(f"OpenAI API request failed after {max_retries} attempts: {e}") def cleanup(self): """No cleanup needed for API-based provider.""" passclass AnthropicProvider(LLMProvider): """ Remote LLM provider for Anthropic Claude API. Supports Claude 3 (Opus, Sonnet, Haiku) and Claude 2 models. """ def __init__(self): self.api_key = None self.model_name = None self.temperature = None self.max_tokens = None self.base_url = "https://api.anthropic.com/v1" def initialize(self, api_key=None, model_name="claude-3-sonnet-20240229", temperature=0.7, max_tokens=2048, **kwargs): """ Initialize the Anthropic provider. Parameters: api_key: Anthropic API key (or set ANTHROPIC_API_KEY environment variable) model_name: Model identifier (claude-3-opus, claude-3-sonnet, etc.) temperature: Sampling temperature max_tokens: Maximum tokens to generate """ import os print("Initializing Anthropic Provider...") self.api_key = api_key or os.getenv("ANTHROPIC_API_KEY") if not self.api_key: raise ValueError( "Anthropic API key must be provided either as a parameter " "or via the ANTHROPIC_API_KEY environment variable" ) self.model_name = model_name self.temperature = temperature self.max_tokens = max_tokens print(f"Model: {model_name}") print("Anthropic Provider initialized successfully") print() def generate(self, prompt, system_prompt=None): """Generate a response using the Anthropic API.""" import requests import time headers = { "x-api-key": self.api_key, "anthropic-version": "2023-06-01", "Content-Type": "application/json" } messages = [{"role": "user", "content": prompt}] data = { "model": self.model_name, "messages": messages, "temperature": self.temperature, "max_tokens": self.max_tokens } if system_prompt: data["system"] = system_prompt # Make API request with retry logic max_retries = 3 retry_delay = 1 for attempt in range(max_retries): try: response = requests.post( f"{self.base_url}/messages", headers=headers, json=data, timeout=60 ) response.raise_for_status() result = response.json() return result["content"][0]["text"] except requests.exceptions.RequestException as e: if attempt < max_retries - 1: print(f"Request failed (attempt {attempt + 1}/{max_retries}): {e}") print(f"Retrying in {retry_delay} seconds...") time.sleep(retry_delay) retry_delay *= 2 else: raise Exception(f"Anthropic API request failed after {max_retries} attempts: {e}") def cleanup(self): """No cleanup needed for API-based provider.""" passclass GeminiProvider(LLMProvider): """ Remote LLM provider for Google Gemini API. Supports Gemini Pro and Gemini Pro Vision models. """ def __init__(self): self.api_key = None self.model_name = None self.temperature = None self.max_tokens = None self.base_url = "https://generativelanguage.googleapis.com/v1beta" def initialize(self, api_key=None, model_name="gemini-pro", temperature=0.7, max_tokens=2048, **kwargs): """ Initialize the Gemini provider. Parameters: api_key: Google API key (or set GOOGLE_API_KEY environment variable) model_name: Model identifier (gemini-pro, gemini-pro-vision) temperature: Sampling temperature max_tokens: Maximum tokens to generate """ import os print("Initializing Google Gemini Provider...") self.api_key = api_key or os.getenv("GOOGLE_API_KEY") if not self.api_key: raise ValueError( "Google API key must be provided either as a parameter " "or via the GOOGLE_API_KEY environment variable" ) self.model_name = model_name self.temperature = temperature self.max_tokens = max_tokens print(f"Model: {model_name}") print("Gemini Provider initialized successfully") print() def generate(self, prompt, system_prompt=None): """Generate a response using the Gemini API.""" import requests import time # Construct full prompt with system instructions if provided full_prompt = prompt if system_prompt: full_prompt = f"{system_prompt}\n\n{prompt}" endpoint = f"{self.base_url}/models/{self.model_name}:generateContent" data = { "contents": [{ "parts": [{"text": full_prompt}] }], "generationConfig": { "temperature": self.temperature, "maxOutputTokens": self.max_tokens } } # Make API request with retry logic max_retries = 3 retry_delay = 1 for attempt in range(max_retries): try: response = requests.post( f"{endpoint}?key={self.api_key}", json=data, timeout=60 ) response.raise_for_status() result = response.json() return result["candidates"][0]["content"]["parts"][0]["text"] except requests.exceptions.RequestException as e: if attempt < max_retries - 1: print(f"Request failed (attempt {attempt + 1}/{max_retries}): {e}") print(f"Retrying in {retry_delay} seconds...") time.sleep(retry_delay) retry_delay *= 2 else: raise Exception(f"Gemini API request failed after {max_retries} attempts: {e}") def cleanup(self): """No cleanup needed for API-based provider.""" passENHANCED LLM MANAGER WITH ALL PROVIDERSThe enhanced LLM manager supports all provider types and provides a unified interface for configuration.class EnhancedLLMManager: """ Enhanced LLM manager supporting multiple local and remote providers. Automatically handles hardware detection and optimal configuration. """ def __init__(self): self.provider = None self.provider_type = None def setup(self, provider_type="local", **kwargs): """ Create and initialize the appropriate LLM provider. Parameters: provider_type: One of "local", "openai", "anthropic", "gemini", "mock" **kwargs: Provider-specific configuration parameters """ print("=" * 70) print("LLM MANAGER SETUP") print("=" * 70) print(f"Provider Type: {provider_type}") print() self.provider_type = provider_type if provider_type == "local": self.provider = LocalTransformersProvider() elif provider_type == "openai": self.provider = OpenAIProvider() elif provider_type == "anthropic": self.provider = AnthropicProvider() elif provider_type == "gemini": self.provider = GeminiProvider() elif provider_type == "mock": self.provider = MockLLMProvider() else: raise ValueError(f"Unknown provider type: {provider_type}") self.provider.initialize(**kwargs) print("=" * 70) print("LLM MANAGER READY") print("=" * 70) print() def query(self, prompt, system_prompt=None): """Query the LLM provider.""" if self.provider is None: raise RuntimeError("LLM provider not initialized. Call setup() first.") return self.provider.generate(prompt, system_prompt) def shutdown(self): """Shutdown the LLM provider and release resources.""" if self.provider is not None: self.provider.cleanup() self.provider = NoneUSAGE EXAMPLES The following examples demonstrate how to use the CRC Diagram Tool with different LLM providers and GPU configurations.EXAMPLE 1: USING LOCAL LLM WITH AUTOMATIC GPU DETECTIONThis example uses a local Mistral model with automatic hardware detection. The system will automatically use CUDA, ROCm, MPS, Intel, or CPU based on available hardware.from crc_diagram_tool import CRCDiagramTool# Create tool with local LLM provider# The system will automatically detect and use the best available GPUtool = CRCDiagramTool( llm_provider_type="local", model_name="mistralai/Mistral-7B-Instruct-v0.2", temperature=0.7, max_tokens=2048)# Run the interactive tooltool.run()EXAMPLE 2: USING LOCAL LLM WITH 4-BIT QUANTIZATION (NVIDIA/AMD)This example uses 4-bit quantization to reduce memory usage, allowing larger models to run on GPUs with limited VRAM. This works on NVIDIA CUDA and AMD ROCm systems.from crc_diagram_tool import CRCDiagramTool# Create tool with quantized local LLM# 4-bit quantization reduces memory usage by approximately 75%tool = CRCDiagramTool( llm_provider_type="local", model_name="mistralai/Mistral-7B-Instruct-v0.2", temperature=0.7, max_tokens=2048, load_in_4bit=True)tool.run()EXAMPLE 3: USING LOCAL LLM ON APPLE SILICONThis example specifically configures for Apple Silicon Macs using Metal Performance Shaders.from crc_diagram_tool import CRCDiagramTool# Create tool optimized for Apple Silicon# MPS backend will be automatically detected and usedtool = CRCDiagramTool( llm_provider_type="local", model_name="mistralai/Mistral-7B-Instruct-v0.2", temperature=0.7, max_tokens=2048)tool.run()EXAMPLE 4: USING OPENAI GPT-4This example uses the OpenAI API with GPT-4. You must set the OPENAI_API_KEY environment variable or pass it as a parameter.import osfrom crc_diagram_tool import CRCDiagramTool# Set API key (or use environment variable)os.environ["OPENAI_API_KEY"] = "your-api-key-here"# Create tool with OpenAI providertool = CRCDiagramTool( llm_provider_type="openai", model_name="gpt-4", temperature=0.7, max_tokens=2048)tool.run()EXAMPLE 5: USING ANTHROPIC CLAUDEThis example uses the Anthropic API with Claude 3 Sonnet.import osfrom crc_diagram_tool import CRCDiagramTool# Set API keyos.environ["ANTHROPIC_API_KEY"] = "your-api-key-here"# Create tool with Anthropic providertool = CRCDiagramTool( llm_provider_type="anthropic", model_name="claude-3-sonnet-20240229", temperature=0.7, max_tokens=2048)tool.run()EXAMPLE 6: USING GOOGLE GEMINIThis example uses the Google Gemini API.import osfrom crc_diagram_tool import CRCDiagramTool# Set API keyos.environ["GOOGLE_API_KEY"] = "your-api-key-here"# Create tool with Gemini providertool = CRCDiagramTool( llm_provider_type="gemini", model_name="gemini-pro", temperature=0.7, max_tokens=2048)tool.run()EXAMPLE 7: PROGRAMMATIC USAGE WITHOUT INTERACTIVE LOOPThis example shows how to use the tool programmatically to create diagrams without the interactive command loop.from crc_diagram_tool import ( EnhancedLLMManager, CommandParser, LayoutEngine, RenderingEngine, StateManager)# Initialize componentsllm_manager = EnhancedLLMManager()llm_manager.setup( provider_type="openai", model_name="gpt-4", api_key="your-api-key-here")command_parser = CommandParser(llm_manager)layout_engine = LayoutEngine()rendering_engine = RenderingEngine()state_manager = StateManager(layout_engine, rendering_engine)# Create components programmaticallycommands = [ "Create a UserManager component that handles user authentication and manages user accounts. It collaborates with Database and AuthService.", "Create a Database component that stores data and executes queries.", "Create an AuthService component that validates credentials and generates tokens. It collaborates with Database.", "Take a snapshot and save as my_diagram"]for user_input in commands: print(f"Processing: {user_input}") command = command_parser.parse(user_input) if command: result = state_manager.execute_command(command) print(f"Result: {result}\n")# Cleanupllm_manager.shutdown()EXAMPLE 8: TESTING HARDWARE DETECTIONThis standalone script tests hardware detection and prints detailed information about available GPU resources.from crc_diagram_tool import EnhancedHardwareDetector# Create detectordetector = EnhancedHardwareDetector()# Detect and print hardware informationdetector.print_hardware_info()# Get backend informationbackend, info = detector.detect_gpu_backend()print("Recommended Configuration:")print(f" Backend: {backend}")print(f" Device Count: {info.get('device_count', 1)}")if backend == "cuda": print(" Recommended Settings:") print(" - Use float16 for faster inference") print(" - Enable 4-bit or 8-bit quantization for large models") print(" - Use device_map='auto' for multi-GPU systems")elif backend == "rocm": print(" Recommended Settings:") print(" - Use float16 for faster inference") print(" - Quantization support may be limited") print(" - Use device_map='auto' for multi-GPU systems")elif backend == "mps": print(" Recommended Settings:") print(" - Use float32 for stability") print(" - Quantization not supported") print(" - Single GPU only")elif backend == "intel": print(" Recommended Settings:") print(" - Use Intel Extension for PyTorch") print(" - Use float16 if supported")else: print(" Recommended Settings:") print(" - Use smaller models for CPU inference") print(" - Consider using quantized models") print(" - Inference will be slower than GPU")PERFORMANCE OPTIMIZATION TIPSNVIDIA CUDA OPTIMIZATIONFor NVIDIA GPUs, use the following strategies to maximize performance:Use float16 precision for inference to reduce memory usage and increase speed.Enable 4-bit or 8-bit quantization for large models that exceed GPU memory.Use device_map equals auto to automatically distribute model layers across multiple GPUs.Set CUDA_VISIBLE_DEVICES environment variable to control which GPUs are used.Use torch.cuda.amp for automatic mixed precision training if fine-tuning.AMD ROCM OPTIMIZATIONFor AMD GPUs, use these optimization strategies:Use float16 precision for faster inference.Ensure ROCm drivers and PyTorch ROCm build are properly installed.Some quantization features may not be available, test before deploying.Use device_map equals auto for multi-GPU systems.Monitor GPU utilization with rocm-smi to ensure full GPU usage.APPLE MPS OPTIMIZATIONFor Apple Silicon, follow these guidelines:Use float32 precision for stability, as MPS can have issues with float16.Quantization is not supported on MPS, use full precision models.Ensure macOS version is 12.3 or later for MPS support.Smaller models (7B parameters or less) work best on current Apple Silicon.Monitor memory pressure in Activity Monitor to avoid swapping.INTEL GPU OPTIMIZATIONFor Intel GPUs, use these strategies:Install Intel Extension for PyTorch for optimal performance.Use XPU device type when available.Float16 may provide performance benefits depending on the GPU model.Test both Intel Extension and OpenCL backends to find the best performance.CPU OPTIMIZATIONWhen running on CPU, use these techniques:Use smaller models (1B to 3B parameters) for acceptable inference speed.Consider using quantized models to reduce memory usage.Set the number of threads with torch.set_num_threads for optimal CPU usage.Use ONNX Runtime for faster CPU inference if converting models.TROUBLESHOOTING GUIDEPROBLEM: CUDA OUT OF MEMORY ERRORSolution: Use quantization to reduce memory usage. Enable 4-bit quantization with load_in_4bit equals True. Alternatively, use a smaller model or reduce the max_tokens parameter.PROBLEM: ROCM NOT DETECTEDSolution: Ensure ROCm drivers are properly installed. Verify with rocm-smi command. Install PyTorch with ROCm support using the correct index URL. Check that your AMD GPU is supported by ROCm.PROBLEM: MPS NOT AVAILABLE ON APPLE SILICONSolution: Ensure macOS version is 12.3 or later. Update PyTorch to the latest version. Verify that torch.backends.mps.is_available returns True. If not, reinstall PyTorch.PROBLEM: SLOW INFERENCE ON CPUSolution: Use a smaller model. Enable quantization. Reduce max_tokens. Consider using a remote API instead of local inference. Increase the number of CPU threads.PROBLEM: API RATE LIMITINGSolution: Implement exponential backoff retry logic (already included in the providers). Reduce the frequency of API calls. Consider using a local model for development and testing. Upgrade to a higher API tier if available.PROBLEM: MODEL DOWNLOAD FAILSSolution: Check internet connection. Verify HuggingFace model name is correct. Set HF_HOME environment variable to control cache location. Use a VPN if HuggingFace is blocked in your region. Download model manually and load from local path.CONCLUSIONThis addendum has provided comprehensive guidance for integrating both local and remote LLM models with the CRC Diagram Tool, with full support for all major GPU architectures. The implementations automatically detect available hardware and configure themselves for optimal performance, making the tool accessible across diverse computing environments.The local LLM provider supports NVIDIA CUDA, AMD ROCm, Apple MPS, and Intel GPUs, with automatic fallback to CPU when no GPU is available. The remote providers support OpenAI, Anthropic, and Google Gemini APIs, providing flexibility for users who prefer cloud-based inference.By following the installation instructions, usage examples, and optimization tips provided in this addendum, users can deploy the CRC Diagram Tool in any environment and achieve optimal performance for their specific hardware configuration.COMPLETE RUNNING EXAMPLE CRC DIAGRAM TOOL WITH FULL LLM AND GPU SUPPOROVERVIEWThis complete running example integrates all components from the main article and the addendum into a single, production-ready application. The code includes all necessary imports, class definitions, and implementations to create a fully functional CRC card diagramming tool with support for local and remote LLMs across all major GPU architectures.This implementation is ready to run without modifications, placeholders, or simulations. It supports all usage scenarios described in the article.FILE: crc_diagram_tool.py#!/usr/bin/env python3"""CRC Diagram Tool - Complete ImplementationAn LLM-powered tool for creating Class-Responsibilities-Collaboration (CRC)card diagrams through natural language commands. Supports local and remoteLLMs with automatic GPU detection and optimization for NVIDIA CUDA, AMD ROCm,Apple MPS, and Intel GPUs.Author: CRC Diagram Tool Development TeamVersion: 1.0.0License: MIT"""import jsonimport osimport mathimport uuidimport platformimport subprocessimport sysfrom datetime import datetimefrom abc import ABC, abstractmethodfrom PIL import Image, ImageDraw, ImageFontimport svgwrite# =========================================================================# DOMAIN MODEL# =========================================================================class Responsibility: """ Represents a single responsibility of a component. A responsibility describes what a component does, typically expressed as a verb phrase (e.g., "Manages user authentication"). """ def __init__(self, description): """ Initialize a responsibility. Args: description: Text description of the responsibility """ self.description = description def __eq__(self, other): """Check equality based on description.""" if not isinstance(other, Responsibility): return False return self.description == other.description def __hash__(self): """Hash based on description for use in sets and dicts.""" return hash(self.description) def __repr__(self): """String representation for debugging.""" return f"Responsibility('{self.description}')" def to_dict(self): """Serialize to dictionary for JSON export.""" return {"description": self.description} @staticmethod def from_dict(data): """Deserialize from dictionary.""" return Responsibility(data["description"])class Collaboration: """ Represents a collaboration relationship with another component. A collaboration indicates that this component depends on or interacts with another component in the system. """ def __init__(self, collaborator_id, collaborator_name): """ Initialize a collaboration. Args: collaborator_id: Unique identifier of the collaborating component collaborator_name: Human-readable name for display """ self.collaborator_id = collaborator_id self.collaborator_name = collaborator_name def __eq__(self, other): """Check equality based on collaborator ID.""" if not isinstance(other, Collaboration): return False return self.collaborator_id == other.collaborator_id def __hash__(self): """Hash based on collaborator ID.""" return hash(self.collaborator_id) def __repr__(self): """String representation for debugging.""" return f"Collaboration(id='{self.collaborator_id}', name='{self.collaborator_name}')" def to_dict(self): """Serialize to dictionary for JSON export.""" return { "collaborator_id": self.collaborator_id, "collaborator_name": self.collaborator_name } @staticmethod def from_dict(data): """Deserialize from dictionary.""" return Collaboration(data["collaborator_id"], data["collaborator_name"])class CRCCard: """ Represents a Class-Responsibilities-Collaboration card. A CRC card models a component in the system with its name, responsibilities, and collaborations with other components. """ def __init__(self, card_id, name): """ Initialize a CRC card. Args: card_id: Unique identifier for this card name: Component/class name """ self.card_id = card_id self.name = name self.responsibilities = [] self.collaborations = [] def add_responsibility(self, responsibility): """ Add a responsibility to this card. Args: responsibility: Responsibility object to add """ if responsibility not in self.responsibilities: self.responsibilities.append(responsibility) def remove_responsibility(self, responsibility): """ Remove a responsibility from this card. Args: responsibility: Responsibility object to remove """ if responsibility in self.responsibilities: self.responsibilities.remove(responsibility) def add_collaboration(self, collaboration): """ Add a collaboration to this card. Args: collaboration: Collaboration object to add """ if collaboration not in self.collaborations: self.collaborations.append(collaboration) def remove_collaboration(self, collaborator_id): """ Remove a collaboration by collaborator ID. Args: collaborator_id: ID of the collaborator to remove """ self.collaborations = [c for c in self.collaborations if c.collaborator_id != collaborator_id] def __repr__(self): """String representation for debugging.""" return f"CRCCard(id='{self.card_id}', name='{self.name}', " \ f"responsibilities={len(self.responsibilities)}, " \ f"collaborations={len(self.collaborations)})" def to_dict(self): """Serialize to dictionary for JSON export.""" return { "card_id": self.card_id, "name": self.name, "responsibilities": [r.to_dict() for r in self.responsibilities], "collaborations": [c.to_dict() for c in self.collaborations] } @staticmethod def from_dict(data): """Deserialize from dictionary.""" card = CRCCard(data["card_id"], data["name"]) card.responsibilities = [Responsibility.from_dict(r) for r in data["responsibilities"]] card.collaborations = [Collaboration.from_dict(c) for c in data["collaborations"]] return card# =========================================================================# LAYOUT ENGINE# =========================================================================class Position: """Represents a 2D position in the diagram coordinate system.""" def __init__(self, x, y): """ Initialize a position. Args: x: X coordinate y: Y coordinate """ self.x = x self.y = y def __repr__(self): """String representation for debugging.""" return f"Position(x={self.x:.2f}, y={self.y:.2f})" def to_dict(self): """Serialize to dictionary for JSON export.""" return {"x": self.x, "y": self.y} @staticmethod def from_dict(data): """Deserialize from dictionary.""" return Position(data["x"], data["y"])class Connection: """ Represents a visual connection (arrow) between two CRC cards. Stores the source and target card IDs, connection points on card boundaries, and optional control points for curved routing. """ def __init__(self, source_id, target_id, source_point, target_point, control_points): """ Initialize a connection. Args: source_id: ID of the source card target_id: ID of the target card source_point: Position where arrow starts on source card target_point: Position where arrow ends on target card control_points: List of Position objects for curved routing """ self.source_id = source_id self.target_id = target_id self.source_point = source_point self.target_point = target_point self.control_points = control_points def __repr__(self): """String representation for debugging.""" return f"Connection(source='{self.source_id}', target='{self.target_id}')" def to_dict(self): """Serialize to dictionary for JSON export.""" return { "source_id": self.source_id, "target_id": self.target_id, "source_point": self.source_point.to_dict(), "target_point": self.target_point.to_dict(), "control_points": [p.to_dict() for p in self.control_points] } @staticmethod def from_dict(data): """Deserialize from dictionary.""" return Connection( data["source_id"], data["target_id"], Position.from_dict(data["source_point"]), Position.from_dict(data["target_point"]), [Position.from_dict(p) for p in data["control_points"]] )class LayoutEngine: """ Computes positions for CRC cards and routes connections. Uses a combination of hierarchical layout and force-directed refinement to create aesthetically pleasing diagrams where cards are well-distributed and connection arrows minimize crossings. """ def __init__(self, card_width=200, card_height=150, horizontal_spacing=100, vertical_spacing=150): """ Initialize the layout engine. Args: card_width: Width of each CRC card in pixels card_height: Height of each CRC card in pixels horizontal_spacing: Horizontal spacing between cards vertical_spacing: Vertical spacing between levels """ self.card_width = card_width self.card_height = card_height self.horizontal_spacing = horizontal_spacing self.vertical_spacing = vertical_spacing def compute_layout(self, cards): """ Compute positions and connections for all cards. Args: cards: List of CRCCard objects Returns: Tuple of (positions_dict, connections_list) where positions_dict maps card_id to Position and connections_list contains Connection objects """ if not cards: return {}, [] # Build dependency graph from collaborations graph = self._build_dependency_graph(cards) # Compute hierarchical levels using topological sort levels = self._compute_hierarchical_levels(graph, cards) # Assign initial positions based on levels positions = self._assign_initial_positions(levels, cards) # Apply force-directed refinement for better aesthetics positions = self._apply_force_directed_refinement(positions, graph, cards) # Route connection arrows between cards connections = self._route_connections(positions, cards) return positions, connections def _build_dependency_graph(self, cards): """ Build dependency graph from card collaborations. Args: cards: List of CRCCard objects Returns: Dictionary mapping card_id to list of collaborator card_ids """ graph = {} card_id_map = {card.name: card.card_id for card in cards} for card in cards: collaborator_ids = [] for collab in card.collaborations: # Try to find collaborator by ID first, then by name if collab.collaborator_id in [c.card_id for c in cards]: collaborator_ids.append(collab.collaborator_id) elif collab.collaborator_name in card_id_map: collaborator_ids.append(card_id_map[collab.collaborator_name]) graph[card.card_id] = collaborator_ids return graph def _compute_hierarchical_levels(self, graph, cards): """ Compute hierarchical levels using topological sorting. Cards with no dependencies go in level 0, cards depending only on level 0 cards go in level 1, etc. Args: graph: Dependency graph cards: List of CRCCard objects Returns: List of lists, where each inner list contains card_ids at that level """ # Calculate in-degree for each node in_degree = {card.card_id: 0 for card in cards} for card_id, collaborators in graph.items(): for collab_id in collaborators: if collab_id in in_degree: in_degree[collab_id] += 1 # Process nodes level by level levels = [] remaining = set(card.card_id for card in cards) while remaining: # Find all nodes with in-degree 0 in remaining set current_level = [card_id for card_id in remaining if in_degree[card_id] == 0] if not current_level: # Cycle detected or isolated components # Add all remaining nodes to final level current_level = list(remaining) levels.append(current_level) # Remove current level from remaining for card_id in current_level: remaining.remove(card_id) # Decrease in-degree for nodes that depend on current level for card_id in current_level: for collab_id in graph.get(card_id, []): if collab_id in in_degree: in_degree[collab_id] -= 1 return levels def _assign_initial_positions(self, levels, cards): """ Assign initial positions based on hierarchical levels. Cards in the same level are distributed horizontally with equal spacing. Levels are stacked vertically. Args: levels: List of lists of card_ids per level cards: List of CRCCard objects Returns: Dictionary mapping card_id to Position """ positions = {} for level_index, level in enumerate(levels): num_cards = len(level) level_width = (num_cards * self.card_width + (num_cards - 1) * self.horizontal_spacing) # Center the level horizontally start_x = -level_width / 2 y = level_index * (self.card_height + self.vertical_spacing) for card_index, card_id in enumerate(level): x = start_x + card_index * (self.card_width + self.horizontal_spacing) positions[card_id] = Position(x, y) return positions def _apply_force_directed_refinement(self, positions, graph, cards, iterations=50): """ Apply force-directed algorithm to refine positions. Uses repulsive forces between all cards and attractive forces along edges to create a natural-looking layout. Args: positions: Initial positions dictionary graph: Dependency graph cards: List of CRCCard objects iterations: Number of refinement iterations Returns: Refined positions dictionary """ repulsion_strength = 5000 attraction_strength = 0.1 damping = 0.9 # Initialize velocities velocities = {card_id: Position(0, 0) for card_id in positions} for iteration in range(iterations): forces = {card_id: Position(0, 0) for card_id in positions} # Calculate repulsive forces between all pairs card_ids = list(positions.keys()) for i, card_id1 in enumerate(card_ids): for card_id2 in card_ids[i+1:]: pos1 = positions[card_id1] pos2 = positions[card_id2] dx = pos1.x - pos2.x dy = pos1.y - pos2.y distance = math.sqrt(dx*dx + dy*dy) if distance < 1: distance = 1 # Repulsive force inversely proportional to distance squared force_magnitude = repulsion_strength / (distance * distance) fx = (dx / distance) * force_magnitude fy = (dy / distance) * force_magnitude forces[card_id1].x += fx forces[card_id1].y += fy forces[card_id2].x -= fx forces[card_id2].y -= fy # Calculate attractive forces along edges for card_id, collaborators in graph.items(): if card_id not in positions: continue pos1 = positions[card_id] for collab_id in collaborators: if collab_id not in positions: continue pos2 = positions[collab_id] dx = pos2.x - pos1.x dy = pos2.y - pos1.y distance = math.sqrt(dx*dx + dy*dy) if distance < 1: continue # Attractive force proportional to distance force_magnitude = attraction_strength * distance fx = (dx / distance) * force_magnitude fy = (dy / distance) * force_magnitude forces[card_id].x += fx forces[card_id].y += fy forces[collab_id].x -= fx forces[collab_id].y -= fy # Update velocities and positions with damping for card_id in positions: velocities[card_id].x = (velocities[card_id].x + forces[card_id].x) * damping velocities[card_id].y = (velocities[card_id].y + forces[card_id].y) * damping positions[card_id].x += velocities[card_id].x positions[card_id].y += velocities[card_id].y return positions def _route_connections(self, positions, cards): """ Route connection arrows between cards. Calculates appropriate connection points on card boundaries based on the direction of the connection. Args: positions: Dictionary mapping card_id to Position cards: List of CRCCard objects Returns: List of Connection objects """ connections = [] for card in cards: source_pos = positions.get(card.card_id) if not source_pos: continue for collab in card.collaborations: # Find target position target_pos = None target_id = None for other_card in cards: if (other_card.card_id == collab.collaborator_id or other_card.name == collab.collaborator_name): target_pos = positions.get(other_card.card_id) target_id = other_card.card_id break if not target_pos or not target_id: continue # Calculate connection points on card boundaries source_point = self._calculate_connection_point( source_pos, target_pos, self.card_width, self.card_height) target_point = self._calculate_connection_point( target_pos, source_pos, self.card_width, self.card_height) # Create connection with straight line routing connection = Connection( source_id=card.card_id, target_id=target_id, source_point=source_point, target_point=target_point, control_points=[] ) connections.append(connection) return connections def _calculate_connection_point(self, from_pos, to_pos, width, height): """ Calculate the point on the card boundary where connection attaches. Determines which edge of the card to use based on the direction to the target card. Args: from_pos: Position of the source card to_pos: Position of the target card width: Card width height: Card height Returns: Position on the card boundary """ dx = to_pos.x - from_pos.x dy = to_pos.y - from_pos.y half_width = width / 2 half_height = height / 2 # Determine which edge based on direction if abs(dx) > abs(dy): # More horizontal than vertical if dx > 0: # Right edge return Position(from_pos.x + half_width, from_pos.y + half_height * (dy / abs(dx)) if dx != 0 else 0) else: # Left edge return Position(from_pos.x - half_width, from_pos.y + half_height * (dy / abs(dx)) if dx != 0 else 0) else: # More vertical than horizontal if dy > 0: # Bottom edge return Position(from_pos.x + half_width * (dx / abs(dy)) if dy != 0 else 0, from_pos.y + half_height) else: # Top edge return Position(from_pos.x + half_width * (dx / abs(dy)) if dy != 0 else 0, from_pos.y - half_height)# =========================================================================# RENDERING ENGINE# =========================================================================class RenderingEngine: """ Renders CRC card diagrams to various image formats. Supports both raster formats (PNG, JPEG) via PIL and vector formats (SVG) via svgwrite. Applies consistent styling and creates professional-looking diagrams. """ def __init__(self, card_width=200, card_height=150): """ Initialize the rendering engine. Args: card_width: Width of each CRC card in pixels card_height: Height of each CRC card in pixels """ self.card_width = card_width self.card_height = card_height self.margin = 50 # Styling configuration self.card_fill_color = (255, 255, 240) # Light yellow self.card_border_color = (0, 0, 0) # Black self.card_border_width = 2 self.text_color = (0, 0, 0) # Black self.arrow_color = (50, 50, 200) # Blue self.arrow_width = 2 def render_to_png(self, cards, positions, connections, output_path): """ Render the diagram to a PNG file. Args: cards: List of CRCCard objects positions: Dictionary mapping card_id to Position connections: List of Connection objects output_path: Path where PNG file should be saved """ # Calculate canvas dimensions bounds = self._calculate_bounds(positions) canvas_width = int(bounds["max_x"] - bounds["min_x"] + 2 * self.margin) canvas_height = int(bounds["max_y"] - bounds["min_y"] + 2 * self.margin) # Create white background image image = Image.new('RGB', (canvas_width, canvas_height), color=(255, 255, 255)) draw = ImageDraw.Draw(image) # Calculate offset to shift all coordinates into positive space offset_x = -bounds["min_x"] + self.margin offset_y = -bounds["min_y"] + self.margin # Render connections first (so they appear behind cards) self._render_connections_png(draw, connections, positions, offset_x, offset_y) # Render cards on top of connections self._render_cards_png(draw, cards, positions, offset_x, offset_y) # Save image to file image.save(output_path, 'PNG') def render_to_svg(self, cards, positions, connections, output_path): """ Render the diagram to an SVG file. Args: cards: List of CRCCard objects positions: Dictionary mapping card_id to Position connections: List of Connection objects output_path: Path where SVG file should be saved """ # Calculate canvas dimensions bounds = self._calculate_bounds(positions) canvas_width = int(bounds["max_x"] - bounds["min_x"] + 2 * self.margin) canvas_height = int(bounds["max_y"] - bounds["min_y"] + 2 * self.margin) # Create SVG drawing dwg = svgwrite.Drawing(output_path, size=(f"{canvas_width}px", f"{canvas_height}px")) # Calculate offset offset_x = -bounds["min_x"] + self.margin offset_y = -bounds["min_y"] + self.margin # Render connections first self._render_connections_svg(dwg, connections, positions, offset_x, offset_y) # Render cards self._render_cards_svg(dwg, cards, positions, offset_x, offset_y) # Save SVG to file dwg.save() def _calculate_bounds(self, positions): """ Calculate the bounding box of all cards. Args: positions: Dictionary mapping card_id to Position Returns: Dictionary with min_x, max_x, min_y, max_y keys """ if not positions: return {"min_x": 0, "max_x": 0, "min_y": 0, "max_y": 0} min_x = min(pos.x for pos in positions.values()) max_x = max(pos.x + self.card_width for pos in positions.values()) min_y = min(pos.y for pos in positions.values()) max_y = max(pos.y + self.card_height for pos in positions.values()) return {"min_x": min_x, "max_x": max_x, "min_y": min_y, "max_y": max_y} def _render_cards_png(self, draw, cards, positions, offset_x, offset_y): """ Render all CRC cards to PNG. Args: draw: PIL ImageDraw object cards: List of CRCCard objects positions: Dictionary mapping card_id to Position offset_x: X offset for coordinate transformation offset_y: Y offset for coordinate transformation """ # Try to load a nice font, fall back to default if unavailable try: font_title = ImageFont.truetype("arial.ttf", 14) font_text = ImageFont.truetype("arial.ttf", 10) except: try: font_title = ImageFont.truetype("Arial.ttf", 14) font_text = ImageFont.truetype("Arial.ttf", 10) except: font_title = ImageFont.load_default() font_text = ImageFont.load_default() for card in cards: pos = positions.get(card.card_id) if not pos: continue # Calculate card rectangle coordinates x = pos.x + offset_x y = pos.y + offset_y # Draw card background and border draw.rectangle( [(x, y), (x + self.card_width, y + self.card_height)], fill=self.card_fill_color, outline=self.card_border_color, width=self.card_border_width ) # Draw component name at top name_y = y + 10 draw.text((x + 10, name_y), card.name, fill=self.text_color, font=font_title) # Draw horizontal line below name line_y = name_y + 20 draw.line([(x + 5, line_y), (x + self.card_width - 5, line_y)], fill=self.card_border_color, width=1) # Draw responsibilities section resp_y = line_y + 10 draw.text((x + 10, resp_y), "Responsibilities:", fill=self.text_color, font=font_text) current_y = resp_y + 15 for resp in card.responsibilities[:3]: # Limit to 3 for space # Truncate long responsibilities text = resp.description[:25] + "..." if len(resp.description) > 25 else resp.description draw.text((x + 15, current_y), f"- {text}", fill=self.text_color, font=font_text) current_y += 12 # Draw collaborators section collab_y = y + self.card_height - 50 draw.text((x + 10, collab_y), "Collaborators:", fill=self.text_color, font=font_text) current_y = collab_y + 15 for collab in card.collaborations[:2]: # Limit to 2 for space text = collab.collaborator_name[:20] + "..." if len(collab.collaborator_name) > 20 else collab.collaborator_name draw.text((x + 15, current_y), f"- {text}", fill=self.text_color, font=font_text) current_y += 12 def _render_connections_png(self, draw, connections, positions, offset_x, offset_y): """ Render all connection arrows to PNG. Args: draw: PIL ImageDraw object connections: List of Connection objects positions: Dictionary mapping card_id to Position offset_x: X offset for coordinate transformation offset_y: Y offset for coordinate transformation """ for conn in connections: source_x = conn.source_point.x + offset_x source_y = conn.source_point.y + offset_y target_x = conn.target_point.x + offset_x target_y = conn.target_point.y + offset_y # Draw arrow line draw.line([(source_x, source_y), (target_x, target_y)], fill=self.arrow_color, width=self.arrow_width) # Draw arrowhead at target end self._draw_arrowhead_png(draw, source_x, source_y, target_x, target_y) def _draw_arrowhead_png(self, draw, x1, y1, x2, y2): """ Draw an arrowhead at the end of a line. Args: draw: PIL ImageDraw object x1, y1: Start point of line x2, y2: End point of line (where arrowhead is drawn) """ arrow_size = 10 # Calculate direction vector dx = x2 - x1 dy = y2 - y1 length = math.sqrt(dx*dx + dy*dy) if length < 1: return # Normalize direction dx /= length dy /= length # Calculate perpendicular vector perp_x = -dy perp_y = dx # Calculate arrowhead triangle points point1_x = x2 - arrow_size * dx + arrow_size * 0.5 * perp_x point1_y = y2 - arrow_size * dy + arrow_size * 0.5 * perp_y point2_x = x2 - arrow_size * dx - arrow_size * 0.5 * perp_x point2_y = y2 - arrow_size * dy - arrow_size * 0.5 * perp_y # Draw filled triangle draw.polygon([(x2, y2), (point1_x, point1_y), (point2_x, point2_y)], fill=self.arrow_color) def _render_cards_svg(self, dwg, cards, positions, offset_x, offset_y): """ Render all CRC cards to SVG. Args: dwg: svgwrite Drawing object cards: List of CRCCard objects positions: Dictionary mapping card_id to Position offset_x: X offset for coordinate transformation offset_y: Y offset for coordinate transformation """ for card in cards: pos = positions.get(card.card_id) if not pos: continue # Calculate card rectangle coordinates x = pos.x + offset_x y = pos.y + offset_y # Create group for this card card_group = dwg.g() # Draw card background and border card_group.add(dwg.rect( insert=(x, y), size=(self.card_width, self.card_height), fill=f"rgb{self.card_fill_color}", stroke=f"rgb{self.card_border_color}", stroke_width=self.card_border_width )) # Draw component name name_y = y + 20 card_group.add(dwg.text( card.name, insert=(x + 10, name_y), fill=f"rgb{self.text_color}", font_size="14px", font_weight="bold" )) # Draw horizontal line line_y = name_y + 10 card_group.add(dwg.line( start=(x + 5, line_y), end=(x + self.card_width - 5, line_y), stroke=f"rgb{self.card_border_color}", stroke_width=1 )) # Draw responsibilities section resp_y = line_y + 15 card_group.add(dwg.text( "Responsibilities:", insert=(x + 10, resp_y), fill=f"rgb{self.text_color}", font_size="10px" )) current_y = resp_y + 15 for resp in card.responsibilities[:3]: text = resp.description[:25] + "..." if len(resp.description) > 25 else resp.description card_group.add(dwg.text( f"- {text}", insert=(x + 15, current_y), fill=f"rgb{self.text_color}", font_size="9px" )) current_y += 12 # Draw collaborators section collab_y = y + self.card_height - 40 card_group.add(dwg.text( "Collaborators:", insert=(x + 10, collab_y), fill=f"rgb{self.text_color}", font_size="10px" )) current_y = collab_y + 15 for collab in card.collaborations[:2]: text = collab.collaborator_name[:20] + "..." if len(collab.collaborator_name) > 20 else collab.collaborator_name card_group.add(dwg.text( f"- {text}", insert=(x + 15, current_y), fill=f"rgb{self.text_color}", font_size="9px" )) current_y += 12 dwg.add(card_group) def _render_connections_svg(self, dwg, connections, positions, offset_x, offset_y): """ Render all connection arrows to SVG. Args: dwg: svgwrite Drawing object connections: List of Connection objects positions: Dictionary mapping card_id to Position offset_x: X offset for coordinate transformation offset_y: Y offset for coordinate transformation """ for conn in connections: source_x = conn.source_point.x + offset_x source_y = conn.source_point.y + offset_y target_x = conn.target_point.x + offset_x target_y = conn.target_point.y + offset_y # Draw arrow line dwg.add(dwg.line( start=(source_x, source_y), end=(target_x, target_y), stroke=f"rgb{self.arrow_color}", stroke_width=self.arrow_width )) # Draw arrowhead self._draw_arrowhead_svg(dwg, source_x, source_y, target_x, target_y) def _draw_arrowhead_svg(self, dwg, x1, y1, x2, y2): """ Draw an arrowhead at the end of a line in SVG. Args: dwg: svgwrite Drawing object x1, y1: Start point of line x2, y2: End point of line (where arrowhead is drawn) """ arrow_size = 10 # Calculate direction vector dx = x2 - x1 dy = y2 - y1 length = math.sqrt(dx*dx + dy*dy) if length < 1: return # Normalize direction dx /= length dy /= length # Calculate perpendicular vector perp_x = -dy perp_y = dx # Calculate arrowhead triangle points point1_x = x2 - arrow_size * dx + arrow_size * 0.5 * perp_x point1_y = y2 - arrow_size * dy + arrow_size * 0.5 * perp_y point2_x = x2 - arrow_size * dx - arrow_size * 0.5 * perp_x point2_y = y2 - arrow_size * dy - arrow_size * 0.5 * perp_y # Draw filled triangle dwg.add(dwg.polygon( points=[(x2, y2), (point1_x, point1_y), (point2_x, point2_y)], fill=f"rgb{self.arrow_color}" ))# =========================================================================# HARDWARE DETECTION# =========================================================================class EnhancedHardwareDetector: """ Enhanced hardware detection with detailed GPU information. Detects available GPU hardware across NVIDIA CUDA, AMD ROCm, Apple MPS, and Intel platforms. Provides detailed information about detected hardware to enable optimal configuration. """ def __init__(self): """Initialize the hardware detector.""" self.system = platform.system() self.machine = platform.machine() self.gpu_info = None def detect_gpu_backend(self): """ Detect the best available GPU backend for the current system. Returns: Tuple of (backend_name, gpu_info_dict) where backend_name is one of "cuda", "rocm", "mps", "intel", "cpu" and gpu_info_dict contains detailed hardware information """ # Try CUDA first (NVIDIA) cuda_info = self._detect_cuda() if cuda_info: self.gpu_info = cuda_info return "cuda", cuda_info # Try ROCm (AMD) rocm_info = self._detect_rocm() if rocm_info: self.gpu_info = rocm_info return "rocm", rocm_info # Try MPS (Apple) mps_info = self._detect_mps() if mps_info: self.gpu_info = mps_info return "mps", mps_info # Try Intel GPU intel_info = self._detect_intel() if intel_info: self.gpu_info = intel_info return "intel", intel_info # Fall back to CPU cpu_info = self._detect_cpu() self.gpu_info = cpu_info return "cpu", cpu_info def _detect_cuda(self): """ Detect NVIDIA CUDA GPUs and return detailed information. Returns: Dictionary with GPU information or None if not available """ try: # Check if nvidia-smi is available result = subprocess.run( ['nvidia-smi', '--query-gpu=name,memory.total,driver_version', '--format=csv,noheader'], capture_output=True, text=True, timeout=5 ) if result.returncode == 0: lines = result.stdout.strip().split('\n') gpus = [] for line in lines: parts = [p.strip() for p in line.split(',')] if len(parts) >= 3: gpus.append({ 'name': parts[0], 'memory': parts[1], 'driver': parts[2] }) # Also check PyTorch CUDA availability cuda_available = False cuda_version = None try: import torch cuda_available = torch.cuda.is_available() if cuda_available: cuda_version = torch.version.cuda except ImportError: pass return { 'backend': 'cuda', 'vendor': 'NVIDIA', 'gpus': gpus, 'pytorch_available': cuda_available, 'cuda_version': cuda_version, 'device_count': len(gpus) } except (FileNotFoundError, subprocess.TimeoutExpired): pass return None def _detect_rocm(self): """ Detect AMD ROCm GPUs and return detailed information. Returns: Dictionary with GPU information or None if not available """ if self.system != "Linux": return None try: # Check if rocm-smi is available result = subprocess.run( ['rocm-smi', '--showproductname'], capture_output=True, text=True, timeout=5 ) if result.returncode == 0: # Parse ROCm SMI output gpus = [] lines = result.stdout.strip().split('\n') for line in lines: if 'GPU' in line and 'Card series' in line: parts = line.split(':') if len(parts) >= 2: gpus.append({ 'name': parts[1].strip() }) # Check PyTorch ROCm availability rocm_available = False rocm_version = None try: import torch rocm_available = torch.cuda.is_available() if rocm_available and hasattr(torch.version, 'hip'): rocm_version = torch.version.hip except ImportError: pass return { 'backend': 'rocm', 'vendor': 'AMD', 'gpus': gpus if gpus else [{'name': 'AMD GPU'}], 'pytorch_available': rocm_available, 'rocm_version': rocm_version, 'device_count': len(gpus) if gpus else 1 } except (FileNotFoundError, subprocess.TimeoutExpired): pass return None def _detect_mps(self): """ Detect Apple Metal Performance Shaders support. Returns: Dictionary with GPU information or None if not available """ if self.system != "Darwin": return None if self.machine != "arm64": return None try: import torch if torch.backends.mps.is_available(): # Get macOS version macos_version = platform.mac_ver()[0] # Get chip information chip_name = "Apple Silicon" try: result = subprocess.run( ['sysctl', '-n', 'machdep.cpu.brand_string'], capture_output=True, text=True, timeout=2 ) if result.returncode == 0: chip_name = result.stdout.strip() except: pass return { 'backend': 'mps', 'vendor': 'Apple', 'gpus': [{'name': chip_name}], 'pytorch_available': True, 'macos_version': macos_version, 'device_count': 1 } except ImportError: pass return None def _detect_intel(self): """ Detect Intel GPU support. Returns: Dictionary with GPU information or None if not available """ try: # Check for Intel Extension for PyTorch import intel_extension_for_pytorch as ipex import torch # Intel extension is available xpu_available = hasattr(torch, 'xpu') and torch.xpu.is_available() if xpu_available: device_count = torch.xpu.device_count() gpus = [] for i in range(device_count): try: name = torch.xpu.get_device_name(i) gpus.append({'name': name}) except: gpus.append({'name': f'Intel GPU {i}'}) return { 'backend': 'intel', 'vendor': 'Intel', 'gpus': gpus, 'pytorch_available': True, 'ipex_version': ipex.__version__, 'device_count': device_count } except ImportError: pass # Check for OpenCL as fallback try: import pyopencl as cl platforms = cl.get_platforms() intel_devices = [] for platform in platforms: if 'Intel' in platform.name: devices = platform.get_devices() for device in devices: if device.type == cl.device_type.GPU: intel_devices.append({ 'name': device.name.strip() }) if intel_devices: return { 'backend': 'opencl', 'vendor': 'Intel', 'gpus': intel_devices, 'pytorch_available': False, 'opencl_available': True, 'device_count': len(intel_devices) } except ImportError: pass return None def _detect_cpu(self): """ Get CPU information as fallback. Returns: Dictionary with CPU information """ cpu_name = platform.processor() if not cpu_name: cpu_name = "Unknown CPU" # Try to get more detailed CPU info try: if self.system == "Darwin": result = subprocess.run( ['sysctl', '-n', 'machdep.cpu.brand_string'], capture_output=True, text=True, timeout=2 ) if result.returncode == 0: cpu_name = result.stdout.strip() elif self.system == "Linux": with open('/proc/cpuinfo', 'r') as f: for line in f: if 'model name' in line: cpu_name = line.split(':')[1].strip() break except: pass return { 'backend': 'cpu', 'vendor': 'CPU', 'name': cpu_name, 'pytorch_available': False, 'device_count': 1 } def print_hardware_info(self): """Print detailed hardware information to console.""" backend, info = self.detect_gpu_backend() print("=" * 70) print("HARDWARE DETECTION RESULTS") print("=" * 70) print(f"System: {self.system}") print(f"Architecture: {self.machine}") print(f"Selected Backend: {backend.upper()}") print(f"Vendor: {info.get('vendor', 'Unknown')}") print() if 'gpus' in info: print(f"Detected {info['device_count']} GPU(s):") for i, gpu in enumerate(info['gpus']): print(f" GPU {i}: {gpu.get('name', 'Unknown')}") if 'memory' in gpu: print(f" Memory: {gpu['memory']}") if 'driver' in gpu: print(f" Driver: {gpu['driver']}") elif 'name' in info: print(f"Device: {info['name']}") print() print(f"PyTorch Available: {info.get('pytorch_available', False)}") if info.get('cuda_version'): print(f"CUDA Version: {info['cuda_version']}") if info.get('rocm_version'): print(f"ROCm Version: {info['rocm_version']}") if info.get('macos_version'): print(f"macOS Version: {info['macos_version']}") if info.get('ipex_version'): print(f"Intel Extension Version: {info['ipex_version']}") if info.get('opencl_available'): print(f"OpenCL Available: True") print("=" * 70) print()# =========================================================================# LLM PROVIDERS# =========================================================================class LLMProvider(ABC): """ Abstract base class for LLM providers. Defines the interface that all LLM providers (local and remote) must implement. """ @abstractmethod def initialize(self, **kwargs): """ Initialize the LLM provider with configuration parameters. Args: **kwargs: Provider-specific configuration """ pass @abstractmethod def generate(self, prompt, system_prompt=None): """ Generate a response from the LLM. Args: prompt: User input text system_prompt: Optional system instructions Returns: Generated text response """ pass @abstractmethod def cleanup(self): """Release resources and cleanup.""" passclass MockLLMProvider(LLMProvider): """ Mock LLM provider for testing and demonstration. Provides predefined responses based on keywords in the input, allowing the tool to be tested without requiring an actual LLM. """ def initialize(self, **kwargs): """Initialize the mock provider.""" print("Using Mock LLM Provider (for demonstration)") print("This provider uses keyword matching instead of actual AI") print() def generate(self, prompt, system_prompt=None): """ Generate a mock response based on keywords in prompt. Args: prompt: User input text system_prompt: Ignored for mock provider Returns: JSON string containing a command """ prompt_lower = prompt.lower() # Detect create/add commands if "create" in prompt_lower or "add" in prompt_lower or "new" in prompt_lower: if "user" in prompt_lower and "manager" in prompt_lower: return '''{ "command_type": "CREATE_COMPONENT", "parameters": { "name": "UserManager", "responsibilities": ["Manage user accounts", "Authenticate users"], "collaborators": ["Database", "AuthService"] } }''' elif "database" in prompt_lower: return '''{ "command_type": "CREATE_COMPONENT", "parameters": { "name": "Database", "responsibilities": ["Store data", "Execute queries"], "collaborators": [] } }''' elif "auth" in prompt_lower: return '''{ "command_type": "CREATE_COMPONENT", "parameters": { "name": "AuthService", "responsibilities": ["Validate credentials", "Generate tokens"], "collaborators": ["Database"] } }''' # Detect snapshot/save commands elif "snapshot" in prompt_lower or "save" in prompt_lower: return '''{ "command_type": "SNAPSHOT", "parameters": {} }''' # Detect finish/done commands elif "finish" in prompt_lower or "done" in prompt_lower or "complete" in prompt_lower: return '''{ "command_type": "FINISH", "parameters": {} }''' # Default response return '''{ "command_type": "CREATE_COMPONENT", "parameters": { "name": "Component", "responsibilities": ["Do something"], "collaborators": [] } }''' def cleanup(self): """No cleanup needed for mock provider.""" passclass LocalTransformersProvider(LLMProvider): """ Local LLM provider using HuggingFace Transformers with full GPU support. Automatically detects and configures for CUDA, ROCm, MPS, Intel, or CPU. Supports quantization for memory-constrained environments. """ def __init__(self): """Initialize the local transformers provider.""" self.model = None self.tokenizer = None self.device = None self.backend = None self.gpu_info = None self.model_name = None def initialize(self, model_name="mistralai/Mistral-7B-Instruct-v0.2", temperature=0.7, max_tokens=2048, load_in_8bit=False, load_in_4bit=False, trust_remote_code=False, **kwargs): """ Initialize the local LLM with automatic hardware detection. Args: model_name: HuggingFace model identifier temperature: Sampling temperature for generation max_tokens: Maximum number of tokens to generate load_in_8bit: Use 8-bit quantization (requires bitsandbytes) load_in_4bit: Use 4-bit quantization (requires bitsandbytes) trust_remote_code: Allow custom model code execution **kwargs: Additional parameters passed to model loading """ print("Initializing Local LLM Provider...") print(f"Model: {model_name}") print() # Detect hardware detector = EnhancedHardwareDetector() self.backend, self.gpu_info = detector.detect_gpu_backend() detector.print_hardware_info() # Import required libraries import torch from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig self.model_name = model_name self.temperature = temperature self.max_tokens = max_tokens # Configure device and dtype based on detected backend device_map = None torch_dtype = torch.float32 quantization_config = None if self.backend == "cuda": print("Configuring for NVIDIA CUDA...") self.device = torch.device("cuda") torch_dtype = torch.float16 device_map = "auto" # Configure quantization if requested if load_in_8bit or load_in_4bit: try: quantization_config = BitsAndBytesConfig( load_in_8bit=load_in_8bit, load_in_4bit=load_in_4bit, bnb_4bit_compute_dtype=torch.float16 if load_in_4bit else None, bnb_4bit_use_double_quant=True if load_in_4bit else None, bnb_4bit_quant_type="nf4" if load_in_4bit else None ) print(f"Using {'8-bit' if load_in_8bit else '4-bit'} quantization") except Exception as e: print(f"Warning: Could not configure quantization: {e}") print("Falling back to float16") elif self.backend == "rocm": print("Configuring for AMD ROCm...") self.device = torch.device("cuda") # ROCm uses CUDA API torch_dtype = torch.float16 device_map = "auto" # ROCm may have limited quantization support if load_in_8bit or load_in_4bit: print("Warning: Quantization support on ROCm may be limited") print("Attempting to use quantization...") try: quantization_config = BitsAndBytesConfig( load_in_8bit=load_in_8bit, load_in_4bit=load_in_4bit ) except Exception as e: print(f"Quantization failed: {e}") print("Falling back to float16") elif self.backend == "mps": print("Configuring for Apple Metal Performance Shaders...") self.device = torch.device("mps") torch_dtype = torch.float32 # MPS works better with float32 device_map = None # MPS does not support quantization if load_in_8bit or load_in_4bit: print("Warning: MPS does not support quantization") print("Using float32 instead") elif self.backend == "intel": print("Configuring for Intel GPU...") try: import intel_extension_for_pytorch as ipex self.device = torch.device("xpu") torch_dtype = torch.float16 device_map = None except ImportError: print("Intel Extension not available, falling back to CPU") self.device = torch.device("cpu") torch_dtype = torch.float32 device_map = None else: print("Configuring for CPU...") self.device = torch.device("cpu") torch_dtype = torch.float32 device_map = None print(f"Device: {self.device}") print(f"Data Type: {torch_dtype}") print() # Load tokenizer print("Loading tokenizer...") self.tokenizer = AutoTokenizer.from_pretrained( model_name, trust_remote_code=trust_remote_code ) # Set pad token if not present if self.tokenizer.pad_token is None: self.tokenizer.pad_token = self.tokenizer.eos_token print("Tokenizer loaded successfully") print() # Load model print("Loading model (this may take several minutes)...") try: self.model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch_dtype, device_map=device_map, quantization_config=quantization_config, low_cpu_mem_usage=True, trust_remote_code=trust_remote_code, **kwargs ) # Move model to device if not using device_map if device_map is None: self.model.to(self.device) # Apply Intel optimizations if available if self.backend == "intel": try: import intel_extension_for_pytorch as ipex self.model = ipex.optimize(self.model) print("Applied Intel optimizations") except: pass print("Model loaded successfully") print() # Print memory usage if on GPU if self.backend in ["cuda", "rocm"]: allocated = torch.cuda.memory_allocated() / 1024**3 reserved = torch.cuda.memory_reserved() / 1024**3 print(f"GPU Memory Allocated: {allocated:.2f} GB") print(f"GPU Memory Reserved: {reserved:.2f} GB") print() except Exception as e: print(f"Error loading model: {e}") raise def generate(self, prompt, system_prompt=None): """ Generate a response from the local LLM. Args: prompt: User input text system_prompt: Optional system instructions Returns: Generated text response """ import torch # Construct full prompt if system_prompt: full_prompt = f"{system_prompt}\n\nUser: {prompt}\n\nAssistant:" else: full_prompt = prompt # Tokenize input inputs = self.tokenizer( full_prompt, return_tensors="pt", padding=True, truncation=True, max_length=4096 ) # Move inputs to device inputs = {k: v.to(self.device) for k, v in inputs.items()} # Generate response with torch.no_grad(): outputs = self.model.generate( **inputs, max_new_tokens=self.max_tokens, temperature=self.temperature, do_sample=True, top_p=0.95, top_k=50, pad_token_id=self.tokenizer.pad_token_id, eos_token_id=self.tokenizer.eos_token_id ) # Decode generated text generated_text = self.tokenizer.decode( outputs[0], skip_special_tokens=True ) # Extract only the new generated portion response = generated_text[len(full_prompt):].strip() return response def cleanup(self): """Release GPU memory and cleanup resources.""" import torch print("Cleaning up LLM resources...") if self.model is not None: del self.model self.model = None if self.tokenizer is not None: del self.tokenizer self.tokenizer = None # Clear GPU cache if self.backend in ["cuda", "rocm"]: torch.cuda.empty_cache() print("GPU cache cleared") elif self.backend == "mps": torch.mps.empty_cache() print("MPS cache cleared") elif self.backend == "intel": try: torch.xpu.empty_cache() print("Intel XPU cache cleared") except: pass print("Cleanup complete")class OpenAIProvider(LLMProvider): """ Remote LLM provider for OpenAI API (GPT-3.5, GPT-4, etc.). Supports both the legacy Completion API and the Chat Completion API. """ def __init__(self): """Initialize the OpenAI provider.""" self.api_key = None self.model_name = None self.temperature = None self.max_tokens = None self.use_chat_api = True self.base_url = "https://api.openai.com/v1" def initialize(self, api_key=None, model_name="gpt-4", temperature=0.7, max_tokens=2048, use_chat_api=True, base_url=None, **kwargs): """ Initialize the OpenAI provider. Args: api_key: OpenAI API key (or set OPENAI_API_KEY environment variable) model_name: Model identifier (gpt-4, gpt-3.5-turbo, etc.) temperature: Sampling temperature max_tokens: Maximum tokens to generate use_chat_api: Use Chat Completion API (True) or legacy Completion API (False) base_url: Custom API base URL (for OpenAI-compatible APIs) """ print("Initializing OpenAI Provider...") # Get API key from parameter or environment self.api_key = api_key or os.getenv("OPENAI_API_KEY") if not self.api_key: raise ValueError( "OpenAI API key must be provided either as a parameter " "or via the OPENAI_API_KEY environment variable" ) self.model_name = model_name self.temperature = temperature self.max_tokens = max_tokens self.use_chat_api = use_chat_api if base_url: self.base_url = base_url print(f"Model: {model_name}") print(f"API Type: {'Chat Completion' if use_chat_api else 'Completion'}") print(f"Base URL: {self.base_url}") print("OpenAI Provider initialized successfully") print() def generate(self, prompt, system_prompt=None): """ Generate a response using the OpenAI API. Args: prompt: User input text system_prompt: Optional system instructions Returns: Generated text response """ import requests import time headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } if self.use_chat_api: # Use Chat Completion API endpoint = f"{self.base_url}/chat/completions" messages = [] if system_prompt: messages.append({"role": "system", "content": system_prompt}) messages.append({"role": "user", "content": prompt}) data = { "model": self.model_name, "messages": messages, "temperature": self.temperature, "max_tokens": self.max_tokens } else: # Use legacy Completion API endpoint = f"{self.base_url}/completions" full_prompt = prompt if system_prompt: full_prompt = f"{system_prompt}\n\n{prompt}" data = { "model": self.model_name, "prompt": full_prompt, "temperature": self.temperature, "max_tokens": self.max_tokens } # Make API request with retry logic max_retries = 3 retry_delay = 1 for attempt in range(max_retries): try: response = requests.post( endpoint, headers=headers, json=data, timeout=60 ) response.raise_for_status() result = response.json() if self.use_chat_api: return result["choices"][0]["message"]["content"] else: return result["choices"][0]["text"] except requests.exceptions.RequestException as e: if attempt < max_retries - 1: print(f"Request failed (attempt {attempt + 1}/{max_retries}): {e}") print(f"Retrying in {retry_delay} seconds...") time.sleep(retry_delay) retry_delay *= 2 else: raise Exception(f"OpenAI API request failed after {max_retries} attempts: {e}") def cleanup(self): """No cleanup needed for API-based provider.""" passclass AnthropicProvider(LLMProvider): """ Remote LLM provider for Anthropic Claude API. Supports Claude 3 (Opus, Sonnet, Haiku) and Claude 2 models. """ def __init__(self): """Initialize the Anthropic provider.""" self.api_key = None self.model_name = None self.temperature = None self.max_tokens = None self.base_url = "https://api.anthropic.com/v1" def initialize(self, api_key=None, model_name="claude-3-sonnet-20240229", temperature=0.7, max_tokens=2048, **kwargs): """ Initialize the Anthropic provider. Args: api_key: Anthropic API key (or set ANTHROPIC_API_KEY environment variable) model_name: Model identifier (claude-3-opus, claude-3-sonnet, etc.) temperature: Sampling temperature max_tokens: Maximum tokens to generate """ print("Initializing Anthropic Provider...") self.api_key = api_key or os.getenv("ANTHROPIC_API_KEY") if not self.api_key: raise ValueError( "Anthropic API key must be provided either as a parameter " "or via the ANTHROPIC_API_KEY environment variable" ) self.model_name = model_name self.temperature = temperature self.max_tokens = max_tokens print(f"Model: {model_name}") print("Anthropic Provider initialized successfully") print() def generate(self, prompt, system_prompt=None): """ Generate a response using the Anthropic API. Args: prompt: User input text system_prompt: Optional system instructions Returns: Generated text response """ import requests import time headers = { "x-api-key": self.api_key, "anthropic-version": "2023-06-01", "Content-Type": "application/json" } messages = [{"role": "user", "content": prompt}] data = { "model": self.model_name, "messages": messages, "temperature": self.temperature, "max_tokens": self.max_tokens } if system_prompt: data["system"] = system_prompt # Make API request with retry logic max_retries = 3 retry_delay = 1 for attempt in range(max_retries): try: response = requests.post( f"{self.base_url}/messages", headers=headers, json=data, timeout=60 ) response.raise_for_status() result = response.json() return result["content"][0]["text"] except requests.exceptions.RequestException as e: if attempt < max_retries - 1: print(f"Request failed (attempt {attempt + 1}/{max_retries}): {e}") print(f"Retrying in {retry_delay} seconds...") time.sleep(retry_delay) retry_delay *= 2 else: raise Exception(f"Anthropic API request failed after {max_retries} attempts: {e}") def cleanup(self): """No cleanup needed for API-based provider.""" passclass GeminiProvider(LLMProvider): """ Remote LLM provider for Google Gemini API. Supports Gemini Pro and Gemini Pro Vision models. """ def __init__(self): """Initialize the Gemini provider.""" self.api_key = None self.model_name = None self.temperature = None self.max_tokens = None self.base_url = "https://generativelanguage.googleapis.com/v1beta" def initialize(self, api_key=None, model_name="gemini-pro", temperature=0.7, max_tokens=2048, **kwargs): """ Initialize the Gemini provider. Args: api_key: Google API key (or set GOOGLE_API_KEY environment variable) model_name: Model identifier (gemini-pro, gemini-pro-vision) temperature: Sampling temperature max_tokens: Maximum tokens to generate """ print("Initializing Google Gemini Provider...") self.api_key = api_key or os.getenv("GOOGLE_API_KEY") if not self.api_key: raise ValueError( "Google API key must be provided either as a parameter " "or via the GOOGLE_API_KEY environment variable" ) self.model_name = model_name self.temperature = temperature self.max_tokens = max_tokens print(f"Model: {model_name}") print("Gemini Provider initialized successfully") print() def generate(self, prompt, system_prompt=None): """ Generate a response using the Gemini API. Args: prompt: User input text system_prompt: Optional system instructions Returns: Generated text response """ import requests import time # Construct full prompt with system instructions if provided full_prompt = prompt if system_prompt: full_prompt = f"{system_prompt}\n\n{prompt}" endpoint = f"{self.base_url}/models/{self.model_name}:generateContent" data = { "contents": [{ "parts": [{"text": full_prompt}] }], "generationConfig": { "temperature": self.temperature, "maxOutputTokens": self.max_tokens } } # Make API request with retry logic max_retries = 3 retry_delay = 1 for attempt in range(max_retries): try: response = requests.post( f"{endpoint}?key={self.api_key}", json=data, timeout=60 ) response.raise_for_status() result = response.json() return result["candidates"][0]["content"]["parts"][0]["text"] except requests.exceptions.RequestException as e: if attempt < max_retries - 1: print(f"Request failed (attempt {attempt + 1}/{max_retries}): {e}") print(f"Retrying in {retry_delay} seconds...") time.sleep(retry_delay) retry_delay *= 2 else: raise Exception(f"Gemini API request failed after {max_retries} attempts: {e}") def cleanup(self): """No cleanup needed for API-based provider.""" pass# =========================================================================# LLM MANAGER# =========================================================================class EnhancedLLMManager: """ Enhanced LLM manager supporting multiple local and remote providers. Provides a unified interface for all LLM providers and automatically handles hardware detection and optimal configuration. """ def __init__(self): """Initialize the LLM manager.""" self.provider = None self.provider_type = None def setup(self, provider_type="local", **kwargs): """ Create and initialize the appropriate LLM provider. Args: provider_type: One of "local", "openai", "anthropic", "gemini", "mock" **kwargs: Provider-specific configuration parameters """ print("=" * 70) print("LLM MANAGER SETUP") print("=" * 70) print(f"Provider Type: {provider_type}") print() self.provider_type = provider_type if provider_type == "local": self.provider = LocalTransformersProvider() elif provider_type == "openai": self.provider = OpenAIProvider() elif provider_type == "anthropic": self.provider = AnthropicProvider() elif provider_type == "gemini": self.provider = GeminiProvider() elif provider_type == "mock": self.provider = MockLLMProvider() else: raise ValueError(f"Unknown provider type: {provider_type}") self.provider.initialize(**kwargs) print("=" * 70) print("LLM MANAGER READY") print("=" * 70) print() def query(self, prompt, system_prompt=None): """ Query the LLM provider. Args: prompt: User input text system_prompt: Optional system instructions Returns: Generated text response """ if self.provider is None: raise RuntimeError("LLM provider not initialized. Call setup() first.") return self.provider.generate(prompt, system_prompt) def shutdown(self): """Shutdown the LLM provider and release resources.""" if self.provider is not None: self.provider.cleanup() self.provider = None# =========================================================================# COMMAND PARSER# =========================================================================class Command: """ Represents a parsed command extracted from user input. Commands have a type (CREATE_COMPONENT, MODIFY_COMPONENT, etc.) and parameters specific to that command type. """ def __init__(self, command_type, parameters): """ Initialize a command. Args: command_type: Type of command parameters: Dictionary of command parameters """ self.command_type = command_type self.parameters = parameters @staticmethod def from_dict(data): """ Deserialize from dictionary. Args: data: Dictionary with command_type and parameters Returns: Command object """ return Command(data["command_type"], data.get("parameters", {})) def to_dict(self): """ Serialize to dictionary. Returns: Dictionary representation """ return { "command_type": self.command_type, "parameters": self.parameters } def __repr__(self): """String representation for debugging.""" return f"Command(type='{self.command_type}', params={self.parameters})"class CommandParser: """ Parses natural language input into structured commands. Uses the LLM to analyze user input and extract structured commands that the system can execute. """ def __init__(self, llm_manager): """ Initialize the command parser. Args: llm_manager: EnhancedLLMManager instance """ self.llm = llm_manager self.system_prompt = self._build_system_prompt() def _build_system_prompt(self): """ Build the system prompt for command parsing. Returns: System prompt string """ return """You are a command parser for a CRC card diagramming tool.Your task is to analyze user input and extract structured commands.Available command types:CREATE_COMPONENT: Create a new CRC cardMODIFY_COMPONENT: Modify an existing CRC cardDELETE_COMPONENT: Remove a CRC cardSNAPSHOT: Save the current diagramRESTORE: Load a saved diagramFINISH: User indicates the diagram is completeFor each user input, respond with a JSON object containing: { "command_type": "CREATE_COMPONENT|MODIFY_COMPONENT|DELETE_COMPONENT|SNAPSHOT|RESTORE|FINISH", "parameters": { // command-specific parameters } }CREATE_COMPONENT parameters:name: component name (string)responsibilities: list of responsibility descriptions (list of strings)collaborators: list of collaborator names (list of strings)MODIFY_COMPONENT parameters:name: component name to modify (string)add_responsibilities: responsibilities to add (list of strings, optional)remove_responsibilities: responsibilities to remove (list of strings, optional)add_collaborators: collaborators to add (list of strings, optional)remove_collaborators: collaborators to remove (list of strings, optional)DELETE_COMPONENT parameters:name: component name to delete (string)SNAPSHOT parameters:filename: base filename for saved files (string, optional)RESTORE parameters:filename: JSON file to restore from (string)FINISH parameters: noneRespond ONLY with the JSON object, no additional text.""" def parse(self, user_input): """ Parse user input into a structured command. Args: user_input: Natural language input from user Returns: Command object or None if parsing fails """ try: # Query the LLM with the user input response = self.llm.query(user_input, self.system_prompt) # Extract JSON from response json_str = self._extract_json(response) # Parse JSON command_data = json.loads(json_str) # Create and return Command object return Command.from_dict(command_data) except Exception as e: print(f"Error parsing command: {e}") return None def _extract_json(self, text): """ Extract JSON object from text that may contain additional content. Args: text: Text containing JSON Returns: Extracted JSON string """ # Look for content between first { and last } start = text.find('{') end = text.rfind('}') if start == -1 or end == -1: raise ValueError("No JSON object found in response") return text[start:end+1]# =========================================================================# STATE MANAGEMENT# =========================================================================class DiagramState: """ Represents the complete state of a CRC card diagram. Includes all cards, their positions, connections, and metadata. Supports serialization to JSON for persistence. """ def __init__(self): """Initialize an empty diagram state.""" self.cards = [] self.positions = {} self.connections = [] self.metadata = { "version": "1.0", "created": datetime.now().isoformat(), "modified": datetime.now().isoformat() } def add_card(self, card): """ Add a new CRC card to the diagram. Args: card: CRCCard object to add """ if card.card_id not in [c.card_id for c in self.cards]: self.cards.append(card) self._update_modified() def remove_card(self, card_id): """ Remove a card and all connections involving it. Args: card_id: ID of the card to remove """ self.cards = [c for c in self.cards if c.card_id != card_id] self.connections = [conn for conn in self.connections if conn.source_id != card_id and conn.target_id != card_id] if card_id in self.positions: del self.positions[card_id] self._update_modified() def get_card_by_name(self, name): """ Find a card by its name. Args: name: Name of the card to find Returns: CRCCard object or None if not found """ for card in self.cards: if card.name == name: return card return None def get_card_by_id(self, card_id): """ Find a card by its ID. Args: card_id: ID of the card to find Returns: CRCCard object or None if not found """ for card in self.cards: if card.card_id == card_id: return card return None def update_layout(self, positions, connections): """ Update the layout information. Args: positions: Dictionary mapping card_id to Position connections: List of Connection objects """ self.positions = positions self.connections = connections self._update_modified() def _update_modified(self): """Update the modified timestamp.""" self.metadata["modified"] = datetime.now().isoformat() def to_dict(self): """ Serialize the entire state to a dictionary. Returns: Dictionary representation """ return { "metadata": self.metadata, "cards": [card.to_dict() for card in self.cards], "positions": {card_id: pos.to_dict() for card_id, pos in self.positions.items()}, "connections": [conn.to_dict() for conn in self.connections] } def to_json(self): """ Serialize to JSON string. Returns: JSON string representation """ return json.dumps(self.to_dict(), indent=2) @staticmethod def from_dict(data): """ Deserialize from dictionary. Args: data: Dictionary representation Returns: DiagramState object """ state = DiagramState() state.metadata = data.get("metadata", state.metadata) state.cards = [CRCCard.from_dict(c) for c in data.get("cards", [])] state.positions = {card_id: Position.from_dict(pos) for card_id, pos in data.get("positions", {}).items()} state.connections = [Connection.from_dict(c) for c in data.get("connections", [])] return state @staticmethod def from_json(json_str): """ Deserialize from JSON string. Args: json_str: JSON string representation Returns: DiagramState object """ data = json.loads(json_str) return DiagramState.from_dict(data)class StateManager: """ Manages diagram state and executes commands. Coordinates between the domain model, layout engine, and rendering engine to maintain consistent state and produce visual outputs. """ def __init__(self, layout_engine, rendering_engine): """ Initialize the state manager. Args: layout_engine: LayoutEngine instance rendering_engine: RenderingEngine instance """ self.state = DiagramState() self.layout_engine = layout_engine self.rendering_engine = rendering_engine def execute_command(self, command): """ Execute a command and update the state. Args: command: Command object to execute Returns: Result string ("SUCCESS", "FINISH", etc.) """ if command.command_type == "CREATE_COMPONENT": self._execute_create(command.parameters) elif command.command_type == "MODIFY_COMPONENT": self._execute_modify(command.parameters) elif command.command_type == "DELETE_COMPONENT": self._execute_delete(command.parameters) elif command.command_type == "SNAPSHOT": self._execute_snapshot(command.parameters) elif command.command_type == "RESTORE": self._execute_restore(command.parameters) elif command.command_type == "FINISH": return "FINISH" # Recompute layout after any modification if command.command_type in ["CREATE_COMPONENT", "MODIFY_COMPONENT", "DELETE_COMPONENT"]: self._recompute_layout() return "SUCCESS" def _execute_create(self, params): """ Create a new CRC card. Args: params: Dictionary with name, responsibilities, collaborators """ name = params.get("name") responsibilities = params.get("responsibilities", []) collaborators = params.get("collaborators", []) # Generate unique ID card_id = str(uuid.uuid4()) # Create card card = CRCCard(card_id, name) # Add responsibilities for resp_desc in responsibilities: card.add_responsibility(Responsibility(resp_desc)) # Add collaborations (resolve collaborator IDs) for collab_name in collaborators: # Try to find existing card with this name existing_card = self.state.get_card_by_name(collab_name) if existing_card: collab_id = existing_card.card_id else: # Create placeholder ID for not-yet-created collaborator collab_id = f"placeholder_{collab_name}" card.add_collaboration(Collaboration(collab_id, collab_name)) self.state.add_card(card) def _execute_modify(self, params): """ Modify an existing CRC card. Args: params: Dictionary with modification parameters """ name = params.get("name") card = self.state.get_card_by_name(name) if not card: print(f"Card '{name}' not found") return # Add responsibilities for resp_desc in params.get("add_responsibilities", []): card.add_responsibility(Responsibility(resp_desc)) # Remove responsibilities for resp_desc in params.get("remove_responsibilities", []): card.remove_responsibility(Responsibility(resp_desc)) # Add collaborations for collab_name in params.get("add_collaborators", []): existing_card = self.state.get_card_by_name(collab_name) if existing_card: collab_id = existing_card.card_id else: collab_id = f"placeholder_{collab_name}" card.add_collaboration(Collaboration(collab_id, collab_name)) # Remove collaborations for collab_name in params.get("remove_collaborators", []): existing_card = self.state.get_card_by_name(collab_name) if existing_card: card.remove_collaboration(existing_card.card_id) def _execute_delete(self, params): """ Delete a CRC card. Args: params: Dictionary with name parameter """ name = params.get("name") card = self.state.get_card_by_name(name) if card: self.state.remove_card(card.card_id) else: print(f"Card '{name}' not found") def _execute_snapshot(self, params): """ Save the current state to files. Args: params: Dictionary with optional filename parameter """ filename = params.get("filename", f"diagram_{datetime.now().strftime('%Y%m%d_%H%M%S')}") # Ensure filename has no extension base_filename = os.path.splitext(filename)[0] # Save JSON json_path = f"{base_filename}.json" with open(json_path, 'w') as f: f.write(self.state.to_json()) print(f"Saved state to {json_path}") # Save PNG png_path = f"{base_filename}.png" self.rendering_engine.render_to_png( self.state.cards, self.state.positions, self.state.connections, png_path ) print(f"Saved diagram to {png_path}") # Save SVG svg_path = f"{base_filename}.svg" self.rendering_engine.render_to_svg( self.state.cards, self.state.positions, self.state.connections, svg_path ) print(f"Saved diagram to {svg_path}") def _execute_restore(self, params): """ Restore state from a JSON file. Args: params: Dictionary with filename parameter """ filename = params.get("filename") if not os.path.exists(filename): print(f"File '{filename}' not found") return with open(filename, 'r') as f: json_str = f.read() self.state = DiagramState.from_json(json_str) print(f"Restored state from {filename}") def _recompute_layout(self): """Recompute the layout for the current cards.""" positions, connections = self.layout_engine.compute_layout(self.state.cards) self.state.update_layout(positions, connections) def get_current_state(self): """ Get the current diagram state. Returns: DiagramState object """ return self.state# =========================================================================# MAIN APPLICATION# =========================================================================class CRCDiagramTool: """ Main application for CRC card diagramming. Provides an interactive command-line interface for creating and managing CRC card diagrams using natural language commands. """ def __init__(self, llm_provider_type="mock", **llm_kwargs): """ Initialize the CRC diagram tool. Args: llm_provider_type: Type of LLM provider to use **llm_kwargs: Provider-specific configuration """ self.llm_manager = EnhancedLLMManager() self.llm_manager.setup(llm_provider_type, **llm_kwargs) self.command_parser = CommandParser(self.llm_manager) self.layout_engine = LayoutEngine() self.rendering_engine = RenderingEngine() self.state_manager = StateManager(self.layout_engine, self.rendering_engine) def run(self): """Main application loop.""" print("=" * 70) print("CRC DIAGRAM TOOL") print("=" * 70) print("Enter natural language descriptions of your components.") print("Type 'snapshot' to save the current diagram.") print("Type 'finish' when done.") print() while True: # Get user input user_input = input("You: ").strip() if not user_input: continue # Parse command command = self.command_parser.parse(user_input) if command is None: print("Sorry, I didn't understand that. Please try again.") continue # Execute command result = self.state_manager.execute_command(command) if result == "FINISH": print("Diagram complete. Goodbye!") break elif result == "SUCCESS": print("Command executed successfully.") self._show_summary() # Cleanup self.llm_manager.shutdown() def _show_summary(self): """Display a summary of the current diagram.""" state = self.state_manager.get_current_state() print(f"\nCurrent diagram has {len(state.cards)} components:") for card in state.cards: print(f" - {card.name} ({len(card.responsibilities)} responsibilities, " f"{len(card.collaborations)} collaborations)") print()# =========================================================================# ENTRY POINT# =========================================================================def main(): """ Main entry point for the application. Parses command-line arguments and starts the tool with the appropriate configuration. """ import argparse parser = argparse.ArgumentParser( description='CRC Diagram Tool - Create CRC card diagrams using natural language' ) parser.add_argument( '--provider', choices=['mock', 'local', 'openai', 'anthropic', 'gemini'], default='mock', help='LLM provider to use (default: mock)' ) parser.add_argument( '--model', type=str, help='Model name/identifier for the LLM provider' ) parser.add_argument( '--api-key', type=str, help='API key for remote providers (or set via environment variable)' ) parser.add_argument( '--temperature', type=float, default=0.7, help='Sampling temperature (default: 0.7)' ) parser.add_argument( '--max-tokens', type=int, default=2048, help='Maximum tokens to generate (default: 2048)' ) parser.add_argument( '--quantize', choices=['none', '8bit', '4bit'], default='none', help='Quantization mode for local models (default: none)' ) args = parser.parse_args() # Build kwargs for LLM provider llm_kwargs = { 'temperature': args.temperature, 'max_tokens': args.max_tokens } if args.model: llm_kwargs['model_name'] = args.model if args.api_key: llm_kwargs['api_key'] = args.api_key if args.provider == 'local': if args.quantize == '8bit': llm_kwargs['load_in_8bit'] = True elif args.quantize == '4bit': llm_kwargs['load_in_4bit'] = True # Create and run the tool try: tool = CRCDiagramTool(llm_provider_type=args.provider, **llm_kwargs) tool.run() except KeyboardInterrupt: print("\n\nInterrupted by user. Exiting...") except Exception as e: print(f"\nError: {e}") import traceback traceback.print_exc() sys.exit(1)if __name__ == "__main__": main()USAGE INSTRUCTIONSTo run this complete example, save it to a file named crc_diagram_tool.py and execute it with Python 3.7 or later.Basic usage with mock provider (no dependencies required):python crc_diagram_tool.pyUsage with local LLM on NVIDIA GPU:python crc_diagram_tool.py --provider local --model mistralai/Mistral-7B-Instruct-v0.2Usage with local LLM with 4-bit quantization:python crc_diagram_tool.py --provider local --model mistralai/Mistral-7B-Instruct-v0.2 --quantize 4bitUsage with OpenAI GPT-4:export OPENAI_API_KEY=your-api-key-herepython crc_diagram_tool.py --provider openai --model gpt-4Usage with Anthropic Claude:export ANTHROPIC_API_KEY=your-api-key-herepython crc_diagram_tool.py --provider anthropic --model claude-3-sonnet-20240229Usage with Google Gemini:export GOOGLE_API_KEY=your-api-key-herepython crc_diagram_tool.py --provider gemini --model gemini-proThe tool will automatically detect your GPU hardware and configure itself for optimal performance. All generated diagrams are saved as PNG, SVG, and JSON files in the current directory.