INTRODUCTION
The analysis of large-scale software systems presents fundamental challenges that traditional tooling struggles to address effectively. When developers attempt to understand complex codebases spanning millions of lines across hundreds of modules, they encounter three primary obstacles: limited context memory in language models, systematic bias in analysis results, and hallucinations where the AI generates plausible but incorrect information. These challenges become particularly acute when analyzing database-intensive architectures where the interplay between data models, business logic, and infrastructure creates intricate dependency networks.
Modern multi-agent AI systems offer promising solutions to these challenges through architectural patterns that distribute cognitive load, validate information through multiple perspectives, and ground analysis in verifiable facts. This article explores how developers building such analysis tools can leverage specific patterns and practices to create robust, accurate, and scalable solutions for understanding large software systems.
THE CHALLENGE LANDSCAPE
Before diving into solutions, we must understand why analyzing large codebases presents unique difficulties. A typical enterprise database application might contain thousands of classes, hundreds of database tables, dozens of microservices, and complex deployment pipelines. When an AI attempts to analyze such a system holistically, it faces the context window limitation where only a fraction of the codebase fits into the model's working memory at once. This is analogous to trying to understand a novel by reading only three pages at a time without notes.
Bias emerges when the training data of language models contains patterns that don't match the specific architectural decisions of the system under analysis. For instance, a model trained predominantly on open-source projects might misinterpret proprietary patterns or domain-specific conventions. Hallucinations occur when the model fills knowledge gaps with plausible-sounding but factually incorrect information, such as inventing method signatures that don't exist or describing architectural layers that aren't present in the actual system.
HUMAN IN THE LOOP PATTERN
The human-in-the-loop pattern represents a fundamental acknowledgment that AI systems should not operate in isolation when making critical architectural assessments. This pattern involves strategically placing decision points where human experts validate assumptions, provide missing context, or approve proposed changes before they propagate through the analysis pipeline.
Consider a scenario where the AI encounters an ambiguous class relationship. Rather than making an assumption that might cascade into incorrect conclusions, the system should pause and query the human operator. The implementation requires careful design of the interaction protocol to avoid overwhelming users with trivial questions while ensuring critical decisions receive human oversight.
A practical implementation might look like this in Python:
class HumanValidationService:
"""
Service for requesting human validation of AI assumptions
during architectural analysis.
"""
def __init__(self, validation_threshold=0.7):
"""
Initialize the validation service.
Args:
validation_threshold: Confidence level below which
human validation is required
"""
self.validation_threshold = validation_threshold
self.pending_validations = []
self.validation_history = {}
def request_validation(self, context, assumption, confidence):
"""
Request human validation for an assumption made during analysis.
Args:
context: Dictionary containing relevant code context
assumption: String describing the assumption being made
confidence: Float between 0 and 1 indicating AI confidence
Returns:
Boolean indicating whether assumption was validated
"""
if confidence >= self.validation_threshold:
# High confidence, proceed without human validation
return True
validation_request = {
'context': context,
'assumption': assumption,
'confidence': confidence,
'timestamp': datetime.now()
}
self.pending_validations.append(validation_request)
# Present to human operator
response = self._present_to_human(validation_request)
# Store in history for learning
self.validation_history[assumption] = response
return response['approved']
def _present_to_human(self, request):
"""
Present validation request to human operator.
This would integrate with a UI in production.
"""
print(f"\nValidation Required (Confidence: {request['confidence']:.2f})")
print(f"Context: {request['context']['description']}")
print(f"Assumption: {request['assumption']}")
print("\nRelevant Code:")
print(request['context']['code_snippet'])
# In production, this would be an async UI interaction
user_input = input("\nApprove this assumption? (y/n): ")
return {
'approved': user_input.lower() == 'y',
'feedback': input("Additional context (optional): ")
}
This service integrates into the broader analysis pipeline at decision points. For example, when determining whether a class implements a particular design pattern, the system checks its confidence level and requests human validation if uncertainty exceeds acceptable thresholds. The validation history allows the system to learn from past decisions and potentially adjust its confidence calibration over time.
The key insight is that human validation should be selective and contextual. Asking for validation on every minor detail creates cognitive overload and defeats the purpose of automation. Instead, the system should identify high-impact decisions where errors would cascade through the analysis, such as identifying system boundaries, determining the primary architectural style, or classifying critical business entities.
LIGHTWEIGHT PROMPT DECOMPOSITION
Rather than constructing massive prompts that attempt to analyze entire subsystems in a single pass, effective analysis tools decompose complex analytical tasks into sequences of focused, lightweight prompts. This approach offers multiple advantages: it reduces the cognitive load on each individual model invocation, allows for better error isolation and recovery, enables parallel processing of independent subtasks, and makes the analysis process more transparent and debuggable.
Consider the task of analyzing a database access layer. A monolithic prompt might attempt to understand the entire layer's architecture, identify all data access patterns, detect potential performance issues, and suggest improvements simultaneously. This creates a complex analytical burden that increases the likelihood of hallucinations and makes it difficult to validate individual conclusions.
Instead, we decompose this into a sequence of focused prompts. First, we identify the structural components of the data access layer. Second, we analyze the patterns used in each component independently. Third, we examine cross-component interactions. Fourth, we assess performance characteristics. Finally, we synthesize findings into recommendations. Each step builds on validated results from previous steps, creating a more reliable analytical chain.
Here is an implementation demonstrating this decomposition:
class PromptDecompositionEngine:
"""
Engine for decomposing complex analysis tasks into
sequences of lightweight, focused prompts.
"""
def __init__(self, llm_client):
"""
Initialize the decomposition engine.
Args:
llm_client: Client for interacting with language model
"""
self.llm_client = llm_client
self.task_graph = {}
self.results_cache = {}
def analyze_data_access_layer(self, code_artifacts):
"""
Analyze a data access layer through decomposed prompts.
Args:
code_artifacts: List of code files in the data access layer
Returns:
Comprehensive analysis results
"""
# Step 1: Identify structural components
components = self._identify_components(code_artifacts)
# Step 2: Analyze patterns in each component
pattern_analysis = {}
for component in components:
pattern_analysis[component['name']] = \
self._analyze_component_patterns(component)
# Step 3: Examine cross-component interactions
interactions = self._analyze_interactions(
components,
pattern_analysis
)
# Step 4: Assess performance characteristics
performance = self._assess_performance(
components,
interactions
)
# Step 5: Synthesize recommendations
recommendations = self._synthesize_recommendations(
components,
pattern_analysis,
interactions,
performance
)
return {
'components': components,
'patterns': pattern_analysis,
'interactions': interactions,
'performance': performance,
'recommendations': recommendations
}
def _identify_components(self, code_artifacts):
"""
First decomposed step: identify structural components.
"""
prompt = self._build_component_identification_prompt(
code_artifacts
)
response = self.llm_client.generate(
prompt,
max_tokens=1000,
temperature=0.3 # Lower temperature for factual analysis
)
# Parse and validate response
components = self._parse_component_response(response)
# Cache results for potential reuse
self.results_cache['components'] = components
return components
def _build_component_identification_prompt(self, artifacts):
"""
Build focused prompt for component identification.
"""
# Include only relevant code snippets, not entire files
code_summary = self._summarize_artifacts(artifacts)
prompt = f"""Analyze the following data access layer code and identify
the distinct structural components (repositories, DAOs, query builders, etc.).
For each component, provide:
1. Component name
2. Primary responsibility
3. Files involved
Code Summary:
{code_summary}
Respond in JSON format with an array of components."""
return prompt
def _analyze_component_patterns(self, component):
"""
Second decomposed step: analyze patterns in a single component.
"""
prompt = f"""Analyze the design patterns used in this component:
Component: {component['name']}
Responsibility: {component['responsibility']}
Code:
{component['code']}
Identify:
1. Primary design pattern (Repository, DAO, Active Record, etc.)
2. Secondary patterns used
3. Pattern implementation quality
Respond in JSON format."""
response = self.llm_client.generate(
prompt,
max_tokens=800,
temperature=0.3
)
return self._parse_pattern_response(response)
def _summarize_artifacts(self, artifacts):
"""
Create concise summary of code artifacts for prompt inclusion.
"""
summary_parts = []
for artifact in artifacts[:10]: # Limit to prevent context overflow
summary_parts.append(
f"File: {artifact['path']}\n"
f"Classes: {', '.join(artifact['classes'])}\n"
f"Key Methods: {', '.join(artifact['key_methods'][:5])}\n"
)
return '\n'.join(summary_parts)
This decomposition approach ensures that each prompt has a clear, focused objective and receives only the context necessary to accomplish that objective. The sequential nature allows each step to build on validated results from previous steps, reducing the compounding of errors that can occur in monolithic analysis.
The temperature parameter in the language model calls deserves attention. Lower temperatures around 0.3 produce more deterministic, factual outputs suitable for structural analysis. Higher temperatures might be appropriate for creative tasks like generating alternative architectural suggestions, but for factual code analysis, consistency and accuracy take precedence over creativity.
PROMPT OPTIMIZATION STRATEGIES
Effective prompt engineering forms the foundation of reliable AI-driven code analysis. Optimization involves both the structure of individual prompts and the orchestration of prompt sequences. The goal is to maximize the accuracy and relevance of responses while minimizing token usage and latency.
Several key principles guide prompt optimization. First, prompts should be specific and unambiguous, leaving no room for interpretation about what information is requested. Second, prompts should provide sufficient context without overwhelming the model with irrelevant details. Third, prompts should specify the desired output format explicitly, particularly when the response will be parsed programmatically. Fourth, prompts should include examples when the task involves pattern recognition or classification.
Consider the difference between a poorly optimized prompt and an optimized version for identifying architectural layers:
Poor version: "Look at this code and tell me about the architecture."
This prompt is vague about what architectural aspects matter, provides no context about the system, and doesn't specify output format. The model might respond with generic architectural principles rather than specific analysis of the provided code.
Optimized version:
def build_layer_identification_prompt(code_context, system_type):
"""
Build optimized prompt for architectural layer identification.
Args:
code_context: Dictionary with code samples and metadata
system_type: String indicating system type (web, batch, etc.)
Returns:
Optimized prompt string
"""
prompt = f"""You are analyzing a {system_type} application to identify
its architectural layers.
System Context:
- Technology Stack: {code_context['tech_stack']}
- Primary Language: {code_context['language']}
- Framework: {code_context['framework']}
Code Sample from Module: {code_context['module_name']}
{code_context['code_sample']}
Task: Identify which architectural layer this module belongs to.
Common layers in {system_type} applications:
- Presentation Layer: UI components, controllers, view models
- Application Layer: Use cases, application services, DTOs
- Domain Layer: Business entities, domain services, business rules
- Infrastructure Layer: Database access, external services, frameworks
Respond in this exact JSON format:
{{
"primary_layer": "layer name",
"confidence": 0.0 to 1.0,
"reasoning": "explanation of classification",
"indicators": ["specific code elements that indicate this layer"]
}}
Base your classification on actual code patterns, not assumptions."""
return prompt
This optimized prompt provides clear context about the system type, includes relevant background about common architectural layers, specifies the exact JSON output format expected, and explicitly instructs the model to base conclusions on observable code patterns rather than assumptions. The confidence field allows the system to trigger human validation when uncertainty is high.
Prompt optimization also involves iterative refinement based on observed model behavior. If the model consistently misclassifies certain patterns, the prompt can be enhanced with negative examples or additional clarifying instructions. This creates a feedback loop where prompt quality improves over time based on real-world usage.
Another optimization technique involves prompt templates with variable injection points. Rather than constructing prompts from scratch for each analysis task, we maintain a library of proven templates and inject specific context at runtime:
class PromptTemplateLibrary:
"""
Library of optimized prompt templates for various analysis tasks.
"""
def __init__(self):
self.templates = self._initialize_templates()
self.performance_metrics = {}
def _initialize_templates(self):
"""
Initialize library with proven prompt templates.
"""
return {
'layer_identification': {
'template': """Analyze the following {language} code module
to determine its architectural layer.
System Type: {system_type}
Module: {module_name}
Code:
{code_sample}
Identify the layer based on these characteristics:
{layer_characteristics}
Output Format:
{output_format}""",
'required_vars': [
'language',
'system_type',
'module_name',
'code_sample',
'layer_characteristics',
'output_format'
]
},
'pattern_detection': {
'template': """Identify design patterns in this {language} code.
Context: {context}
Code:
{code_sample}
Focus on these pattern categories:
{pattern_categories}
For each pattern found, provide:
1. Pattern name
2. Confidence level
3. Code elements demonstrating the pattern
4. Pattern quality assessment
{output_format}""",
'required_vars': [
'language',
'context',
'code_sample',
'pattern_categories',
'output_format'
]
}
}
def get_prompt(self, template_name, variables):
"""
Generate prompt from template with variable substitution.
Args:
template_name: Name of template to use
variables: Dictionary of variables to inject
Returns:
Formatted prompt string
"""
if template_name not in self.templates:
raise ValueError(f"Unknown template: {template_name}")
template_config = self.templates[template_name]
# Validate all required variables are provided
missing = set(template_config['required_vars']) - set(variables.keys())
if missing:
raise ValueError(f"Missing required variables: {missing}")
# Substitute variables into template
prompt = template_config['template'].format(**variables)
return prompt
def track_performance(self, template_name, accuracy, latency):
"""
Track performance metrics for prompt templates.
Args:
template_name: Name of template used
accuracy: Accuracy score for this usage
latency: Response latency in seconds
"""
if template_name not in self.performance_metrics:
self.performance_metrics[template_name] = {
'uses': 0,
'total_accuracy': 0.0,
'total_latency': 0.0
}
metrics = self.performance_metrics[template_name]
metrics['uses'] += 1
metrics['total_accuracy'] += accuracy
metrics['total_latency'] += latency
def get_best_template(self, task_type):
"""
Get best performing template for a task type.
Args:
task_type: Type of analysis task
Returns:
Name of best performing template
"""
candidates = [
name for name in self.templates.keys()
if task_type in name
]
if not candidates:
return None
# Select based on accuracy-weighted performance
best = max(
candidates,
key=lambda name: (
self.performance_metrics.get(name, {}).get('total_accuracy', 0) /
max(self.performance_metrics.get(name, {}).get('uses', 1), 1)
)
)
return best
This template library approach provides several benefits. Templates are tested and refined over time, reducing the need to engineer prompts from scratch for each new analysis task. Performance tracking enables data-driven decisions about which templates work best for specific scenarios. The variable injection mechanism ensures consistency while allowing customization for specific contexts.
REASONING STRATEGIES FOR ENHANCED ACCURACY
Advanced reasoning strategies significantly improve the accuracy and reliability of AI-driven code analysis. Three particularly effective approaches are self-reflection, chain-of-thought reasoning, and tree-of-thought exploration. Each strategy addresses different aspects of the analytical challenge.
Self-reflection involves having the AI critique and validate its own conclusions. After generating an initial analysis, the system prompts the model to review its reasoning, identify potential errors or biases, and revise conclusions if necessary. This metacognitive approach often catches logical inconsistencies or unsupported assumptions that would otherwise go unnoticed.
Implementation of self-reflection requires a two-phase process. The first phase generates the initial analysis. The second phase presents that analysis back to the model with instructions to critically evaluate it:
class SelfReflectionAnalyzer:
"""
Analyzer that uses self-reflection to validate conclusions.
"""
def __init__(self, llm_client):
self.llm_client = llm_client
self.reflection_history = []
def analyze_with_reflection(self, code_sample, analysis_type):
"""
Perform analysis with self-reflection validation.
Args:
code_sample: Code to analyze
analysis_type: Type of analysis to perform
Returns:
Validated analysis results
"""
# Phase 1: Initial analysis
initial_prompt = self._build_analysis_prompt(
code_sample,
analysis_type
)
initial_response = self.llm_client.generate(
initial_prompt,
max_tokens=1500,
temperature=0.4
)
initial_analysis = self._parse_response(initial_response)
# Phase 2: Self-reflection
reflection_prompt = self._build_reflection_prompt(
code_sample,
initial_analysis,
analysis_type
)
reflection_response = self.llm_client.generate(
reflection_prompt,
max_tokens=1000,
temperature=0.3
)
reflection_result = self._parse_reflection(reflection_response)
# Combine initial analysis with reflection insights
final_analysis = self._integrate_reflection(
initial_analysis,
reflection_result
)
# Store for learning
self.reflection_history.append({
'initial': initial_analysis,
'reflection': reflection_result,
'final': final_analysis
})
return final_analysis
def _build_reflection_prompt(self, code, initial_analysis, analysis_type):
"""
Build prompt for self-reflection phase.
"""
prompt = f"""Review the following analysis of {analysis_type} and
critically evaluate its accuracy.
Original Code:
{code}
Initial Analysis:
{json.dumps(initial_analysis, indent=2)}
Critically assess this analysis by answering:
1. Are the conclusions supported by observable code patterns?
2. Are there alternative interpretations that better fit the evidence?
3. Are there any logical inconsistencies in the reasoning?
4. What assumptions were made, and are they justified?
5. What evidence contradicts or weakens the conclusions?
Provide your reflection in this format:
{{
"validity_assessment": "strong/moderate/weak",
"unsupported_claims": ["list any claims not supported by code"],
"alternative_interpretations": ["list better interpretations if any"],
"logical_issues": ["list any logical problems"],
"unjustified_assumptions": ["list problematic assumptions"],
"recommended_revisions": ["specific changes to improve accuracy"]
}}"""
return prompt
def _integrate_reflection(self, initial, reflection):
"""
Integrate reflection insights into final analysis.
"""
final = initial.copy()
# Apply recommended revisions
if reflection.get('recommended_revisions'):
for revision in reflection['recommended_revisions']:
final = self._apply_revision(final, revision)
# Add confidence adjustment based on reflection
if reflection.get('validity_assessment') == 'weak':
final['confidence'] *= 0.6
elif reflection.get('validity_assessment') == 'moderate':
final['confidence'] *= 0.8
# Include reflection metadata
final['reflection_metadata'] = {
'validity': reflection.get('validity_assessment'),
'issues_found': len(reflection.get('logical_issues', [])),
'revisions_applied': len(reflection.get('recommended_revisions', []))
}
return final
Chain-of-thought reasoning makes the analytical process explicit by requiring the model to articulate its reasoning steps before reaching conclusions. This approach reduces errors by forcing the model to construct a logical argument rather than jumping directly to answers. For architectural analysis, chain-of-thought prompts might ask the model to first identify relevant code patterns, then explain how those patterns relate to architectural principles, and finally conclude what architectural style is present.
Here is an implementation of chain-of-thought reasoning for dependency analysis:
class ChainOfThoughtAnalyzer:
"""
Analyzer using chain-of-thought reasoning for dependency analysis.
"""
def __init__(self, llm_client):
self.llm_client = llm_client
def analyze_dependencies(self, module_code, module_name):
"""
Analyze module dependencies using chain-of-thought reasoning.
Args:
module_code: Source code of the module
module_name: Name of the module
Returns:
Dependency analysis with reasoning chain
"""
prompt = f"""Analyze the dependencies of this module using step-by-step
reasoning. Work through each step explicitly before reaching conclusions.
Module: {module_name}
Code:
{module_code}
Step 1: Identify all import statements and external references.
List each import and what it provides.
Step 2: Classify each dependency by type.
For each dependency, determine if it is:
- Framework dependency (infrastructure)
- Domain dependency (business logic)
- Utility dependency (helpers, tools)
- External service dependency (APIs, databases)
Step 3: Analyze dependency direction and layering.
For each dependency, determine:
- Which architectural layer does it belong to?
- Is the dependency direction appropriate (following dependency inversion)?
- Are there any circular dependencies?
Step 4: Assess dependency quality.
For each dependency, evaluate:
- Is it necessary or could it be eliminated?
- Does it create tight coupling?
- Does it violate architectural principles?
Step 5: Synthesize conclusions.
Based on the above analysis, provide:
- Overall dependency health score (0-100)
- Key issues identified
- Recommended improvements
Work through each step explicitly, showing your reasoning."""
response = self.llm_client.generate(
prompt,
max_tokens=2500,
temperature=0.3
)
# Parse the step-by-step reasoning
reasoning_chain = self._parse_reasoning_chain(response)
return reasoning_chain
def _parse_reasoning_chain(self, response):
"""
Parse chain-of-thought response into structured format.
"""
# Extract each reasoning step
steps = {}
current_step = None
for line in response.split('\n'):
if line.startswith('Step '):
current_step = line.split(':')[0].strip()
steps[current_step] = []
elif current_step and line.strip():
steps[current_step].append(line.strip())
# Extract final conclusions
conclusions = steps.get('Step 5', [])
return {
'reasoning_steps': steps,
'conclusions': conclusions,
'full_response': response
}
Tree-of-thought reasoning extends chain-of-thought by exploring multiple reasoning paths simultaneously. Rather than following a single analytical path, the system generates several alternative interpretations and evaluates them comparatively. This is particularly valuable when analyzing ambiguous code where multiple architectural interpretations might be plausible.
Implementation of tree-of-thought requires generating multiple reasoning branches and then evaluating them:
class TreeOfThoughtAnalyzer:
"""
Analyzer using tree-of-thought reasoning to explore multiple
interpretations of architectural patterns.
"""
def __init__(self, llm_client, num_branches=3):
self.llm_client = llm_client
self.num_branches = num_branches
def analyze_architecture(self, code_artifacts):
"""
Analyze architecture by exploring multiple interpretation branches.
Args:
code_artifacts: Collection of code files to analyze
Returns:
Best interpretation with comparative analysis
"""
# Generate multiple interpretation branches
branches = []
for i in range(self.num_branches):
branch = self._generate_interpretation_branch(
code_artifacts,
branch_id=i
)
branches.append(branch)
# Evaluate and compare branches
evaluation = self._evaluate_branches(branches, code_artifacts)
# Select best interpretation
best_branch = self._select_best_branch(branches, evaluation)
return {
'selected_interpretation': best_branch,
'alternative_interpretations': [
b for b in branches if b != best_branch
],
'evaluation': evaluation
}
def _generate_interpretation_branch(self, artifacts, branch_id):
"""
Generate one interpretation branch.
"""
prompt = f"""Analyze this codebase and propose an interpretation of
its architectural style. This is interpretation branch {branch_id + 1}.
Consider these possible architectural styles:
- Layered Architecture
- Hexagonal Architecture (Ports and Adapters)
- Clean Architecture
- Microservices Architecture
- Event-Driven Architecture
- CQRS (Command Query Responsibility Segregation)
Code Artifacts Summary:
{self._summarize_artifacts(artifacts)}
Provide your interpretation including:
1. Primary architectural style identified
2. Key evidence supporting this interpretation
3. How major components fit this style
4. Confidence level in this interpretation
Focus on evidence-based reasoning. Consider alternative explanations."""
response = self.llm_client.generate(
prompt,
max_tokens=1500,
temperature=0.6 + (branch_id * 0.1) # Vary temperature per branch
)
return self._parse_interpretation(response, branch_id)
def _evaluate_branches(self, branches, artifacts):
"""
Evaluate and compare interpretation branches.
"""
prompt = f"""Compare these {len(branches)} different interpretations
of the same codebase's architecture.
Code Context:
{self._summarize_artifacts(artifacts)}
Interpretations:
"""
for i, branch in enumerate(branches):
prompt += f"\nInterpretation {i + 1}:\n"
prompt += f"Style: {branch['style']}\n"
prompt += f"Evidence: {branch['evidence']}\n"
prompt += f"Confidence: {branch['confidence']}\n"
prompt += """
Evaluate each interpretation by:
1. How well the evidence supports the claimed style
2. Whether the interpretation accounts for all major components
3. Consistency of the interpretation across the codebase
4. Plausibility given common architectural patterns
Rank the interpretations from most to least plausible and explain why."""
response = self.llm_client.generate(
prompt,
max_tokens=1200,
temperature=0.3
)
return self._parse_evaluation(response)
def _select_best_branch(self, branches, evaluation):
"""
Select best interpretation based on evaluation.
"""
# Extract ranking from evaluation
ranking = evaluation.get('ranking', [])
if not ranking:
# Fallback to highest confidence
return max(branches, key=lambda b: b.get('confidence', 0))
# Return top-ranked interpretation
best_id = ranking[0]
return branches[best_id]
def _summarize_artifacts(self, artifacts):
"""
Create summary of code artifacts for prompt.
"""
summary = []
for artifact in artifacts[:15]:
summary.append(
f"File: {artifact['path']}\n"
f"Type: {artifact['type']}\n"
f"Key Elements: {', '.join(artifact['key_elements'][:5])}"
)
return '\n\n'.join(summary)
These reasoning strategies significantly improve analytical accuracy by making the reasoning process explicit and subject to validation. Self-reflection catches errors in initial analysis. Chain-of-thought ensures logical consistency. Tree-of-thought explores alternative interpretations to avoid premature conclusions. Together, they create a robust analytical framework that produces more reliable results than single-pass analysis.
LANGUAGE PARSERS AND ABSTRACT SYNTAX TREES
Accurate code analysis requires parsing source code into structured representations that AI systems can reliably process. Abstract Syntax Trees provide this structure by representing code as hierarchical trees of syntactic elements. While Python has excellent built-in parsing support through the ast module, analyzing polyglot codebases requires parsers for multiple languages.
For Python, the ast module provides comprehensive parsing capabilities. Here is an implementation that extracts architectural information from Python code:
import ast
import os
from typing import Dict, List, Set
class PythonCodeParser:
"""
Parser for extracting architectural information from Python code.
"""
def __init__(self):
self.parsed_modules = {}
self.dependency_graph = {}
def parse_file(self, filepath):
"""
Parse a Python file and extract architectural elements.
Args:
filepath: Path to Python source file
Returns:
Dictionary containing parsed architectural information
"""
with open(filepath, 'r', encoding='utf-8') as f:
source_code = f.read()
try:
tree = ast.parse(source_code, filename=filepath)
except SyntaxError as e:
return {
'error': f"Syntax error in {filepath}: {str(e)}",
'filepath': filepath
}
# Extract architectural elements
visitor = ArchitectureVisitor(filepath)
visitor.visit(tree)
result = {
'filepath': filepath,
'classes': visitor.classes,
'functions': visitor.functions,
'imports': visitor.imports,
'dependencies': visitor.dependencies,
'complexity_metrics': visitor.complexity_metrics
}
self.parsed_modules[filepath] = result
return result
def parse_directory(self, directory_path):
"""
Recursively parse all Python files in a directory.
Args:
directory_path: Root directory to parse
Returns:
Dictionary mapping filepaths to parse results
"""
results = {}
for root, dirs, files in os.walk(directory_path):
# Skip common non-source directories
dirs[:] = [d for d in dirs if d not in [
'__pycache__', '.git', 'venv', 'env', '.tox'
]]
for filename in files:
if filename.endswith('.py'):
filepath = os.path.join(root, filename)
results[filepath] = self.parse_file(filepath)
return results
def build_dependency_graph(self):
"""
Build dependency graph from parsed modules.
Returns:
Dictionary representing module dependencies
"""
graph = {}
for filepath, module_info in self.parsed_modules.items():
module_name = self._filepath_to_module_name(filepath)
dependencies = set()
for imp in module_info.get('imports', []):
dep_module = imp.get('module')
if dep_module:
dependencies.add(dep_module)
graph[module_name] = list(dependencies)
self.dependency_graph = graph
return graph
def _filepath_to_module_name(self, filepath):
"""
Convert filepath to Python module name.
"""
# Remove .py extension and convert path separators to dots
module = filepath.replace('.py', '').replace(os.sep, '.')
# Remove leading dots
while module.startswith('.'):
module = module[1:]
return module
class ArchitectureVisitor(ast.NodeVisitor):
"""
AST visitor for extracting architectural information.
"""
def __init__(self, filepath):
self.filepath = filepath
self.classes = []
self.functions = []
self.imports = []
self.dependencies = set()
self.complexity_metrics = {
'total_lines': 0,
'code_lines': 0,
'cyclomatic_complexity': 0
}
self.current_class = None
def visit_ClassDef(self, node):
"""
Visit class definition and extract information.
"""
class_info = {
'name': node.name,
'line_number': node.lineno,
'bases': [self._get_name(base) for base in node.bases],
'methods': [],
'decorators': [self._get_name(dec) for dec in node.decorator_list],
'docstring': ast.get_docstring(node)
}
# Store current class context for method processing
previous_class = self.current_class
self.current_class = class_info
# Visit child nodes (methods)
self.generic_visit(node)
# Restore previous context
self.current_class = previous_class
self.classes.append(class_info)
def visit_FunctionDef(self, node):
"""
Visit function/method definition and extract information.
"""
func_info = {
'name': node.name,
'line_number': node.lineno,
'parameters': [arg.arg for arg in node.args.args],
'decorators': [self._get_name(dec) for dec in node.decorator_list],
'docstring': ast.get_docstring(node),
'is_async': isinstance(node, ast.AsyncFunctionDef),
'complexity': self._calculate_complexity(node)
}
if self.current_class:
# This is a method
self.current_class['methods'].append(func_info)
else:
# This is a module-level function
self.functions.append(func_info)
# Update complexity metrics
self.complexity_metrics['cyclomatic_complexity'] += func_info['complexity']
self.generic_visit(node)
def visit_Import(self, node):
"""
Visit import statement and track dependencies.
"""
for alias in node.names:
import_info = {
'type': 'import',
'module': alias.name,
'alias': alias.asname,
'line_number': node.lineno
}
self.imports.append(import_info)
self.dependencies.add(alias.name.split('.')[0])
self.generic_visit(node)
def visit_ImportFrom(self, node):
"""
Visit from-import statement and track dependencies.
"""
module = node.module if node.module else ''
for alias in node.names:
import_info = {
'type': 'from_import',
'module': module,
'name': alias.name,
'alias': alias.asname,
'line_number': node.lineno
}
self.imports.append(import_info)
if module:
self.dependencies.add(module.split('.')[0])
self.generic_visit(node)
def _get_name(self, node):
"""
Extract name from AST node.
"""
if isinstance(node, ast.Name):
return node.id
elif isinstance(node, ast.Attribute):
return f"{self._get_name(node.value)}.{node.attr}"
elif isinstance(node, ast.Call):
return self._get_name(node.func)
else:
return str(node)
def _calculate_complexity(self, node):
"""
Calculate cyclomatic complexity of a function.
Cyclomatic complexity counts decision points in code.
"""
complexity = 1 # Base complexity
for child in ast.walk(node):
# Each decision point adds to complexity
if isinstance(child, (ast.If, ast.While, ast.For, ast.ExceptHandler)):
complexity += 1
elif isinstance(child, ast.BoolOp):
# Each boolean operator adds a decision point
complexity += len(child.values) - 1
return complexity
This Python parser extracts classes, methods, functions, imports, and calculates basic complexity metrics. The AST visitor pattern allows systematic traversal of the syntax tree to collect architectural information. The dependency graph construction enables analysis of module relationships and potential circular dependencies.
For languages beyond Python, tree-sitter provides a universal parsing framework. Tree-sitter generates parsers from grammar definitions and produces consistent ASTs across languages. Here is an implementation using tree-sitter for multi-language parsing:
from tree_sitter import Language, Parser
import tree_sitter_python
import tree_sitter_java
import tree_sitter_javascript
import tree_sitter_go
class MultiLanguageParser:
"""
Parser supporting multiple programming languages via tree-sitter.
"""
def __init__(self):
# Initialize parsers for supported languages
self.parsers = {
'python': self._init_python_parser(),
'java': self._init_java_parser(),
'javascript': self._init_javascript_parser(),
'go': self._init_go_parser()
}
self.language_extensions = {
'.py': 'python',
'.java': 'java',
'.js': 'javascript',
'.jsx': 'javascript',
'.go': 'go'
}
def _init_python_parser(self):
"""Initialize Python parser."""
parser = Parser()
parser.set_language(Language(tree_sitter_python.language()))
return parser
def _init_java_parser(self):
"""Initialize Java parser."""
parser = Parser()
parser.set_language(Language(tree_sitter_java.language()))
return parser
def _init_javascript_parser(self):
"""Initialize JavaScript parser."""
parser = Parser()
parser.set_language(Language(tree_sitter_javascript.language()))
return parser
def _init_go_parser(self):
"""Initialize Go parser."""
parser = Parser()
parser.set_language(Language(tree_sitter_go.language()))
return parser
def parse_file(self, filepath):
"""
Parse a source file in any supported language.
Args:
filepath: Path to source file
Returns:
Dictionary containing parsed information
"""
# Determine language from file extension
extension = os.path.splitext(filepath)[1]
language = self.language_extensions.get(extension)
if not language:
return {
'error': f"Unsupported file type: {extension}",
'filepath': filepath
}
# Read source code
with open(filepath, 'rb') as f:
source_code = f.read()
# Parse with appropriate parser
parser = self.parsers[language]
tree = parser.parse(source_code)
# Extract architectural elements based on language
extractor = self._get_extractor(language)
elements = extractor.extract(tree, source_code)
return {
'filepath': filepath,
'language': language,
'elements': elements,
'ast': tree
}
def _get_extractor(self, language):
"""
Get appropriate element extractor for language.
"""
extractors = {
'python': PythonElementExtractor(),
'java': JavaElementExtractor(),
'javascript': JavaScriptElementExtractor(),
'go': GoElementExtractor()
}
return extractors[language]
class PythonElementExtractor:
"""
Extract architectural elements from Python AST.
"""
def extract(self, tree, source_code):
"""
Extract classes, functions, and imports from Python code.
"""
root_node = tree.root_node
elements = {
'classes': [],
'functions': [],
'imports': []
}
# Query for class definitions
class_query = """
(class_definition
name: (identifier) @class_name
body: (block) @class_body)
"""
# Query for function definitions
function_query = """
(function_definition
name: (identifier) @func_name
parameters: (parameters) @params)
"""
# Query for imports
import_query = """
(import_statement
name: (dotted_name) @module)
"""
# Execute queries and extract information
# Note: Actual tree-sitter query execution would use the query API
# This is a simplified representation
elements['classes'] = self._extract_classes(root_node, source_code)
elements['functions'] = self._extract_functions(root_node, source_code)
elements['imports'] = self._extract_imports(root_node, source_code)
return elements
def _extract_classes(self, node, source_code):
"""Extract class definitions from AST."""
classes = []
def visit(n):
if n.type == 'class_definition':
class_name = None
for child in n.children:
if child.type == 'identifier':
class_name = source_code[child.start_byte:child.end_byte].decode('utf-8')
break
if class_name:
classes.append({
'name': class_name,
'line': n.start_point[0] + 1,
'methods': self._extract_methods(n, source_code)
})
for child in n.children:
visit(child)
visit(node)
return classes
def _extract_methods(self, class_node, source_code):
"""Extract methods from class node."""
methods = []
for node in class_node.children:
if node.type == 'block':
for child in node.children:
if child.type == 'function_definition':
method_name = None
for subchild in child.children:
if subchild.type == 'identifier':
method_name = source_code[subchild.start_byte:subchild.end_byte].decode('utf-8')
break
if method_name:
methods.append({
'name': method_name,
'line': child.start_point[0] + 1
})
return methods
def _extract_functions(self, node, source_code):
"""Extract top-level function definitions."""
functions = []
for child in node.children:
if child.type == 'function_definition':
func_name = None
for subchild in child.children:
if subchild.type == 'identifier':
func_name = source_code[subchild.start_byte:subchild.end_byte].decode('utf-8')
break
if func_name:
functions.append({
'name': func_name,
'line': child.start_point[0] + 1
})
return functions
def _extract_imports(self, node, source_code):
"""Extract import statements."""
imports = []
def visit(n):
if n.type in ['import_statement', 'import_from_statement']:
import_text = source_code[n.start_byte:n.end_byte].decode('utf-8')
imports.append({
'statement': import_text,
'line': n.start_point[0] + 1
})
for child in n.children:
visit(child)
visit(node)
return imports
class JavaElementExtractor:
"""
Extract architectural elements from Java AST.
"""
def extract(self, tree, source_code):
"""
Extract classes, interfaces, methods from Java code.
"""
root_node = tree.root_node
elements = {
'classes': [],
'interfaces': [],
'methods': [],
'imports': []
}
elements['classes'] = self._extract_classes(root_node, source_code)
elements['interfaces'] = self._extract_interfaces(root_node, source_code)
elements['imports'] = self._extract_imports(root_node, source_code)
return elements
def _extract_classes(self, node, source_code):
"""Extract class declarations from Java AST."""
classes = []
def visit(n):
if n.type == 'class_declaration':
class_name = None
modifiers = []
for child in n.children:
if child.type == 'identifier':
class_name = source_code[child.start_byte:child.end_byte].decode('utf-8')
elif child.type == 'modifiers':
modifiers = self._extract_modifiers(child, source_code)
if class_name:
classes.append({
'name': class_name,
'modifiers': modifiers,
'line': n.start_point[0] + 1,
'methods': self._extract_class_methods(n, source_code)
})
for child in n.children:
visit(child)
visit(node)
return classes
def _extract_interfaces(self, node, source_code):
"""Extract interface declarations from Java AST."""
interfaces = []
def visit(n):
if n.type == 'interface_declaration':
interface_name = None
for child in n.children:
if child.type == 'identifier':
interface_name = source_code[child.start_byte:child.end_byte].decode('utf-8')
break
if interface_name:
interfaces.append({
'name': interface_name,
'line': n.start_point[0] + 1
})
for child in n.children:
visit(child)
visit(node)
return interfaces
def _extract_class_methods(self, class_node, source_code):
"""Extract methods from Java class."""
methods = []
def visit(n):
if n.type == 'method_declaration':
method_name = None
return_type = None
for child in n.children:
if child.type == 'identifier':
method_name = source_code[child.start_byte:child.end_byte].decode('utf-8')
elif child.type in ['type_identifier', 'void_type']:
return_type = source_code[child.start_byte:child.end_byte].decode('utf-8')
if method_name:
methods.append({
'name': method_name,
'return_type': return_type,
'line': n.start_point[0] + 1
})
for child in n.children:
visit(child)
visit(class_node)
return methods
def _extract_modifiers(self, modifiers_node, source_code):
"""Extract access modifiers from Java code."""
modifiers = []
for child in modifiers_node.children:
modifier = source_code[child.start_byte:child.end_byte].decode('utf-8')
modifiers.append(modifier)
return modifiers
def _extract_imports(self, node, source_code):
"""Extract import statements from Java code."""
imports = []
def visit(n):
if n.type == 'import_declaration':
import_text = source_code[n.start_byte:n.end_byte].decode('utf-8')
imports.append({
'statement': import_text.strip(),
'line': n.start_point[0] + 1
})
for child in n.children:
visit(child)
visit(node)
return imports
This multi-language parsing infrastructure provides consistent extraction of architectural elements across different programming languages. The tree-sitter approach generates parsers from grammar files, ensuring accurate parsing even as languages evolve. The extracted AST information feeds into downstream analysis agents that identify patterns, assess quality, and map architectural relationships.
The key advantage of using proper parsers rather than regex-based extraction is accuracy and completeness. Parsers understand language syntax and can correctly handle edge cases like nested structures, comments, and string literals that would confuse simpler extraction methods. This accuracy is essential for building reliable knowledge graphs and vector databases that subsequent analysis depends upon.
SPECIALIZED AGENT ARCHITECTURE
A robust code analysis system employs multiple specialized agents, each focused on a specific analytical domain. This multi-agent architecture distributes cognitive load, enables parallel processing, and allows each agent to develop deep expertise in its domain. The agents collaborate through shared knowledge stores and coordinated workflows.
The agent architecture includes several key roles. The code analysis agent focuses on understanding implementation details, identifying patterns, and assessing code quality. The architecture analysis agent examines system structure, component relationships, and architectural styles. The quality attribute agent evaluates non-functional requirements like performance, security, and maintainability. Domain analysis agents understand business concepts and map them to technical implementations. The requirements analysis agent traces requirements through the system. The repository agent analyzes version control history, issues, and development patterns. The knowledge graph agent maintains the semantic network of system entities and relationships. The vector database agent manages embedding-based retrieval. The report agent synthesizes findings into human-readable documentation. The test and quality agent examines test coverage and quality metrics. The operations agent understands deployment and runtime characteristics. The project agent tracks project management aspects. Technology-specific agents provide expertise in particular frameworks or platforms.
Here is an implementation framework for this multi-agent system:
from abc import ABC, abstractmethod
from typing import Dict, List, Any
import asyncio
from dataclasses import dataclass
from enum import Enum
class AgentType(Enum):
"""Types of specialized agents in the system."""
CODE_ANALYSIS = "code_analysis"
ARCHITECTURE_ANALYSIS = "architecture_analysis"
QUALITY_ATTRIBUTE = "quality_attribute"
DOMAIN_ANALYSIS = "domain_analysis"
REQUIREMENTS_ANALYSIS = "requirements_analysis"
REPOSITORY_ANALYSIS = "repository_analysis"
KNOWLEDGE_GRAPH = "knowledge_graph"
VECTOR_DATABASE = "vector_database"
REPORT_GENERATION = "report_generation"
TEST_QUALITY = "test_quality"
OPERATIONS = "operations"
PROJECT_MANAGEMENT = "project_management"
TECHNOLOGY_SPECIFIC = "technology_specific"
@dataclass
class AgentMessage:
"""Message passed between agents."""
sender: str
recipient: str
message_type: str
payload: Dict[str, Any]
correlation_id: str
timestamp: float
class BaseAgent(ABC):
"""
Abstract base class for all specialized agents.
"""
def __init__(self, agent_id: str, agent_type: AgentType, llm_client):
self.agent_id = agent_id
self.agent_type = agent_type
self.llm_client = llm_client
self.message_queue = asyncio.Queue()
self.knowledge_store = {}
self.is_running = False
async def start(self):
"""Start the agent's message processing loop."""
self.is_running = True
await self.process_messages()
async def stop(self):
"""Stop the agent."""
self.is_running = False
async def process_messages(self):
"""Process incoming messages from message queue."""
while self.is_running:
try:
message = await asyncio.wait_for(
self.message_queue.get(),
timeout=1.0
)
await self.handle_message(message)
except asyncio.TimeoutError:
continue
@abstractmethod
async def handle_message(self, message: AgentMessage):
"""
Handle incoming message. Must be implemented by subclasses.
Args:
message: Message to process
"""
pass
@abstractmethod
async def analyze(self, context: Dict[str, Any]) -> Dict[str, Any]:
"""
Perform agent-specific analysis.
Args:
context: Analysis context and input data
Returns:
Analysis results
"""
pass
async def send_message(self, recipient: str, message_type: str,
payload: Dict[str, Any], correlation_id: str):
"""
Send message to another agent.
Args:
recipient: ID of recipient agent
message_type: Type of message
payload: Message data
correlation_id: ID for tracking related messages
"""
message = AgentMessage(
sender=self.agent_id,
recipient=recipient,
message_type=message_type,
payload=payload,
correlation_id=correlation_id,
timestamp=time.time()
)
# In production, this would route through a message broker
await self._route_message(message)
async def _route_message(self, message: AgentMessage):
"""Route message to recipient agent."""
# This would integrate with message broker in production
pass
class CodeAnalysisAgent(BaseAgent):
"""
Agent specialized in analyzing code implementation details.
"""
def __init__(self, agent_id: str, llm_client, code_parser):
super().__init__(agent_id, AgentType.CODE_ANALYSIS, llm_client)
self.code_parser = code_parser
self.pattern_library = self._initialize_pattern_library()
async def handle_message(self, message: AgentMessage):
"""Handle incoming analysis requests."""
if message.message_type == "analyze_code":
result = await self.analyze(message.payload)
# Send results to requesting agent
await self.send_message(
recipient=message.sender,
message_type="analysis_complete",
payload=result,
correlation_id=message.correlation_id
)
async def analyze(self, context: Dict[str, Any]) -> Dict[str, Any]:
"""
Analyze code artifacts for patterns, quality, and structure.
Args:
context: Dictionary containing code files and analysis parameters
Returns:
Analysis results including patterns, metrics, and issues
"""
code_files = context.get('code_files', [])
analysis_depth = context.get('depth', 'standard')
results = {
'patterns': [],
'quality_metrics': {},
'issues': [],
'dependencies': []
}
for code_file in code_files:
# Parse code into AST
parsed = self.code_parser.parse_file(code_file['path'])
# Identify design patterns
patterns = await self._identify_patterns(parsed)
results['patterns'].extend(patterns)
# Calculate quality metrics
metrics = self._calculate_metrics(parsed)
results['quality_metrics'][code_file['path']] = metrics
# Detect code issues
issues = await self._detect_issues(parsed, code_file)
results['issues'].extend(issues)
# Extract dependencies
deps = self._extract_dependencies(parsed)
results['dependencies'].extend(deps)
# Perform cross-file analysis if deep analysis requested
if analysis_depth == 'deep':
cross_file_analysis = await self._cross_file_analysis(code_files)
results['cross_file_insights'] = cross_file_analysis
return results
async def _identify_patterns(self, parsed_code):
"""
Identify design patterns in parsed code using LLM.
"""
# Extract relevant code structures
classes = parsed_code.get('classes', [])
patterns_found = []
for cls in classes:
# Build prompt for pattern identification
prompt = f"""Analyze this class for design patterns:
Class: {cls['name']}
Methods: {', '.join([m['name'] for m in cls.get('methods', [])])}
Base Classes: {', '.join(cls.get('bases', []))}
Common patterns to check:
- Singleton
- Factory
- Observer
- Strategy
- Decorator
- Adapter
Identify any patterns present and explain the evidence."""
response = await self.llm_client.generate_async(
prompt,
max_tokens=500,
temperature=0.3
)
# Parse response and extract patterns
identified = self._parse_pattern_response(response)
if identified:
patterns_found.append({
'class': cls['name'],
'patterns': identified,
'file': parsed_code.get('filepath')
})
return patterns_found
def _calculate_metrics(self, parsed_code):
"""Calculate code quality metrics."""
metrics = {
'lines_of_code': 0,
'cyclomatic_complexity': 0,
'class_count': len(parsed_code.get('classes', [])),
'function_count': len(parsed_code.get('functions', [])),
'average_method_length': 0,
'max_method_complexity': 0
}
# Calculate from parsed data
complexity_values = []
for cls in parsed_code.get('classes', []):
for method in cls.get('methods', []):
complexity = method.get('complexity', 1)
complexity_values.append(complexity)
metrics['cyclomatic_complexity'] += complexity
metrics['max_method_complexity'] = max(
metrics['max_method_complexity'],
complexity
)
if complexity_values:
metrics['average_method_complexity'] = (
sum(complexity_values) / len(complexity_values)
)
return metrics
async def _detect_issues(self, parsed_code, code_file):
"""Detect potential code issues using LLM analysis."""
issues = []
# Check for high complexity methods
for cls in parsed_code.get('classes', []):
for method in cls.get('methods', []):
if method.get('complexity', 0) > 10:
issues.append({
'type': 'high_complexity',
'severity': 'warning',
'location': f"{cls['name']}.{method['name']}",
'message': f"Method has cyclomatic complexity of {method['complexity']}",
'file': code_file['path'],
'line': method.get('line_number')
})
# Use LLM to detect code smells
code_sample = code_file.get('content', '')[:2000] # Limit size
prompt = f"""Analyze this code for common code smells and issues:
{code_sample}
Identify:
1. Long methods (over 50 lines)
2. Large classes (over 500 lines)
3. Duplicate code
4. Poor naming
5. Missing error handling
6. Security concerns
List each issue found with location and severity."""
response = await self.llm_client.generate_async(
prompt,
max_tokens=800,
temperature=0.3
)
llm_issues = self._parse_issues_response(response)
issues.extend(llm_issues)
return issues
def _extract_dependencies(self, parsed_code):
"""Extract code dependencies from imports."""
dependencies = []
for imp in parsed_code.get('imports', []):
dependencies.append({
'module': imp.get('module'),
'type': imp.get('type'),
'file': parsed_code.get('filepath'),
'line': imp.get('line_number')
})
return dependencies
async def _cross_file_analysis(self, code_files):
"""Perform analysis across multiple files."""
# Detect duplicate code across files
duplicates = await self._detect_duplicates(code_files)
# Identify coupling between modules
coupling = self._analyze_coupling(code_files)
return {
'duplicates': duplicates,
'coupling': coupling
}
def _initialize_pattern_library(self):
"""Initialize library of known design patterns."""
return {
'singleton': {
'indicators': ['__new__', 'instance', '_instance'],
'structure': 'single_instance'
},
'factory': {
'indicators': ['create', 'build', 'make'],
'structure': 'object_creation'
},
'observer': {
'indicators': ['subscribe', 'notify', 'observer'],
'structure': 'event_handling'
}
}
class ArchitectureAnalysisAgent(BaseAgent):
"""
Agent specialized in analyzing system architecture.
"""
def __init__(self, agent_id: str, llm_client):
super().__init__(agent_id, AgentType.ARCHITECTURE_ANALYSIS, llm_client)
self.architectural_patterns = self._load_architectural_patterns()
async def handle_message(self, message: AgentMessage):
"""Handle architecture analysis requests."""
if message.message_type == "analyze_architecture":
result = await self.analyze(message.payload)
await self.send_message(
recipient=message.sender,
message_type="architecture_analysis_complete",
payload=result,
correlation_id=message.correlation_id
)
async def analyze(self, context: Dict[str, Any]) -> Dict[str, Any]:
"""
Analyze system architecture including layers, components, and patterns.
Args:
context: System information including code structure and dependencies
Returns:
Architecture analysis results
"""
system_structure = context.get('system_structure', {})
dependency_graph = context.get('dependency_graph', {})
results = {
'architectural_style': None,
'layers': [],
'components': [],
'patterns': [],
'quality_assessment': {}
}
# Identify architectural style
style = await self._identify_architectural_style(
system_structure,
dependency_graph
)
results['architectural_style'] = style
# Identify layers
layers = await self._identify_layers(system_structure)
results['layers'] = layers
# Identify major components
components = await self._identify_components(
system_structure,
dependency_graph
)
results['components'] = components
# Assess architectural quality
quality = await self._assess_architecture_quality(
style,
layers,
components,
dependency_graph
)
results['quality_assessment'] = quality
return results
async def _identify_architectural_style(self, structure, dependencies):
"""
Identify the primary architectural style of the system.
"""
# Analyze structure to determine style
prompt = f"""Analyze this system structure to identify its architectural style.
System has {len(structure.get('modules', []))} modules
Dependency relationships: {len(dependencies)} connections
Module organization:
{self._summarize_structure(structure)}
Common architectural styles:
- Layered Architecture
- Microservices Architecture
- Hexagonal Architecture (Ports and Adapters)
- Event-Driven Architecture
- Clean Architecture
- Monolithic Architecture
Identify the primary style and explain the evidence."""
response = await self.llm_client.generate_async(
prompt,
max_tokens=800,
temperature=0.3
)
return self._parse_style_response(response)
async def _identify_layers(self, structure):
"""Identify architectural layers in the system."""
layers = []
# Group modules by likely layer
module_groups = self._group_modules_by_layer(structure)
for layer_name, modules in module_groups.items():
layers.append({
'name': layer_name,
'modules': modules,
'responsibilities': self._infer_layer_responsibilities(
layer_name,
modules
)
})
return layers
async def _identify_components(self, structure, dependencies):
"""Identify major architectural components."""
components = []
# Use clustering to identify component boundaries
clusters = self._cluster_modules(structure, dependencies)
for cluster in clusters:
component = {
'name': cluster['name'],
'modules': cluster['modules'],
'interfaces': cluster.get('interfaces', []),
'dependencies': cluster.get('dependencies', [])
}
components.append(component)
return components
async def _assess_architecture_quality(self, style, layers,
components, dependencies):
"""Assess quality of the architecture."""
quality_metrics = {}
# Assess layer separation
quality_metrics['layer_separation'] = self._assess_layer_separation(
layers,
dependencies
)
# Assess component cohesion
quality_metrics['component_cohesion'] = self._assess_cohesion(
components
)
# Assess coupling
quality_metrics['coupling'] = self._assess_coupling(
components,
dependencies
)
# Check for architectural violations
violations = await self._detect_violations(
style,
layers,
dependencies
)
quality_metrics['violations'] = violations
# Calculate overall quality score
quality_metrics['overall_score'] = self._calculate_quality_score(
quality_metrics
)
return quality_metrics
def _load_architectural_patterns(self):
"""Load known architectural patterns for reference."""
return {
'layered': {
'layers': ['presentation', 'application', 'domain', 'infrastructure'],
'dependency_rules': 'top_down_only'
},
'hexagonal': {
'core': 'domain',
'adapters': ['primary', 'secondary'],
'dependency_rules': 'inward_only'
},
'microservices': {
'characteristics': ['independent_deployment', 'bounded_context'],
'communication': ['async', 'api']
}
}
def _summarize_structure(self, structure):
"""Create summary of system structure for prompt."""
summary = []
modules = structure.get('modules', [])[:20] # Limit for context
for module in modules:
summary.append(f"- {module['name']}: {module.get('type', 'unknown')}")
return '\n'.join(summary)
def _group_modules_by_layer(self, structure):
"""Group modules into likely architectural layers."""
groups = {
'presentation': [],
'application': [],
'domain': [],
'infrastructure': []
}
for module in structure.get('modules', []):
# Heuristic layer assignment based on module characteristics
module_name = module['name'].lower()
if any(term in module_name for term in ['view', 'controller', 'ui', 'web']):
groups['presentation'].append(module)
elif any(term in module_name for term in ['service', 'usecase', 'application']):
groups['application'].append(module)
elif any(term in module_name for term in ['model', 'entity', 'domain']):
groups['domain'].append(module)
elif any(term in module_name for term in ['repository', 'dao', 'database', 'api']):
groups['infrastructure'].append(module)
return groups
def _cluster_modules(self, structure, dependencies):
"""Cluster related modules into components."""
# Simple clustering based on dependency relationships
# In production, use more sophisticated clustering algorithms
clusters = []
visited = set()
modules = structure.get('modules', [])
for module in modules:
if module['name'] in visited:
continue
# Find connected modules
cluster = self._find_connected_modules(
module,
modules,
dependencies,
visited
)
if cluster:
clusters.append(cluster)
return clusters
def _find_connected_modules(self, start_module, all_modules,
dependencies, visited):
"""Find modules connected to start module."""
cluster_modules = [start_module]
visited.add(start_module['name'])
# Find directly connected modules
connected = [
dep for dep in dependencies.get(start_module['name'], [])
if dep not in visited
]
for conn in connected[:5]: # Limit cluster size
for module in all_modules:
if module['name'] == conn and conn not in visited:
cluster_modules.append(module)
visited.add(conn)
return {
'name': f"Component_{start_module['name']}",
'modules': cluster_modules
}
class MultiAgentOrchestrator:
"""
Orchestrates collaboration between specialized agents.
"""
def __init__(self):
self.agents = {}
self.message_broker = MessageBroker()
self.workflow_engine = WorkflowEngine()
def register_agent(self, agent: BaseAgent):
"""Register an agent with the orchestrator."""
self.agents[agent.agent_id] = agent
self.message_broker.register_agent(agent)
async def start_all_agents(self):
"""Start all registered agents."""
tasks = [agent.start() for agent in self.agents.values()]
await asyncio.gather(*tasks)
async def execute_analysis_workflow(self, system_path: str):
"""
Execute complete analysis workflow across all agents.
Args:
system_path: Path to system to analyze
Returns:
Comprehensive analysis results
"""
workflow = self.workflow_engine.create_workflow([
('parse_code', 'code_analysis'),
('analyze_architecture', 'architecture_analysis'),
('assess_quality', 'quality_attribute'),
('analyze_domain', 'domain_analysis'),
('generate_report', 'report_generation')
])
results = await workflow.execute({
'system_path': system_path
})
return results
class MessageBroker:
"""
Message broker for inter-agent communication.
"""
def __init__(self):
self.agent_queues = {}
def register_agent(self, agent: BaseAgent):
"""Register agent's message queue."""
self.agent_queues[agent.agent_id] = agent.message_queue
async def route_message(self, message: AgentMessage):
"""Route message to recipient agent."""
recipient_queue = self.agent_queues.get(message.recipient)
if recipient_queue:
await recipient_queue.put(message)
class WorkflowEngine:
"""
Engine for orchestrating multi-agent workflows.
"""
def create_workflow(self, steps: List[tuple]):
"""
Create workflow from step definitions.
Args:
steps: List of (step_name, agent_type) tuples
Returns:
Workflow instance
"""
return Workflow(steps)
class Workflow:
"""
Represents a multi-agent analysis workflow.
"""
def __init__(self, steps: List[tuple]):
self.steps = steps
self.results = {}
async def execute(self, initial_context: Dict[str, Any]):
"""Execute workflow steps sequentially."""
context = initial_context.copy()
for step_name, agent_type in self.steps:
# Execute step with current context
step_result = await self._execute_step(
step_name,
agent_type,
context
)
# Store results
self.results[step_name] = step_result
# Update context for next step
context.update(step_result)
return self.results
async def _execute_step(self, step_name, agent_type, context):
"""Execute single workflow step."""
# In production, this would send message to appropriate agent
# and wait for response
pass
This multi-agent architecture provides specialized expertise while enabling collaboration through message passing and shared knowledge stores. Each agent focuses on its domain, reducing the complexity any single agent must handle. The orchestrator coordinates workflows that span multiple agents, ensuring analysis proceeds in logical sequence and results are properly integrated.
The asynchronous message-based communication allows agents to work concurrently when analyzing independent aspects of the system. For example, the code analysis agent can examine implementation patterns while the architecture agent analyzes system structure, with both working in parallel until their results need to be synthesized.
NAMED ENTITY RECOGNITION FOR INFORMATION EXTRACTION
Named Entity Recognition plays a crucial role in extracting structured information from various document types encountered during system analysis. Different document types require different NER approaches because they contain distinct entity types and follow different conventions. Source code entities differ from requirement document entities, which differ from architecture documentation entities.
For source code, entities include classes, methods, variables, packages, and modules. For requirements documents, entities include features, user stories, acceptance criteria, and stakeholders. For architecture documents, entities include components, layers, connectors, and quality attributes. Each requires specialized extraction logic.
Here is an implementation of multi-domain NER for code analysis:
import re
from typing import List, Dict, Set
from dataclasses import dataclass
@dataclass
class Entity:
"""Represents a recognized entity."""
text: str
entity_type: str
start_position: int
end_position: int
confidence: float
context: str
metadata: Dict[str, any]
class CodeEntityRecognizer:
"""
NER for extracting entities from source code.
"""
def __init__(self, llm_client):
self.llm_client = llm_client
self.entity_patterns = self._initialize_patterns()
def _initialize_patterns(self):
"""Initialize regex patterns for common code entities."""
return {
'class_name': re.compile(r'class\s+([A-Z][a-zA-Z0-9_]*)\s*[:(]'),
'function_name': re.compile(r'def\s+([a-z_][a-zA-Z0-9_]*)\s*\('),
'variable_name': re.compile(r'([a-z_][a-zA-Z0-9_]*)\s*='),
'import_module': re.compile(r'import\s+([a-zA-Z0-9_.]+)'),
'decorator': re.compile(r'@([a-zA-Z_][a-zA-Z0-9_.]*)'),
}
def recognize_entities(self, code_text: str, filepath: str) -> List[Entity]:
"""
Extract entities from source code.
Args:
code_text: Source code text
filepath: Path to source file
Returns:
List of recognized entities
"""
entities = []
# Extract pattern-based entities
for entity_type, pattern in self.entity_patterns.items():
matches = pattern.finditer(code_text)
for match in matches:
entity = Entity(
text=match.group(1),
entity_type=entity_type,
start_position=match.start(),
end_position=match.end(),
confidence=0.95, # High confidence for pattern matches
context=self._extract_context(code_text, match.start()),
metadata={'filepath': filepath}
)
entities.append(entity)
# Use LLM for complex entity recognition
llm_entities = self._llm_entity_extraction(code_text, filepath)
entities.extend(llm_entities)
# Deduplicate and resolve conflicts
entities = self._deduplicate_entities(entities)
return entities
def _extract_context(self, text: str, position: int, window: int = 100):
"""Extract surrounding context for an entity."""
start = max(0, position - window)
end = min(len(text), position + window)
return text[start:end]
def _llm_entity_extraction(self, code_text: str, filepath: str) -> List[Entity]:
"""
Use LLM to extract complex entities that patterns might miss.
"""
# Limit code size for LLM processing
if len(code_text) > 3000:
code_text = code_text[:3000]
prompt = f"""Extract architectural entities from this code:
{code_text}
Identify:
1. Design pattern implementations (name the pattern and the classes involved)
2. Architectural layer indicators (presentation, domain, infrastructure)
3. Business domain concepts (entities, value objects, aggregates)
4. Integration points (external APIs, databases, message queues)
For each entity, provide:
- Entity text (exact name from code)
- Entity type
- Confidence (0.0 to 1.0)
- Brief context
Respond in JSON format."""
response = self.llm_client.generate(
prompt,
max_tokens=1000,
temperature=0.3
)
# Parse LLM response into entities
return self._parse_llm_entities(response, filepath)
def _parse_llm_entities(self, response: str, filepath: str) -> List[Entity]:
"""Parse LLM response into Entity objects."""
entities = []
try:
import json
data = json.loads(response)
for item in data.get('entities', []):
entity = Entity(
text=item['text'],
entity_type=item['type'],
start_position=-1, # Position unknown from LLM
end_position=-1,
confidence=item.get('confidence', 0.7),
context=item.get('context', ''),
metadata={'filepath': filepath, 'source': 'llm'}
)
entities.append(entity)
except json.JSONDecodeError:
# LLM response wasn't valid JSON
pass
return entities
def _deduplicate_entities(self, entities: List[Entity]) -> List[Entity]:
"""Remove duplicate entities, keeping highest confidence."""
seen = {}
for entity in entities:
key = (entity.text, entity.entity_type)
if key not in seen or entity.confidence > seen[key].confidence:
seen[key] = entity
return list(seen.values())
class RequirementsEntityRecognizer:
"""
NER for extracting entities from requirements documents.
"""
def __init__(self, llm_client):
self.llm_client = llm_client
def recognize_entities(self, requirements_text: str) -> List[Entity]:
"""
Extract entities from requirements document.
Args:
requirements_text: Requirements document text
Returns:
List of recognized entities
"""
entities = []
# Extract user stories
user_stories = self._extract_user_stories(requirements_text)
entities.extend(user_stories)
# Extract features
features = self._extract_features(requirements_text)
entities.extend(features)
# Extract actors/stakeholders
actors = self._extract_actors(requirements_text)
entities.extend(actors)
# Extract quality attributes
quality_attrs = self._extract_quality_attributes(requirements_text)
entities.extend(quality_attrs)
return entities
def _extract_user_stories(self, text: str) -> List[Entity]:
"""Extract user story entities."""
# Pattern: "As a [role], I want [feature] so that [benefit]"
pattern = re.compile(
r'As\s+a\s+([^,]+),\s+I\s+want\s+([^,]+)(?:,?\s+so\s+that\s+(.+))?',
re.IGNORECASE
)
entities = []
for match in pattern.finditer(text):
entity = Entity(
text=match.group(0),
entity_type='user_story',
start_position=match.start(),
end_position=match.end(),
confidence=0.9,
context=self._extract_context(text, match.start()),
metadata={
'role': match.group(1).strip(),
'feature': match.group(2).strip(),
'benefit': match.group(3).strip() if match.group(3) else None
}
)
entities.append(entity)
return entities
def _extract_features(self, text: str) -> List[Entity]:
"""Extract feature entities using LLM."""
prompt = f"""Extract feature descriptions from this requirements text:
{text[:2000]}
A feature is a distinct capability or functionality the system should provide.
For each feature, provide:
- Feature name
- Brief description
- Priority (if mentioned)
Respond in JSON format."""
response = self.llm_client.generate(
prompt,
max_tokens=800,
temperature=0.3
)
return self._parse_feature_response(response)
def _extract_actors(self, text: str) -> List[Entity]:
"""Extract actor/stakeholder entities."""
# Common actor indicators
actor_patterns = [
r'(?:user|customer|administrator|manager|operator|developer)',
r'(?:system|service|application|component)',
]
entities = []
for pattern_str in actor_patterns:
pattern = re.compile(pattern_str, re.IGNORECASE)
for match in pattern.finditer(text):
entity = Entity(
text=match.group(0),
entity_type='actor',
start_position=match.start(),
end_position=match.end(),
confidence=0.7,
context=self._extract_context(text, match.start()),
metadata={}
)
entities.append(entity)
return self._deduplicate_by_text(entities)
def _extract_quality_attributes(self, text: str) -> List[Entity]:
"""Extract quality attribute requirements."""
qa_keywords = {
'performance': ['fast', 'response time', 'throughput', 'latency'],
'security': ['secure', 'authentication', 'authorization', 'encryption'],
'scalability': ['scale', 'concurrent users', 'load'],
'reliability': ['available', 'uptime', 'fault tolerance'],
'usability': ['user-friendly', 'intuitive', 'easy to use']
}
entities = []
for qa_type, keywords in qa_keywords.items():
for keyword in keywords:
pattern = re.compile(r'\b' + re.escape(keyword) + r'\b', re.IGNORECASE)
for match in pattern.finditer(text):
entity = Entity(
text=match.group(0),
entity_type='quality_attribute',
start_position=match.start(),
end_position=match.end(),
confidence=0.8,
context=self._extract_context(text, match.start()),
metadata={'category': qa_type}
)
entities.append(entity)
return entities
def _extract_context(self, text: str, position: int, window: int = 150):
"""Extract surrounding context."""
start = max(0, position - window)
end = min(len(text), position + window)
return text[start:end]
def _deduplicate_by_text(self, entities: List[Entity]) -> List[Entity]:
"""Remove duplicate entities based on text."""
seen = set()
unique = []
for entity in entities:
if entity.text.lower() not in seen:
seen.add(entity.text.lower())
unique.append(entity)
return unique
def _parse_feature_response(self, response: str) -> List[Entity]:
"""Parse LLM response for features."""
entities = []
try:
import json
data = json.loads(response)
for feature in data.get('features', []):
entity = Entity(
text=feature['name'],
entity_type='feature',
start_position=-1,
end_position=-1,
confidence=0.75,
context=feature.get('description', ''),
metadata={
'description': feature.get('description'),
'priority': feature.get('priority')
}
)
entities.append(entity)
except json.JSONDecodeError:
pass
return entities
class ArchitectureDocumentEntityRecognizer:
"""
NER for extracting entities from architecture documentation.
"""
def __init__(self, llm_client):
self.llm_client = llm_client
def recognize_entities(self, doc_text: str) -> List[Entity]:
"""
Extract architectural entities from documentation.
Args:
doc_text: Architecture documentation text
Returns:
List of recognized entities
"""
entities = []
# Extract components
components = self._extract_components(doc_text)
entities.extend(components)
# Extract layers
layers = self._extract_layers(doc_text)
entities.extend(layers)
# Extract connectors/interfaces
connectors = self._extract_connectors(doc_text)
entities.extend(connectors)
# Extract quality attributes
quality_attrs = self._extract_quality_attributes(doc_text)
entities.extend(quality_attrs)
# Extract architectural patterns
patterns = self._extract_patterns(doc_text)
entities.extend(patterns)
return entities
def _extract_components(self, text: str) -> List[Entity]:
"""Extract component entities."""
prompt = f"""Extract architectural components from this documentation:
{text[:2500]}
Components are major structural elements like services, modules, subsystems.
For each component, provide:
- Component name
- Component type (service, module, subsystem, etc.)
- Responsibilities
Respond in JSON format."""
response = self.llm_client.generate(
prompt,
max_tokens=1000,
temperature=0.3
)
return self._parse_component_response(response)
def _extract_layers(self, text: str) -> List[Entity]:
"""Extract architectural layer entities."""
layer_keywords = [
'presentation layer',
'application layer',
'business layer',
'domain layer',
'data layer',
'infrastructure layer',
'persistence layer'
]
entities = []
for keyword in layer_keywords:
pattern = re.compile(r'\b' + re.escape(keyword) + r'\b', re.IGNORECASE)
for match in pattern.finditer(text):
entity = Entity(
text=match.group(0),
entity_type='architectural_layer',
start_position=match.start(),
end_position=match.end(),
confidence=0.9,
context=self._extract_context(text, match.start()),
metadata={}
)
entities.append(entity)
return entities
def _extract_connectors(self, text: str) -> List[Entity]:
"""Extract connector/interface entities."""
connector_patterns = [
r'(?:REST|HTTP|gRPC|SOAP)\s+(?:API|interface|endpoint)',
r'message\s+(?:queue|bus|broker)',
r'database\s+connection',
r'event\s+(?:bus|stream)'
]
entities = []
for pattern_str in connector_patterns:
pattern = re.compile(pattern_str, re.IGNORECASE)
for match in pattern.finditer(text):
entity = Entity(
text=match.group(0),
entity_type='connector',
start_position=match.start(),
end_position=match.end(),
confidence=0.85,
context=self._extract_context(text, match.start()),
metadata={}
)
entities.append(entity)
return entities
def _extract_quality_attributes(self, text: str) -> List[Entity]:
"""Extract quality attribute mentions."""
qa_patterns = {
'performance': r'(?:performance|throughput|latency|response\s+time)',
'scalability': r'(?:scalability|horizontal\s+scaling|load\s+balancing)',
'availability': r'(?:availability|uptime|fault\s+tolerance)',
'security': r'(?:security|authentication|authorization|encryption)',
'maintainability': r'(?:maintainability|modularity|testability)'
}
entities = []
for qa_type, pattern_str in qa_patterns.items():
pattern = re.compile(pattern_str, re.IGNORECASE)
for match in pattern.finditer(text):
entity = Entity(
text=match.group(0),
entity_type='quality_attribute',
start_position=match.start(),
end_position=match.end(),
confidence=0.8,
context=self._extract_context(text, match.start()),
metadata={'category': qa_type}
)
entities.append(entity)
return entities
def _extract_patterns(self, text: str) -> List[Entity]:
"""Extract architectural pattern mentions."""
pattern_names = [
'microservices',
'layered architecture',
'hexagonal architecture',
'event-driven',
'CQRS',
'event sourcing',
'service mesh',
'API gateway'
]
entities = []
for pattern_name in pattern_names:
pattern = re.compile(r'\b' + re.escape(pattern_name) + r'\b', re.IGNORECASE)
for match in pattern.finditer(text):
entity = Entity(
text=match.group(0),
entity_type='architectural_pattern',
start_position=match.start(),
end_position=match.end(),
confidence=0.9,
context=self._extract_context(text, match.start()),
metadata={'pattern_name': pattern_name}
)
entities.append(entity)
return entities
def _extract_context(self, text: str, position: int, window: int = 200):
"""Extract surrounding context."""
start = max(0, position - window)
end = min(len(text), position + window)
return text[start:end]
def _parse_component_response(self, response: str) -> List[Entity]:
"""Parse LLM response for components."""
entities = []
try:
import json
data = json.loads(response)
for comp in data.get('components', []):
entity = Entity(
text=comp['name'],
entity_type='component',
start_position=-1,
end_position=-1,
confidence=0.8,
context=comp.get('responsibilities', ''),
metadata={
'component_type': comp.get('type'),
'responsibilities': comp.get('responsibilities')
}
)
entities.append(entity)
except json.JSONDecodeError:
pass
return entities
These specialized NER implementations extract domain-specific entities from different document types. The combination of pattern-based extraction for well-defined entities and LLM-based extraction for complex or contextual entities provides comprehensive coverage. The extracted entities feed into knowledge graphs and vector databases where they become queryable architectural knowledge.
KNOWLEDGE STORAGE WITH RAG AND KNOWLEDGE GRAPHS
Effective storage and retrieval of extracted information requires dual approaches: vector databases for semantic similarity search and knowledge graphs for relationship-based queries. RAG systems use vector embeddings to find relevant information based on semantic similarity, while knowledge graphs explicitly model entities and their relationships for structured queries.
The integration of both approaches creates a powerful information retrieval system. Vector databases excel at finding conceptually similar code or documentation even when exact keywords don't match. Knowledge graphs excel at traversing relationships like finding all classes that implement a particular interface or all requirements that trace to a specific architectural component.
Here is an implementation of the dual storage system:
import numpy as np
from typing import List, Dict, Tuple
import json
class VectorDatabase:
"""
Vector database for semantic search using embeddings.
"""
def __init__(self, embedding_model, dimension=768):
"""
Initialize vector database.
Args:
embedding_model: Model for generating embeddings
dimension: Dimension of embedding vectors
"""
self.embedding_model = embedding_model
self.dimension = dimension
self.vectors = []
self.metadata = []
self.index = None
def add_document(self, text: str, metadata: Dict):
"""
Add document to vector database.
Args:
text: Document text to embed
metadata: Associated metadata
"""
# Generate embedding
embedding = self.embedding_model.embed(text)
# Store vector and metadata
self.vectors.append(embedding)
self.metadata.append({
'text': text,
**metadata
})
# Rebuild index if using approximate search
if len(self.vectors) % 100 == 0:
self._rebuild_index()
def add_code_chunk(self, code: str, file_path: str,
chunk_type: str, entities: List[str]):
"""
Add code chunk with metadata.
Args:
code: Code text
file_path: Source file path
chunk_type: Type of code chunk (class, function, etc.)
entities: Entities present in chunk
"""
metadata = {
'type': 'code',
'file_path': file_path,
'chunk_type': chunk_type,
'entities': entities,
'language': self._detect_language(file_path)
}
self.add_document(code, metadata)
def search(self, query: str, top_k: int = 5,
filters: Dict = None) -> List[Dict]:
"""
Search for similar documents.
Args:
query: Search query
top_k: Number of results to return
filters: Optional metadata filters
Returns:
List of matching documents with scores
"""
# Generate query embedding
query_embedding = self.embedding_model.embed(query)
# Calculate similarities
similarities = []
for i, vec in enumerate(self.vectors):
# Apply filters if specified
if filters and not self._matches_filters(self.metadata[i], filters):
continue
similarity = self._cosine_similarity(query_embedding, vec)
similarities.append((i, similarity))
# Sort by similarity and return top k
similarities.sort(key=lambda x: x[1], reverse=True)
top_results = similarities[:top_k]
results = []
for idx, score in top_results:
result = self.metadata[idx].copy()
result['similarity_score'] = score
results.append(result)
return results
def _cosine_similarity(self, vec1, vec2):
"""Calculate cosine similarity between vectors."""
dot_product = np.dot(vec1, vec2)
norm1 = np.linalg.norm(vec1)
norm2 = np.linalg.norm(vec2)
if norm1 == 0 or norm2 == 0:
return 0.0
return dot_product / (norm1 * norm2)
def _matches_filters(self, metadata: Dict, filters: Dict) -> bool:
"""Check if metadata matches filter criteria."""
for key, value in filters.items():
if key not in metadata:
return False
if isinstance(value, list):
if metadata[key] not in value:
return False
else:
if metadata[key] != value:
return False
return True
def _detect_language(self, file_path: str) -> str:
"""Detect programming language from file extension."""
extension_map = {
'.py': 'python',
'.java': 'java',
'.js': 'javascript',
'.go': 'go',
'.cpp': 'cpp',
'.cs': 'csharp'
}
import os
ext = os.path.splitext(file_path)[1]
return extension_map.get(ext, 'unknown')
def _rebuild_index(self):
"""Rebuild search index for efficiency."""
# In production, use FAISS or similar for approximate nearest neighbor
pass
class KnowledgeGraph:
"""
Knowledge graph for storing entities and relationships.
"""
def __init__(self):
self.nodes = {} # node_id -> node_data
self.edges = [] # list of (source, target, relationship_type, properties)
self.node_types = {} # node_type -> set of node_ids
self.relationship_index = {} # relationship_type -> list of edges
def add_node(self, node_id: str, node_type: str, properties: Dict):
"""
Add node to knowledge graph.
Args:
node_id: Unique identifier for node
node_type: Type of node (class, method, component, etc.)
properties: Node properties
"""
self.nodes[node_id] = {
'id': node_id,
'type': node_type,
'properties': properties
}
# Index by type
if node_type not in self.node_types:
self.node_types[node_type] = set()
self.node_types[node_type].add(node_id)
def add_edge(self, source_id: str, target_id: str,
relationship_type: str, properties: Dict = None):
"""
Add relationship between nodes.
Args:
source_id: Source node ID
target_id: Target node ID
relationship_type: Type of relationship
properties: Optional relationship properties
"""
edge = {
'source': source_id,
'target': target_id,
'type': relationship_type,
'properties': properties or {}
}
self.edges.append(edge)
# Index by relationship type
if relationship_type not in self.relationship_index:
self.relationship_index[relationship_type] = []
self.relationship_index[relationship_type].append(edge)
def add_class_node(self, class_name: str, file_path: str,
methods: List[str], base_classes: List[str]):
"""
Add class node with standard properties.
Args:
class_name: Name of the class
file_path: Source file path
methods: List of method names
base_classes: List of base class names
"""
node_id = f"class:{class_name}"
properties = {
'name': class_name,
'file_path': file_path,
'methods': methods,
'base_classes': base_classes
}
self.add_node(node_id, 'class', properties)
# Add inheritance relationships
for base_class in base_classes:
base_id = f"class:{base_class}"
self.add_edge(node_id, base_id, 'inherits_from')
# Add method relationships
for method in methods:
method_id = f"method:{class_name}.{method}"
self.add_node(method_id, 'method', {'name': method})
self.add_edge(node_id, method_id, 'has_method')
def add_component_node(self, component_name: str,
component_type: str, modules: List[str]):
"""
Add architectural component node.
Args:
component_name: Name of component
component_type: Type of component
modules: List of modules in component
"""
node_id = f"component:{component_name}"
properties = {
'name': component_name,
'type': component_type,
'modules': modules
}
self.add_node(node_id, 'component', properties)
# Add module relationships
for module in modules:
module_id = f"module:{module}"
self.add_edge(node_id, module_id, 'contains')
def query_nodes_by_type(self, node_type: str) -> List[Dict]:
"""
Query all nodes of a specific type.
Args:
node_type: Type of nodes to retrieve
Returns:
List of matching nodes
"""
node_ids = self.node_types.get(node_type, set())
return [self.nodes[nid] for nid in node_ids]
def query_relationships(self, relationship_type: str) -> List[Dict]:
"""
Query all relationships of a specific type.
Args:
relationship_type: Type of relationships to retrieve
Returns:
List of matching relationships
"""
return self.relationship_index.get(relationship_type, [])
def find_related_nodes(self, node_id: str,
relationship_type: str = None,
direction: str = 'outgoing') -> List[Dict]:
"""
Find nodes related to a given node.
Args:
node_id: Starting node ID
relationship_type: Optional filter by relationship type
direction: 'outgoing', 'incoming', or 'both'
Returns:
List of related nodes
"""
related = []
for edge in self.edges:
# Filter by relationship type if specified
if relationship_type and edge['type'] != relationship_type:
continue
# Check direction
if direction in ['outgoing', 'both'] and edge['source'] == node_id:
target_node = self.nodes.get(edge['target'])
if target_node:
related.append({
'node': target_node,
'relationship': edge['type'],
'direction': 'outgoing'
})
if direction in ['incoming', 'both'] and edge['target'] == node_id:
source_node = self.nodes.get(edge['source'])
if source_node:
related.append({
'node': source_node,
'relationship': edge['type'],
'direction': 'incoming'
})
return related
def find_path(self, start_id: str, end_id: str,
max_depth: int = 5) -> List[List[str]]:
"""
Find paths between two nodes.
Args:
start_id: Starting node ID
end_id: Ending node ID
max_depth: Maximum path length
Returns:
List of paths (each path is a list of node IDs)
"""
paths = []
visited = set()
def dfs(current_id, target_id, path, depth):
if depth > max_depth:
return
if current_id == target_id:
paths.append(path.copy())
return
if current_id in visited:
return
visited.add(current_id)
# Explore outgoing edges
for edge in self.edges:
if edge['source'] == current_id:
next_id = edge['target']
dfs(next_id, target_id, path + [next_id], depth + 1)
visited.remove(current_id)
dfs(start_id, end_id, [start_id], 0)
return paths
def export_to_json(self, filepath: str):
"""Export knowledge graph to JSON file."""
data = {
'nodes': list(self.nodes.values()),
'edges': self.edges
}
with open(filepath, 'w') as f:
json.dump(data, f, indent=2)
def import_from_json(self, filepath: str):
"""Import knowledge graph from JSON file."""
with open(filepath, 'r') as f:
data = json.load(f)
# Clear existing data
self.nodes = {}
self.edges = []
self.node_types = {}
self.relationship_index = {}
# Import nodes
for node in data.get('nodes', []):
self.add_node(
node['id'],
node['type'],
node['properties']
)
# Import edges
for edge in data.get('edges', []):
self.add_edge(
edge['source'],
edge['target'],
edge['type'],
edge.get('properties')
)
class HybridKnowledgeStore:
"""
Combines vector database and knowledge graph for comprehensive retrieval.
"""
def __init__(self, embedding_model):
self.vector_db = VectorDatabase(embedding_model)
self.knowledge_graph = KnowledgeGraph()
self.entity_to_vector_index = {} # Maps KG node IDs to vector indices
def add_code_artifact(self, code: str, file_path: str,
parsed_data: Dict):
"""
Add code artifact to both stores.
Args:
code: Source code text
file_path: Path to source file
parsed_data: Parsed AST data
"""
# Add to vector database
entities = []
for cls in parsed_data.get('classes', []):
entities.append(cls['name'])
# Add class to knowledge graph
self.knowledge_graph.add_class_node(
cls['name'],
file_path,
[m['name'] for m in cls.get('methods', [])],
cls.get('bases', [])
)
self.vector_db.add_code_chunk(
code,
file_path,
'module',
entities
)
# Create links between vector DB and knowledge graph
vector_index = len(self.vector_db.vectors) - 1
for entity in entities:
node_id = f"class:{entity}"
if node_id not in self.entity_to_vector_index:
self.entity_to_vector_index[node_id] = []
self.entity_to_vector_index[node_id].append(vector_index)
def hybrid_search(self, query: str, top_k: int = 5) -> List[Dict]:
"""
Perform hybrid search using both vector similarity and graph traversal.
Args:
query: Search query
top_k: Number of results
Returns:
Combined search results
"""
# Vector search for semantic similarity
vector_results = self.vector_db.search(query, top_k=top_k * 2)
# Extract entities from top vector results
entities_found = set()
for result in vector_results[:top_k]:
entities_found.update(result.get('entities', []))
# Graph traversal to find related entities
graph_results = []
for entity in list(entities_found)[:5]: # Limit to avoid explosion
node_id = f"class:{entity}"
if node_id in self.knowledge_graph.nodes:
related = self.knowledge_graph.find_related_nodes(
node_id,
direction='both'
)
graph_results.extend(related)
# Combine and rank results
combined = self._combine_results(vector_results, graph_results)
return combined[:top_k]
def _combine_results(self, vector_results: List[Dict],
graph_results: List[Dict]) -> List[Dict]:
"""Combine and rank results from both sources."""
combined = {}
# Add vector results with their scores
for i, result in enumerate(vector_results):
key = result.get('file_path', str(i))
combined[key] = {
**result,
'vector_score': result.get('similarity_score', 0),
'graph_score': 0,
'combined_score': result.get('similarity_score', 0)
}
# Enhance with graph results
for graph_result in graph_results:
node = graph_result['node']
file_path = node['properties'].get('file_path')
if file_path in combined:
# Boost score for graph-connected results
combined[file_path]['graph_score'] += 0.2
combined[file_path]['combined_score'] += 0.2
# Sort by combined score
results = list(combined.values())
results.sort(key=lambda x: x['combined_score'], reverse=True)
return results
This hybrid storage system leverages the strengths of both approaches. The vector database enables finding semantically similar code even when exact keywords don't match. The knowledge graph enables precise relationship queries like finding all classes that implement an interface or all components that depend on a particular module. The integration between the two allows queries that combine semantic similarity with structural relationships, producing more relevant and comprehensive results than either approach alone.
HIERARCHICAL SUMMARIZATION FOR CONTEXT MANAGEMENT
When analyzing large systems, the volume of information quickly exceeds the context window of even the largest language models. Hierarchical summarization addresses this by creating multi-level summaries where high-level summaries provide overview information and progressively detailed summaries are available on demand.
The approach works by first summarizing individual code files or components. These file-level summaries are then aggregated into module-level summaries. Module summaries aggregate into subsystem summaries, and subsystem summaries aggregate into system-level summaries. This creates a pyramid structure where the analyst can start with a high-level system overview and drill down into specific areas of interest without loading the entire codebase into context.
Here is an implementation of hierarchical summarization:
class HierarchicalSummarizer:
"""
Creates multi-level summaries of code artifacts.
"""
def __init__(self, llm_client):
self.llm_client = llm_client
self.summaries = {
'file': {},
'module': {},
'subsystem': {},
'system': None
}
def summarize_file(self, file_path: str, code: str,
parsed_data: Dict) -> Dict:
"""
Create file-level summary.
Args:
file_path: Path to file
code: Source code
parsed_data: Parsed AST data
Returns:
File summary
"""
# Extract key information
classes = [cls['name'] for cls in parsed_data.get('classes', [])]
functions = [fn['name'] for fn in parsed_data.get('functions', [])]
imports = [imp['module'] for imp in parsed_data.get('imports', [])]
# Build summarization prompt
prompt = f"""Summarize this code file concisely.
File: {file_path}
Classes: {', '.join(classes) if classes else 'None'}
Functions: {', '.join(functions) if functions else 'None'}
Key Imports: {', '.join(imports[:10])}
Code excerpt:
{code[:1500]}
Provide a 2-3 sentence summary covering:
1. Primary purpose of this file
2. Key classes or functions it provides
3. Its likely role in the larger system"""
response = self.llm_client.generate(
prompt,
max_tokens=200,
temperature=0.3
)
summary = {
'file_path': file_path,
'summary_text': response.strip(),
'classes': classes,
'functions': functions,
'key_imports': imports[:10],
'lines_of_code': len(code.split('\n'))
}
self.summaries['file'][file_path] = summary
return summary
def summarize_module(self, module_name: str,
file_summaries: List[Dict]) -> Dict:
"""
Create module-level summary from file summaries.
Args:
module_name: Name of module
file_summaries: List of file-level summaries
Returns:
Module summary
"""
# Aggregate information from file summaries
all_classes = []
all_functions = []
total_loc = 0
for file_sum in file_summaries:
all_classes.extend(file_sum.get('classes', []))
all_functions.extend(file_sum.get('functions', []))
total_loc += file_sum.get('lines_of_code', 0)
# Build module summary prompt
file_summaries_text = '\n'.join([
f"- {fs['file_path']}: {fs['summary_text']}"
for fs in file_summaries[:10] # Limit to prevent context overflow
])
prompt = f"""Summarize this module based on its file summaries.
Module: {module_name}
Number of files: {len(file_summaries)}
Total lines of code: {total_loc}
Key classes: {', '.join(all_classes[:15])}
File summaries:
{file_summaries_text}
Provide a 3-4 sentence summary covering:
1. Overall purpose of this module
2. Main responsibilities and functionality
3. Key architectural patterns or design approaches
4. How it likely fits into the larger system"""
response = self.llm_client.generate(
prompt,
max_tokens=300,
temperature=0.3
)
summary = {
'module_name': module_name,
'summary_text': response.strip(),
'file_count': len(file_summaries),
'total_lines': total_loc,
'key_classes': all_classes[:20],
'key_functions': all_functions[:20]
}
self.summaries['module'][module_name] = summary
return summary
def summarize_subsystem(self, subsystem_name: str,
module_summaries: List[Dict]) -> Dict:
"""
Create subsystem-level summary from module summaries.
Args:
subsystem_name: Name of subsystem
module_summaries: List of module summaries
Returns:
Subsystem summary
"""
total_files = sum(ms.get('file_count', 0) for ms in module_summaries)
total_loc = sum(ms.get('total_lines', 0) for ms in module_summaries)
module_summaries_text = '\n'.join([
f"- {ms['module_name']}: {ms['summary_text']}"
for ms in module_summaries[:8]
])
prompt = f"""Summarize this subsystem based on its module summaries.
Subsystem: {subsystem_name}
Number of modules: {len(module_summaries)}
Total files: {total_files}
Total lines of code: {total_loc}
Module summaries:
{module_summaries_text}
Provide a 4-5 sentence summary covering:
1. High-level purpose of this subsystem
2. Major functional areas it covers
3. Architectural approach and key patterns
4. Dependencies and integration points
5. Quality characteristics (if evident)"""
response = self.llm_client.generate(
prompt,
max_tokens=400,
temperature=0.3
)
summary = {
'subsystem_name': subsystem_name,
'summary_text': response.strip(),
'module_count': len(module_summaries),
'total_files': total_files,
'total_lines': total_loc
}
self.summaries['subsystem'][subsystem_name] = summary
return summary
def summarize_system(self, system_name: str,
subsystem_summaries: List[Dict]) -> Dict:
"""
Create system-level summary from subsystem summaries.
Args:
system_name: Name of system
subsystem_summaries: List of subsystem summaries
Returns:
System summary
"""
total_modules = sum(
ss.get('module_count', 0) for ss in subsystem_summaries
)
total_files = sum(
ss.get('total_files', 0) for ss in subsystem_summaries
)
total_loc = sum(
ss.get('total_lines', 0) for ss in subsystem_summaries
)
subsystem_summaries_text = '\n'.join([
f"- {ss['subsystem_name']}: {ss['summary_text']}"
for ss in subsystem_summaries
])
prompt = f"""Summarize this entire system based on its subsystem summaries.
System: {system_name}
Number of subsystems: {len(subsystem_summaries)}
Total modules: {total_modules}
Total files: {total_files}
Total lines of code: {total_loc}
Subsystem summaries:
{subsystem_summaries_text}
Provide a comprehensive 5-7 sentence summary covering:
1. Overall system purpose and domain
2. Major subsystems and their roles
3. Architectural style and key patterns
4. Technology stack and frameworks
5. Integration approach between subsystems
6. Notable quality attributes or design decisions"""
response = self.llm_client.generate(
prompt,
max_tokens=600,
temperature=0.3
)
summary = {
'system_name': system_name,
'summary_text': response.strip(),
'subsystem_count': len(subsystem_summaries),
'total_modules': total_modules,
'total_files': total_files,
'total_lines': total_loc
}
self.summaries['system'] = summary
return summary
def get_contextual_summary(self, target_entity: str,
max_context_tokens: int = 2000) -> str:
"""
Get contextual summary for a specific entity, including relevant
higher-level context without exceeding token budget.
Args:
target_entity: Entity to get context for (file, module, etc.)
max_context_tokens: Maximum tokens for context
Returns:
Contextual summary text
"""
context_parts = []
estimated_tokens = 0
# Start with system-level summary if available
if self.summaries['system']:
system_text = self.summaries['system']['summary_text']
tokens = len(system_text.split()) * 1.3 # Rough token estimate
if estimated_tokens + tokens < max_context_tokens:
context_parts.append(f"System Overview: {system_text}")
estimated_tokens += tokens
# Add relevant subsystem summary
relevant_subsystem = self._find_containing_subsystem(target_entity)
if relevant_subsystem and relevant_subsystem in self.summaries['subsystem']:
subsys_text = self.summaries['subsystem'][relevant_subsystem]['summary_text']
tokens = len(subsys_text.split()) * 1.3
if estimated_tokens + tokens < max_context_tokens:
context_parts.append(
f"Subsystem ({relevant_subsystem}): {subsys_text}"
)
estimated_tokens += tokens
# Add relevant module summary
relevant_module = self._find_containing_module(target_entity)
if relevant_module and relevant_module in self.summaries['module']:
module_text = self.summaries['module'][relevant_module]['summary_text']
tokens = len(module_text.split()) * 1.3
if estimated_tokens + tokens < max_context_tokens:
context_parts.append(
f"Module ({relevant_module}): {module_text}"
)
estimated_tokens += tokens
# Add file-level summary if target is a file
if target_entity in self.summaries['file']:
file_text = self.summaries['file'][target_entity]['summary_text']
tokens = len(file_text.split()) * 1.3
if estimated_tokens + tokens < max_context_tokens:
context_parts.append(
f"File ({target_entity}): {file_text}"
)
estimated_tokens += tokens
return '\n\n'.join(context_parts)
def _find_containing_subsystem(self, entity: str) -> str:
"""Find which subsystem contains the entity."""
# Simple heuristic based on path structure
# In production, use proper module hierarchy tracking
if '/' in entity:
parts = entity.split('/')
if len(parts) > 2:
return parts[0]
return None
def _find_containing_module(self, entity: str) -> str:
"""Find which module contains the entity."""
if '/' in entity:
parts = entity.split('/')
if len(parts) > 1:
return '/'.join(parts[:-1])
return None
This hierarchical approach ensures that analysts always have appropriate context without overwhelming the model's context window. When examining a specific class, the system provides file-level, module-level, and system-level context to situate that class within the broader architecture. This contextual awareness significantly improves the quality of analysis and recommendations.
REQUIREMENT TO CODE TRACEABILITY
Establishing bidirectional traceability between requirements, architecture, code, and tests provides critical insights into system coverage and implementation completeness. This traceability enables impact analysis when requirements change, verification that all requirements have implementations, and identification of orphaned code that doesn't trace to any requirement.
The implementation requires extracting requirements from various sources including user stories, feature specifications, and acceptance criteria. These requirements are then mapped to architectural components that implement them. Architectural components map to code artifacts including classes, modules, and services. Code artifacts map to test cases that verify their behavior. The complete traceability chain connects business requirements through implementation to verification.
Here is an implementation of the traceability system:
class TraceabilityManager:
"""
Manages traceability links between requirements, architecture, code, and tests.
"""
def __init__(self, knowledge_graph: KnowledgeGraph):
self.knowledge_graph = knowledge_graph
self.trace_links = {
'requirement_to_architecture': [],
'architecture_to_code': [],
'code_to_test': [],
'requirement_to_code': [], # Derived from above
'requirement_to_test': [] # Derived from above
}
def add_requirement(self, req_id: str, req_text: str,
req_type: str, priority: str):
"""
Add requirement to traceability system.
Args:
req_id: Unique requirement identifier
req_text: Requirement text
req_type: Type of requirement (functional, non-functional, etc.)
priority: Requirement priority
"""
self.knowledge_graph.add_node(
req_id,
'requirement',
{
'text': req_text,
'type': req_type,
'priority': priority
}
)
def link_requirement_to_architecture(self, req_id: str,
component_id: str,
llm_client):
"""
Create traceability link from requirement to architectural component.
Args:
req_id: Requirement identifier
component_id: Component identifier
llm_client: LLM client for validation
"""
# Validate the link makes sense
req_node = self.knowledge_graph.nodes.get(req_id)
comp_node = self.knowledge_graph.nodes.get(component_id)
if not req_node or not comp_node:
return False
# Use LLM to verify the mapping is reasonable
validation = self._validate_requirement_mapping(
req_node,
comp_node,
llm_client
)
if validation['is_valid']:
self.knowledge_graph.add_edge(
req_id,
component_id,
'implemented_by',
{'confidence': validation['confidence']}
)
self.trace_links['requirement_to_architecture'].append({
'requirement': req_id,
'component': component_id,
'confidence': validation['confidence']
})
return True
return False
def link_architecture_to_code(self, component_id: str,
code_artifact_id: str):
"""
Link architectural component to code artifact.
Args:
component_id: Component identifier
code_artifact_id: Code artifact identifier (class, module, etc.)
"""
self.knowledge_graph.add_edge(
component_id,
code_artifact_id,
'realized_by'
)
self.trace_links['architecture_to_code'].append({
'component': component_id,
'code_artifact': code_artifact_id
})
def link_code_to_test(self, code_artifact_id: str, test_id: str,
coverage_percentage: float):
"""
Link code artifact to test case.
Args:
code_artifact_id: Code artifact identifier
test_id: Test case identifier
coverage_percentage: Test coverage percentage
"""
self.knowledge_graph.add_edge(
code_artifact_id,
test_id,
'tested_by',
{'coverage': coverage_percentage}
)
self.trace_links['code_to_test'].append({
'code_artifact': code_artifact_id,
'test': test_id,
'coverage': coverage_percentage
})
def compute_derived_traces(self):
"""
Compute derived traceability links by following chains.
"""
# Requirement to Code (via Architecture)
for req_to_arch in self.trace_links['requirement_to_architecture']:
req_id = req_to_arch['requirement']
comp_id = req_to_arch['component']
# Find code artifacts for this component
for arch_to_code in self.trace_links['architecture_to_code']:
if arch_to_code['component'] == comp_id:
code_id = arch_to_code['code_artifact']
self.trace_links['requirement_to_code'].append({
'requirement': req_id,
'code_artifact': code_id,
'via_component': comp_id
})
# Requirement to Test (via Code)
for req_to_code in self.trace_links['requirement_to_code']:
req_id = req_to_code['requirement']
code_id = req_to_code['code_artifact']
# Find tests for this code
for code_to_test in self.trace_links['code_to_test']:
if code_to_test['code_artifact'] == code_id:
test_id = code_to_test['test']
self.trace_links['requirement_to_test'].append({
'requirement': req_id,
'test': test_id,
'via_code': code_id,
'coverage': code_to_test['coverage']
})
def find_untraceable_requirements(self) -> List[str]:
"""
Find requirements with no implementation trace.
Returns:
List of requirement IDs with no implementation
"""
all_requirements = set(
self.knowledge_graph.node_types.get('requirement', set())
)
traced_requirements = set(
link['requirement']
for link in self.trace_links['requirement_to_architecture']
)
return list(all_requirements - traced_requirements)
def find_orphaned_code(self) -> List[str]:
"""
Find code artifacts not traced to any requirement.
Returns:
List of code artifact IDs with no requirement trace
"""
all_code = set()
for node_type in ['class', 'module', 'function']:
all_code.update(
self.knowledge_graph.node_types.get(node_type, set())
)
traced_code = set(
link['code_artifact']
for link in self.trace_links['requirement_to_code']
)
return list(all_code - traced_code)
def find_untested_code(self) -> List[str]:
"""
Find code artifacts with no test coverage.
Returns:
List of code artifact IDs with no tests
"""
all_code = set()
for node_type in ['class', 'module', 'function']:
all_code.update(
self.knowledge_graph.node_types.get(node_type, set())
)
tested_code = set(
link['code_artifact']
for link in self.trace_links['code_to_test']
)
return list(all_code - tested_code)
def generate_traceability_report(self) -> Dict:
"""
Generate comprehensive traceability report.
Returns:
Report with traceability metrics and gaps
"""
self.compute_derived_traces()
total_requirements = len(
self.knowledge_graph.node_types.get('requirement', set())
)
traced_requirements = len(set(
link['requirement']
for link in self.trace_links['requirement_to_architecture']
))
total_code = sum(
len(self.knowledge_graph.node_types.get(t, set()))
for t in ['class', 'module', 'function']
)
traced_code = len(set(
link['code_artifact']
for link in self.trace_links['requirement_to_code']
))
tested_code = len(set(
link['code_artifact']
for link in self.trace_links['code_to_test']
))
report = {
'requirements': {
'total': total_requirements,
'traced': traced_requirements,
'untraced': total_requirements - traced_requirements,
'trace_percentage': (
(traced_requirements / total_requirements * 100)
if total_requirements > 0 else 0
)
},
'code': {
'total': total_code,
'traced_to_requirements': traced_code,
'orphaned': total_code - traced_code,
'tested': tested_code,
'untested': total_code - tested_code,
'requirement_trace_percentage': (
(traced_code / total_code * 100)
if total_code > 0 else 0
),
'test_coverage_percentage': (
(tested_code / total_code * 100)
if total_code > 0 else 0
)
},
'gaps': {
'untraceable_requirements': self.find_untraceable_requirements(),
'orphaned_code': self.find_orphaned_code(),
'untested_code': self.find_untested_code()
}
}
return report
def _validate_requirement_mapping(self, req_node: Dict,
comp_node: Dict,
llm_client) -> Dict:
"""
Validate that requirement to component mapping is reasonable.
Args:
req_node: Requirement node data
comp_node: Component node data
llm_client: LLM client for validation
Returns:
Validation result with confidence score
"""
prompt = f"""Evaluate if this requirement-to-component mapping is valid.
Requirement: {req_node['properties']['text']}
Requirement Type: {req_node['properties']['type']}
Component: {comp_node['properties']['name']}
Component Type: {comp_node['properties'].get('type', 'unknown')}
Component Responsibilities: {comp_node['properties'].get('responsibilities', 'unknown')}
Is this component a reasonable implementation of this requirement?
Provide:
1. is_valid: true/false
2. confidence: 0.0 to 1.0
3. reasoning: brief explanation
Respond in JSON format."""
response = llm_client.generate(
prompt,
max_tokens=300,
temperature=0.3
)
try:
import json
result = json.loads(response)
return result
except json.JSONDecodeError:
return {
'is_valid': False,
'confidence': 0.0,
'reasoning': 'Failed to parse validation response'
}
This traceability system provides visibility into the completeness and correctness of the implementation. It identifies requirements that have no implementation, code that serves no documented requirement, and code that lacks test coverage. These insights guide development priorities and quality improvement efforts.
MULTI-MODEL LLM SUPPORT AND GPU ACCELERATION
Supporting both local and remote LLMs with various GPU architectures provides flexibility in deployment scenarios. Some organizations prefer local models for data privacy and cost control. Others prefer remote models for access to the latest capabilities without infrastructure management. The system should support both seamlessly.
GPU acceleration is essential for acceptable performance with local models. Different hardware platforms require different acceleration libraries. NVIDIA GPUs use CUDA. AMD GPUs use ROCm. Intel GPUs use their oneAPI toolkit. Apple Silicon uses Metal Performance Shaders. The system must detect available hardware and configure acceleration appropriately.
Here is an implementation supporting multiple LLM backends and GPU platforms:
import os
import platform
from abc import ABC, abstractmethod
from typing import Optional, Dict, List
import asyncio
class LLMBackend(ABC):
"""
Abstract base for LLM backends.
"""
@abstractmethod
def generate(self, prompt: str, max_tokens: int = 1000,
temperature: float = 0.7) -> str:
"""Generate text from prompt."""
pass
@abstractmethod
async def generate_async(self, prompt: str, max_tokens: int = 1000,
temperature: float = 0.7) -> str:
"""Async text generation."""
pass
@abstractmethod
def embed(self, text: str) -> List[float]:
"""Generate embedding for text."""
pass
class LocalLLMBackend(LLMBackend):
"""
Backend for local LLM inference with GPU acceleration.
"""
def __init__(self, model_path: str, device: str = 'auto'):
"""
Initialize local LLM backend.
Args:
model_path: Path to model files
device: Device to use ('cuda', 'rocm', 'mps', 'cpu', 'auto')
"""
self.model_path = model_path
self.device = self._detect_device() if device == 'auto' else device
self.model = None
self.tokenizer = None
self._initialize_model()
def _detect_device(self) -> str:
"""
Detect available GPU acceleration.
Returns:
Device string ('cuda', 'rocm', 'mps', or 'cpu')
"""
# Check for NVIDIA CUDA
try:
import torch
if torch.cuda.is_available():
return 'cuda'
except ImportError:
pass
# Check for AMD ROCm
try:
import torch
if hasattr(torch.version, 'hip') and torch.version.hip is not None:
return 'rocm'
except (ImportError, AttributeError):
pass
# Check for Apple Metal
if platform.system() == 'Darwin':
try:
import torch
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
return 'mps'
except (ImportError, AttributeError):
pass
# Fallback to CPU
return 'cpu'
def _initialize_model(self):
"""Initialize model with appropriate acceleration."""
if self.device == 'cuda':
self._initialize_cuda_model()
elif self.device == 'rocm':
self._initialize_rocm_model()
elif self.device == 'mps':
self._initialize_mps_model()
else:
self._initialize_cpu_model()
def _initialize_cuda_model(self):
"""Initialize model with NVIDIA CUDA acceleration."""
try:
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tokenizer = AutoTokenizer.from_pretrained(self.model_path)
self.model = AutoModelForCausalLM.from_pretrained(
self.model_path,
torch_dtype=torch.float16, # Use FP16 for efficiency
device_map='auto' # Automatically distribute across GPUs
)
print(f"Model loaded on CUDA device: {torch.cuda.get_device_name(0)}")
except Exception as e:
print(f"Failed to initialize CUDA model: {e}")
self._initialize_cpu_model()
def _initialize_rocm_model(self):
"""Initialize model with AMD ROCm acceleration."""
try:
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tokenizer = AutoTokenizer.from_pretrained(self.model_path)
self.model = AutoModelForCausalLM.from_pretrained(
self.model_path,
torch_dtype=torch.float16,
device_map='auto'
)
# Move to ROCm device
if torch.cuda.is_available():
self.model = self.model.to('cuda')
print("Model loaded with ROCm acceleration")
except Exception as e:
print(f"Failed to initialize ROCm model: {e}")
self._initialize_cpu_model()
def _initialize_mps_model(self):
"""Initialize model with Apple Metal Performance Shaders."""
try:
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tokenizer = AutoTokenizer.from_pretrained(self.model_path)
self.model = AutoModelForCausalLM.from_pretrained(
self.model_path,
torch_dtype=torch.float32 # MPS may not support FP16 for all ops
)
# Move to MPS device
self.model = self.model.to('mps')
print("Model loaded with Apple Metal acceleration")
except Exception as e:
print(f"Failed to initialize MPS model: {e}")
self._initialize_cpu_model()
def _initialize_cpu_model(self):
"""Initialize model for CPU inference."""
try:
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tokenizer = AutoTokenizer.from_pretrained(self.model_path)
self.model = AutoModelForCausalLM.from_pretrained(
self.model_path,
torch_dtype='auto'
)
print("Model loaded on CPU (no GPU acceleration available)")
except Exception as e:
print(f"Failed to initialize CPU model: {e}")
raise
def generate(self, prompt: str, max_tokens: int = 1000,
temperature: float = 0.7) -> str:
"""
Generate text from prompt.
Args:
prompt: Input prompt
max_tokens: Maximum tokens to generate
temperature: Sampling temperature
Returns:
Generated text
"""
import torch
inputs = self.tokenizer(prompt, return_tensors='pt')
# Move inputs to same device as model
if self.device != 'cpu':
inputs = {k: v.to(self.model.device) for k, v in inputs.items()}
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=max_tokens,
temperature=temperature,
do_sample=temperature > 0,
pad_token_id=self.tokenizer.eos_token_id
)
generated_text = self.tokenizer.decode(
outputs[0],
skip_special_tokens=True
)
# Remove the prompt from output
if generated_text.startswith(prompt):
generated_text = generated_text[len(prompt):].strip()
return generated_text
async def generate_async(self, prompt: str, max_tokens: int = 1000,
temperature: float = 0.7) -> str:
"""
Async text generation (runs in thread pool to avoid blocking).
"""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None,
self.generate,
prompt,
max_tokens,
temperature
)
def embed(self, text: str) -> List[float]:
"""
Generate embedding for text.
Args:
text: Input text
Returns:
Embedding vector
"""
import torch
inputs = self.tokenizer(
text,
return_tensors='pt',
truncation=True,
max_length=512
)
if self.device != 'cpu':
inputs = {k: v.to(self.model.device) for k, v in inputs.items()}
with torch.no_grad():
outputs = self.model(**inputs, output_hidden_states=True)
# Use last hidden state, mean pooling
hidden_states = outputs.hidden_states[-1]
embedding = torch.mean(hidden_states[0], dim=0)
return embedding.cpu().numpy().tolist()
class RemoteLLMBackend(LLMBackend):
"""
Backend for remote LLM APIs (OpenAI, Anthropic, etc.).
"""
def __init__(self, api_key: str, model_name: str,
provider: str = 'openai'):
"""
Initialize remote LLM backend.
Args:
api_key: API key for service
model_name: Model identifier
provider: Service provider ('openai', 'anthropic', etc.)
"""
self.api_key = api_key
self.model_name = model_name
self.provider = provider
self.client = self._initialize_client()
def _initialize_client(self):
"""Initialize API client based on provider."""
if self.provider == 'openai':
try:
import openai
openai.api_key = self.api_key
return openai
except ImportError:
raise ImportError("OpenAI package not installed")
elif self.provider == 'anthropic':
try:
import anthropic
return anthropic.Anthropic(api_key=self.api_key)
except ImportError:
raise ImportError("Anthropic package not installed")
else:
raise ValueError(f"Unsupported provider: {self.provider}")
def generate(self, prompt: str, max_tokens: int = 1000,
temperature: float = 0.7) -> str:
"""Generate text using remote API."""
if self.provider == 'openai':
response = self.client.ChatCompletion.create(
model=self.model_name,
messages=[{'role': 'user', 'content': prompt}],
max_tokens=max_tokens,
temperature=temperature
)
return response.choices[0].message.content
elif self.provider == 'anthropic':
message = self.client.messages.create(
model=self.model_name,
max_tokens=max_tokens,
temperature=temperature,
messages=[{'role': 'user', 'content': prompt}]
)
return message.content[0].text
async def generate_async(self, prompt: str, max_tokens: int = 1000,
temperature: float = 0.7) -> str:
"""Async text generation."""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None,
self.generate,
prompt,
max_tokens,
temperature
)
def embed(self, text: str) -> List[float]:
"""Generate embedding using remote API."""
if self.provider == 'openai':
response = self.client.Embedding.create(
model='text-embedding-ada-002',
input=text
)
return response.data[0].embedding
else:
raise NotImplementedError(
f"Embedding not implemented for {self.provider}"
)
class LLMManager:
"""
Manages multiple LLM backends and routes requests appropriately.
"""
def __init__(self):
self.backends = {}
self.default_backend = None
def register_backend(self, name: str, backend: LLMBackend,
is_default: bool = False):
"""
Register an LLM backend.
Args:
name: Backend identifier
backend: LLM backend instance
is_default: Whether this is the default backend
"""
self.backends[name] = backend
if is_default or self.default_backend is None:
self.default_backend = name
def get_backend(self, name: Optional[str] = None) -> LLMBackend:
"""
Get LLM backend by name.
Args:
name: Backend name (uses default if None)
Returns:
LLM backend instance
"""
backend_name = name or self.default_backend
if backend_name not in self.backends:
raise ValueError(f"Unknown backend: {backend_name}")
return self.backends[backend_name]
def generate(self, prompt: str, backend: Optional[str] = None,
**kwargs) -> str:
"""
Generate text using specified or default backend.
Args:
prompt: Input prompt
backend: Backend name (optional)
**kwargs: Additional generation parameters
Returns:
Generated text
"""
llm = self.get_backend(backend)
return llm.generate(prompt, **kwargs)
async def generate_async(self, prompt: str,
backend: Optional[str] = None,
**kwargs) -> str:
"""Async text generation."""
llm = self.get_backend(backend)
return await llm.generate_async(prompt, **kwargs)
This multi-backend architecture allows the system to use local models for privacy-sensitive analysis while leveraging remote models for tasks requiring the latest capabilities. The automatic GPU detection and configuration ensures optimal performance across different hardware platforms without manual configuration.
CONCLUSION
Building effective multi-agent AI systems for analyzing large database architectures and codebases requires careful attention to numerous technical challenges. The patterns and practices described in this article address the fundamental issues of limited context memory, bias, and hallucinations through systematic approaches.
Human-in-the-loop validation ensures critical decisions receive expert oversight. Lightweight prompt decomposition breaks complex analysis into manageable pieces. Advanced reasoning strategies including self-reflection, chain-of-thought, and tree-of-thought improve analytical accuracy. Multi-language parsers extract structured information from diverse codebases. Specialized agents provide focused expertise while collaborating through shared knowledge stores. Named entity recognition extracts domain-specific information from various document types. Dual storage using vector databases and knowledge graphs enables both semantic and structural queries. Hierarchical summarization manages context windows effectively. Requirement-to-code traceability provides visibility into implementation completeness. Multi-model LLM support with GPU acceleration offers deployment flexibility.
Together, these patterns create a comprehensive framework for building analysis tools that provide accurate, reliable insights into complex software systems. The implementations provided demonstrate practical approaches to each challenge, offering starting points for developers building such systems. While each pattern addresses specific issues, their real power emerges from integration into cohesive systems where they complement and reinforce each other.
The field of AI-assisted code analysis continues evolving rapidly. New language models with larger context windows, better reasoning capabilities, and improved factual accuracy will enhance these systems. However, the fundamental architectural patterns described here will remain relevant because they address inherent challenges in analyzing large, complex systems regardless of the specific AI technologies employed. By combining human expertise with AI capabilities through well-designed patterns, we can build tools that significantly improve our ability to understand, maintain, and evolve large software systems.
