INTRODUCTION
Class-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 OVERVIEW
The 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 REPRESENTATION
The 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 card
The 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 LAYER
The 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 platform
import subprocess
import os
class 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 False
The LLM provider interface defines the contract that all implementations must fulfill:
from abc import ABC, abstractmethod
class 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
pass
For 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
pass
The 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 RECOGNITION
The 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 json
class 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 card
2. MODIFY_COMPONENT: Modify an existing CRC card
3. DELETE_COMPONENT: Remove a CRC card
4. SNAPSHOT: Save the current diagram
5. RESTORE: Load a saved diagram
6. FINISH: User indicates the diagram is complete
For 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: none
Respond 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 ENGINE
The 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 math
class 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 ENGINE
The 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, ImageFont
import svgwrite
class 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 PERSISTENC
The 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 json
import os
from datetime import datetime
import uuid
class 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.state
The 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 SUPPORT
Supporting 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 CONSIDERATIONS
Building 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 INTEGRATION
Integrating 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 STRATEGIES
Comprehensive 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 ENHANCEMENTS
This 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 EXAMPLE
The 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 tool
import json
import os
import math
import uuid
import platform
import subprocess
from datetime import datetime
from abc import ABC, abstractmethod
from PIL import Image, ImageDraw, ImageFont
import 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 False
class 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."""
pass
class 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."""
pass
class 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 card
2. MODIFY_COMPONENT: Modify an existing CRC card
3. DELETE_COMPONENT: Remove a CRC card
4. SNAPSHOT: Save the current diagram
5. RESTORE: Load a saved diagram
6. FINISH: User indicates the diagram is complete
For 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: none
Respond 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 SUPPOR
OVERVIEW
This 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 REQUIREMENTS
Before 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 requests
LOCAL LLM DEPENDENCIES
For 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 INSTALLATION
For 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/cu118
For CUDA 12.1:
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121
After installing PyTorch, install the Transformers library and acceleration packages:
pip install transformers accelerate bitsandbytes sentencepiece protobuf
AMD ROCM INSTALLATION
For 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.6
After installing PyTorch, install the Transformers library:
pip install transformers accelerate sentencepiece protobuf
Note 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 INSTALLATION
For Apple Silicon Macs (M1, M2, M3 series), install PyTorch with MPS support. Recent PyTorch versions include MPS support by default:
pip install torch torchvision torchaudio
Then install the Transformers library:
pip install transformers accelerate sentencepiece protobuf
Note that MPS support requires macOS 12.3 or later and works best with float32 precision for stability.
INTEL GPU INSTALLATION
For Intel GPUs, you can use Intel Extension for PyTorch with oneAPI. First install PyTorch:
pip install torch torchvision torchaudio
Then install Intel Extension for PyTorch:
pip install intel-extension-for-pytorch
Alternatively, for OpenCL support, you can install PyOpenCL:
pip install pyopencl
REMOTE API DEPENDENCIES
For 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 openai
For Anthropic:
pip install anthropic
For Google Gemini:
pip install google-generativeai
ENHANCED HARDWARE DETECTION IMPLEMENTATION
The 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 platform
import subprocess
import os
import sys
class 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 SUPPORT
The 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 IMPLEMENTATION
The 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."""
pass
class 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."""
pass
class 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."""
pass
ENHANCED LLM MANAGER WITH ALL PROVIDERS
The 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 = None
USAGE 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 DETECTION
This 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 GPU
tool = CRCDiagramTool(
llm_provider_type="local",
model_name="mistralai/Mistral-7B-Instruct-v0.2",
temperature=0.7,
max_tokens=2048
)
# Run the interactive tool
tool.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 SILICON
This 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 used
tool = 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-4
This example uses the OpenAI API with GPT-4. You must set the OPENAI_API_KEY environment variable or pass it as a parameter.
import os
from 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 provider
tool = CRCDiagramTool(
llm_provider_type="openai",
model_name="gpt-4",
temperature=0.7,
max_tokens=2048
)
tool.run()
EXAMPLE 5: USING ANTHROPIC CLAUDE
This example uses the Anthropic API with Claude 3 Sonnet.
import os
from crc_diagram_tool import CRCDiagramTool
# Set API key
os.environ["ANTHROPIC_API_KEY"] = "your-api-key-here"
# Create tool with Anthropic provider
tool = CRCDiagramTool(
llm_provider_type="anthropic",
model_name="claude-3-sonnet-20240229",
temperature=0.7,
max_tokens=2048
)
tool.run()
EXAMPLE 6: USING GOOGLE GEMINI
This example uses the Google Gemini API.
import os
from crc_diagram_tool import CRCDiagramTool
# Set API key
os.environ["GOOGLE_API_KEY"] = "your-api-key-here"
# Create tool with Gemini provider
tool = CRCDiagramTool(
llm_provider_type="gemini",
model_name="gemini-pro",
temperature=0.7,
max_tokens=2048
)
tool.run()
EXAMPLE 7: PROGRAMMATIC USAGE WITHOUT INTERACTIVE LOOP
This 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 components
llm_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 programmatically
commands = [
"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")
# Cleanup
llm_manager.shutdown()
EXAMPLE 8: TESTING HARDWARE DETECTION
This standalone script tests hardware detection and prints detailed information about available GPU resources.
from crc_diagram_tool import EnhancedHardwareDetector
# Create detector
detector = EnhancedHardwareDetector()
# Detect and print hardware information
detector.print_hardware_info()
# Get backend information
backend, 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 TIPS
NVIDIA CUDA OPTIMIZATION
For 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 OPTIMIZATION
For 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 OPTIMIZATION
For 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 OPTIMIZATION
For 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 OPTIMIZATION
When 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 GUIDE
PROBLEM: CUDA OUT OF MEMORY ERROR
Solution: 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 DETECTED
Solution: 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 SILICON
Solution: 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 CPU
Solution: 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 LIMITING
Solution: 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 FAILS
Solution: 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.
CONCLUSION
This 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 SUPPOR
OVERVIEW
This 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 Implementation
An LLM-powered tool for creating Class-Responsibilities-Collaboration (CRC)
card diagrams through natural language commands. Supports local and remote
LLMs with automatic GPU detection and optimization for NVIDIA CUDA, AMD ROCm,
Apple MPS, and Intel GPUs.
Author: CRC Diagram Tool Development Team
Version: 1.0.0
License: MIT
"""
import json
import os
import math
import uuid
import platform
import subprocess
import sys
from datetime import datetime
from abc import ABC, abstractmethod
from PIL import Image, ImageDraw, ImageFont
import 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."""
pass
class 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."""
pass
class 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."""
pass
class 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."""
pass
class 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 card
- MODIFY_COMPONENT: Modify an existing CRC card
- DELETE_COMPONENT: Remove a CRC card
- SNAPSHOT: Save the current diagram
- RESTORE: Load a saved diagram
- FINISH: User indicates the diagram is complete
For 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: none
Respond 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 INSTRUCTIONS
To 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.py
Usage with local LLM on NVIDIA GPU:
python crc_diagram_tool.py --provider local --model mistralai/Mistral-7B-Instruct-v0.2
Usage with local LLM with 4-bit quantization:
python crc_diagram_tool.py --provider local --model mistralai/Mistral-7B-Instruct-v0.2 --quantize 4bit
Usage with OpenAI GPT-4:
export OPENAI_API_KEY=your-api-key-here
python crc_diagram_tool.py --provider openai --model gpt-4
Usage with Anthropic Claude:
export ANTHROPIC_API_KEY=your-api-key-here
python crc_diagram_tool.py --provider anthropic --model claude-3-sonnet-20240229
Usage with Google Gemini:
export GOOGLE_API_KEY=your-api-key-here
python crc_diagram_tool.py --provider gemini --model gemini-pro
The 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.
