feat: Add actual AI/ML capabilities with LLM, semantic embeddings, and reinforcement learning

- Add semantic_encoder.py: CodeBERT-based semantic code embeddings with fallback TF-IDF
- Add llm_generator.py: LLM-based code generation (CodeGen, CodeT5, StarCoder, etc.)
- Add learning_module.py: Reinforcement learning + pattern learning from validation feedback
- Update enhanced_ml_model.py:
- Add last_retrieval property
- Add learning module strategy selection
- Add semantic similarity enhancement
- Add LLM generation as strategy option
- Record validation feedback to learning module
- Update config.py: Add MLConfig options for LLM, semantic encoder, learning module
- Update pipeline.py: Pass new ML config options
- Update requirements.txt: Add torch, transformers, sentence-transformers, accelerate
- Recreate ml_generation_model.py with MLModelConfig, NameNormalizer, RetrievalInfo

Key AI/ML capabilities:
1. LLM code generation with few-shot UVM examples
2. Semantic code embeddings for intelligent similarity
3. Reinforcement learning (Q-learning) from validation feedback
4. Pattern learning from success/failure patterns
5. Auto-improving generation strategy selection
6. Graceful fallback when torch/transformers not available

requirements.txt

@@ -7,3 +7,9 @@ gunicorn>=23.0
 
 numpy>=1.21.0
 scikit-learn>=1.0.0
+scipy>=1.7.0
+
+torch>=2.0.0
+transformers>=4.35.0
+sentence-transformers>=2.2.0
+accelerate>=0.24.0

src/config.py

@@ -86,15 +86,30 @@ class AutoTrainConfig(BaseModel):
 
 
 class MLConfig(BaseModel):
-    """Configuration for ML-augmented generation."""
+    """Configuration for AI/ML-augmented generation with actual learning capabilities."""
     enabled: bool = False
-    model_type: str = Field(default="template", pattern=r"^(template|ml|hybrid)$")
+    model_type: str = Field(default="template", pattern=r"^(template|ml|hybrid|llm|semantic)$")
     similarity_threshold: float = Field(default=0.75, ge=0.0, le=1.0)
     auto_learn: bool = True
     index_path: Optional[str] = None
     top_k_retrieval: int = Field(default=3, ge=1, le=10)
     fallback_to_templates: bool = True
 
+    use_llm: bool = True
+    llm_model_name: Optional[str] = None
+    llm_max_tokens: int = Field(default=1024, ge=64, le=4096)
+    llm_temperature: float = Field(default=0.2, ge=0.0, le=1.0)
+    llm_use_few_shot: bool = True
+
+    use_semantic_encoder: bool = True
+    semantic_model_name: str = "microsoft/codebert-base"
+
+    use_learning: bool = True
+    learning_storage_path: Optional[str] = None
+    learning_rate: float = Field(default=0.1, ge=0.001, le=1.0)
+    reinforcement_discount: float = Field(default=0.9, ge=0.0, le=1.0)
+    exploration_epsilon: float = Field(default=0.05, ge=0.0, le=0.5)
+
 
 class PipelineConfig(BaseModel):
     generation: GenerationConfig = GenerationConfig()

src/models/enhanced_ml_model.py

@@ -1,18 +1,20 @@
 """
-Industry-level enhanced ML generation model with:
-- Multi-strategy retrieval
+Industry-level AI/ML generation model with:
+- LLM-based code generation (CodeGen, CodeT5, StarCoder)
+- Semantic code embeddings for intelligent similarity
+- Reinforcement learning from validation feedback
+- Multi-strategy retrieval (protocol-first, semantic, text)
 - Spec-aware adaptation
 - Code validation
 - Multi-level fallback
 - Comprehensive reporting
 
-This model ensures output quality through:
-1. Protocol-first retrieval
-2. Coverage-aware selection
-3. Full adaptation with signal/register mapping
-4. Pre-validation before writing
-5. Automatic fallback to templates if issues found
-6. Detailed generation reports
+This model uses actual AI/ML:
+1. Neural semantic embeddings (CodeBERT) for similarity
+2. LLM generation (CodeGen, CodeT5) for actual code generation
+3. Reinforcement learning that learns from validation feedback
+4. Pattern learning from success/failure patterns
+5. Auto-improving generation strategies
 """
 
 from __future__ import annotations
@@ -43,6 +45,16 @@ from src.models.spec_adapter import (
 from src.models.similarity_index import SimilarityIndex, get_global_index
 from src.models.template_model import TemplateModel
 
+try:
+    from src.models.semantic_encoder import SemanticCodeEncoder, SemanticEmbedding
+    from src.models.llm_generator import LLMCodeGenerator, LLMGenerationResult
+    from src.models.learning_module import LearningModule, ValidationFeedback
+
+    ML_MODULES_AVAILABLE = True
+except ImportError as e:
+    logger.warning("Advanced ML modules not available: %s", e)
+    ML_MODULES_AVAILABLE = False
+
 logger = logging.getLogger("uvmgen")
 
 
@@ -50,9 +62,12 @@ class GenerationSource(Enum):
     RETRIEVAL_HIGH_CONF = "retrieval_high_confidence"
     RETRIEVAL_MEDIUM_CONF = "retrieval_medium_confidence"
     RETRIEVAL_LOW_CONF = "retrieval_low_confidence"
+    LLM_GENERATION = "llm_generation"
+    LLM_FALLBACK = "llm_fallback"
     TEMPLATE_FALLBACK = "template_fallback"
     BLENDED = "blended"
     HYBRID = "hybrid"
+    LEARNING_IMPROVED = "learning_improved"
 
 
 @dataclass
@@ -117,15 +132,21 @@ class RetrievalCandidate:
 
 class EnhancedMLGenerationModel(GenerationModel):
     """
-    Industry-level enhanced ML generation model.
-
-    Key features:
-    1. Multi-strategy retrieval (protocol-first, then similarity)
-    2. Spec-aware adaptation with signal/register mapping
-    3. Pre-validation before output
-    4. Multi-level fallback strategies
-    5. Comprehensive reporting and audit trail
-    6. Coverage-aware candidate selection
+    Industry-level AI/ML generation model with actual learning capabilities.
+
+    Key AI/ML features:
+    1. LLM-based code generation (CodeGen, CodeT5, StarCoder)
+    2. Semantic code embeddings (CodeBERT) for intelligent similarity
+    3. Reinforcement learning from validation feedback
+    4. Pattern learning from success/failure patterns
+    5. Multi-strategy retrieval with intelligent selection
+    6. Auto-improving generation strategies
+
+    Traditional features:
+    - Spec-aware adaptation with signal/register mapping
+    - Pre-validation before output
+    - Multi-level fallback strategies
+    - Comprehensive reporting and audit trail
     """
 
     def __init__(
@@ -135,6 +156,11 @@ class EnhancedMLGenerationModel(GenerationModel):
         index: Optional[SimilarityIndex] = None,
         templates_dir: Optional[str] = None,
         strict_validation: bool = True,
+        use_llm: bool = True,
+        use_semantic_encoder: bool = True,
+        use_learning: bool = True,
+        llm_model_name: Optional[str] = None,
+        learning_storage_path: Optional[str] = None,
     ):
         super().__init__(name)
         self.config = config or MLModelConfig()
@@ -144,6 +170,27 @@ class EnhancedMLGenerationModel(GenerationModel):
         self._strict_validation = strict_validation
         self._metadata: Dict[str, Any] = {}
         self._last_result: Optional[GenerationResult] = None
+        self._last_retrieval: Optional[Any] = None
+
+        self._use_llm = use_llm and ML_MODULES_AVAILABLE
+        self._use_semantic = use_semantic_encoder and ML_MODULES_AVAILABLE
+        self._use_learning = use_learning and ML_MODULES_AVAILABLE
+
+        self._llm_generator: Optional[LLMCodeGenerator] = None
+        self._semantic_encoder: Optional[SemanticCodeEncoder] = None
+        self._learning_module: Optional[LearningModule] = None
+
+        if self._use_llm:
+            self._llm_generator = LLMCodeGenerator(model_name=llm_model_name)
+            logger.info("LLM generator enabled: %s", llm_model_name or "default")
+
+        if self._use_semantic:
+            self._semantic_encoder = SemanticCodeEncoder()
+            logger.info("Semantic encoder enabled")
+
+        if self._use_learning:
+            self._learning_module = LearningModule(storage_path=learning_storage_path)
+            logger.info("Learning module enabled")
 
     @property
     def index(self) -> SimilarityIndex:
@@ -163,6 +210,23 @@ class EnhancedMLGenerationModel(GenerationModel):
             )
         return self._template_model
 
+    @property
+    def last_retrieval(self) -> Optional[Any]:
+        """Get information about the last retrieval operation."""
+        from src.models.ml_generation_model import RetrievalInfo
+
+        if self._last_retrieval is not None:
+            return self._last_retrieval
+
+        if self._last_result is not None:
+            return RetrievalInfo(
+                used_similarity=(self._last_result.similar_specs_found > 0),
+                similar_specs=self._last_result.similar_specs_found,
+                best_score=self._last_result.best_match_score,
+            )
+
+        return RetrievalInfo(used_similarity=False, similar_specs=0, best_score=0.0)
+
     def train(self, specs: List[DesignSpec]) -> Dict[str, Any]:
         """Train the model by adding specs to the similarity index."""
         from src.features.extractors import RichSpecFeatureExtractor
@@ -226,20 +290,21 @@ class EnhancedMLGenerationModel(GenerationModel):
         extra_seqs: Optional[List[str]] = None,
     ) -> Dict[str, str]:
         """
-        Generate testbench with full validation and fallback.
-
-        Workflow:
-        1. Extract rich features
-        2. Search for similar specs
-        3. For each candidate:
-           - Create adaptation plan
-           - Pre-validate
-           - Score
-        4. Select best candidate or fallback
-        5. Adapt best candidate
-        6. Validate output
-        7. If validation fails, fallback to templates
-        8. If auto_learn, add to index
+        Generate testbench with AI/ML-powered generation and fallback.
+
+        AI/ML Workflow:
+        1. Use learning module to select best generation strategy
+        2. Try semantic similarity search (if semantic encoder available)
+        3. Try LLM-based code generation (if LLM available)
+        4. Try traditional retrieval-based generation
+        5. Fallback to templates
+        6. Record validation feedback to learning module
+        7. Auto-learn from successful generation
+
+        Traditional features:
+        - Spec-aware adaptation
+        - Pre-validation before writing
+        - Multi-level fallback
         """
         if not self._is_trained:
             self.train([])
@@ -249,12 +314,41 @@ class EnhancedMLGenerationModel(GenerationModel):
         query_fv = extractor.extract(spec)
         query_dict = self._spec_to_dict(spec)
 
+        protocol = query_dict.get("protocol", "unknown")
+
+        available_strategies = ["retrieval"]
+        if self._use_llm and self._llm_generator:
+            available_strategies.append("llm")
+        available_strategies.append("template")
+
+        selected_strategy = "retrieval"
+        strategy_confidence = 0.5
+
+        if self._use_learning and self._learning_module:
+            selected_strategy, strategy_confidence = (
+                self._learning_module.select_best_generation_strategy(
+                    spec_dict=query_dict,
+                    file_type="testbench",
+                    available_sources=available_strategies,
+                )
+            )
+            logger.info(
+                "Learning module selected strategy: '%s' (confidence: %.2f)",
+                selected_strategy,
+                strategy_confidence,
+            )
+
         similar = self.index.search(
             query_fv,
             top_k=self.config.top_k_retrieval,
             min_similarity=0.3,
         )
 
+        if self._use_semantic and self._semantic_encoder and similar:
+            similar = self._enhance_with_semantic_similarity(
+                similar, query_dict
+            )
+
         logger.info(
             "Enhanced ML generation: found %d similar specs, best score: %.3f",
             len(similar), similar[0].similarity if similar else 0.0
@@ -262,21 +356,29 @@ class EnhancedMLGenerationModel(GenerationModel):
 
         result: Optional[GenerationResult] = None
 
-        if similar and similar[0].similarity >= self.config.similarity_threshold:
+        if selected_strategy == "llm" and self._use_llm and self._llm_generator:
+            logger.info("Trying LLM-based generation (selected by learning module)")
+            result = self._try_llm_generation(query_dict, spec, cfg)
+
+        if result is None and similar and similar[0].similarity >= self.config.similarity_threshold:
             result = self._try_retrieval_generation(
                 similar, query_fv, query_dict, spec, cfg
             )
 
+        if result is None and self._use_llm and self._llm_generator:
+            logger.info("Trying LLM-based generation as fallback")
+            result = self._try_llm_generation(query_dict, spec, cfg)
+
         if (
             result is None
             or (self._strict_validation and not result.passed)
             and self.config.fallback_to_templates
         ):
             if result is None:
-                logger.info("No valid retrieval candidate, falling back to templates")
+                logger.info("No valid ML/LLM candidate, falling back to templates")
             else:
                 logger.warning(
-                    "Retrieval-based generation failed validation (errors: %d), falling back to templates",
+                    "LLM/retrieval generation failed validation (errors: %d), falling back to templates",
                     result.validation_report.total_errors if result.validation_report else 0
                 )
             result = self._generate_with_fallback(spec, cfg, extra_seqs, result)
@@ -284,6 +386,15 @@ class EnhancedMLGenerationModel(GenerationModel):
         if result is None:
             raise RuntimeError("All generation strategies failed")
 
+        if self._use_learning and self._learning_module and result.validation_report:
+            logger.info("Recording validation feedback to learning module")
+            self._learning_module.record_feedback(
+                design_name=spec.design_name,
+                generation_source=result.source.value,
+                spec_dict=query_dict,
+                validation_results=result.validation_report.to_dict(),
+            )
+
         if self.config.auto_learn and result.passed:
             self._learn_from_result(result, query_fv, query_dict)
 
@@ -292,6 +403,186 @@ class EnhancedMLGenerationModel(GenerationModel):
 
         return result.generated_files
 
+    def _enhance_with_semantic_similarity(
+        self,
+        similar: List[Any],
+        query_dict: Dict[str, Any],
+    ) -> List[Any]:
+        """Enhance similarity scores using semantic code embeddings."""
+        if not self._semantic_encoder or not self._semantic_encoder.is_available():
+            return similar
+
+        try:
+            query_text = self._spec_dict_to_text(query_dict)
+            query_emb = self._semantic_encoder.encode(
+                text=query_text,
+                embedding_type="spec",
+                metadata=query_dict,
+            )
+
+            for item in similar:
+                spec_text = self._spec_dict_to_text(item.spec_dict)
+                cand_emb = self._semantic_encoder.encode(
+                    text=spec_text,
+                    embedding_type="spec",
+                    metadata=item.spec_dict,
+                )
+
+                semantic_sim = self._semantic_encoder.similarity(query_emb, cand_emb)
+
+                original_sim = item.similarity
+                item.similarity = (original_sim * 0.6) + (semantic_sim * 0.4)
+
+                logger.debug(
+                    "Semantic enhancement: original=%.3f, semantic=%.3f, combined=%.3f",
+                    original_sim, semantic_sim, item.similarity
+                )
+
+            similar = sorted(similar, key=lambda x: x.similarity, reverse=True)
+
+        except Exception as e:
+            logger.warning("Semantic similarity enhancement failed: %s", e)
+
+        return similar
+
+    def _spec_dict_to_text(self, spec_dict: Dict[str, Any]) -> str:
+        """Convert spec dict to text for semantic encoding."""
+        parts = []
+        parts.append(f"design: {spec_dict.get('design_name', 'unknown')}")
+        parts.append(f"protocol: {spec_dict.get('protocol', 'unknown')}")
+
+        signals = spec_dict.get("signals", [])
+        if signals:
+            signal_names = [s.get("name", "") for s in signals if isinstance(s, dict)]
+            parts.append(f"signals: {', '.join(signal_names[:20])}")
+
+        registers = spec_dict.get("registers", [])
+        if registers:
+            reg_names = [r.get("name", "") for r in registers if isinstance(r, dict)]
+            parts.append(f"registers: {', '.join(reg_names[:10])}")
+
+        features = spec_dict.get("features", [])
+        if features:
+            parts.append(f"features: {', '.join(features[:10])}")
+
+        return " | ".join(parts)
+
+    def _try_llm_generation(
+        self,
+        query_dict: Dict[str, Any],
+        spec: DesignSpec,
+        cfg: PipelineConfig,
+    ) -> Optional[GenerationResult]:
+        """
+        Try LLM-based code generation.
+
+        This uses actual AI/ML:
+        1. LLM (CodeGen, CodeT5, etc.) generates SystemVerilog code
+        2. Uses few-shot examples for UVM patterns
+        3. Validates generated code
+        4. Falls back to templates if needed
+        """
+        if not self._llm_generator:
+            return None
+
+        design_name = spec.design_name.lower()
+
+        file_types_to_generate = [
+            "driver",
+            "monitor",
+            "agent",
+        ]
+
+        generated_files: Dict[str, str] = {}
+        llm_results: Dict[str, LLMGenerationResult] = {}
+        all_warnings: List[str] = []
+        avg_confidence = 0.0
+
+        for file_type in file_types_to_generate:
+            try:
+                llm_result = self._llm_generator.generate(
+                    spec_dict=query_dict,
+                    file_type=file_type,
+                    use_few_shot=True,
+                    max_tokens=1024,
+                    temperature=0.2,
+                )
+
+                llm_results[file_type] = llm_result
+                avg_confidence += llm_result.confidence
+                all_warnings.extend(llm_result.warnings)
+
+                file_name = f"{design_name}_{file_type}.sv"
+                generated_files[file_name] = llm_result.generated_code
+
+                logger.info(
+                    "LLM generated %s (confidence: %.2f, tokens: %d)",
+                    file_name,
+                    llm_result.confidence,
+                    llm_result.tokens_generated,
+                )
+
+            except Exception as e:
+                logger.warning("LLM generation failed for %s: %s", file_type, e)
+                all_warnings.append(f"LLM generation failed for {file_type}: {e}")
+
+        if not generated_files:
+            logger.warning("LLM generated no files, falling back")
+            return None
+
+        if llm_results:
+            avg_confidence /= len(llm_results)
+
+        try:
+            template_files = self.template_model.predict(spec, cfg)
+            template_contents: Dict[str, str] = {}
+            for fname, fpath in template_files.items():
+                try:
+                    template_contents[fname] = Path(fpath).read_text(encoding="utf-8")
+                except Exception:
+                    pass
+
+            for fname, content in template_contents.items():
+                if fname not in generated_files:
+                    generated_files[fname] = content
+
+        except Exception as e:
+            logger.warning("Could not fill missing files from templates: %s", e)
+
+        validator = CodeValidator()
+        val_report = validator.validate_files(generated_files, query_dict)
+
+        total_errors = val_report.total_errors
+        total_warnings = val_report.total_warnings + len(all_warnings)
+
+        passed = val_report.overall_passed
+        if self._strict_validation:
+            passed = passed and (total_errors == 0)
+
+        generation_source = GenerationSource.LLM_GENERATION
+        if avg_confidence < 0.5:
+            generation_source = GenerationSource.LLM_FALLBACK
+
+        result = GenerationResult(
+            design_name=spec.design_name,
+            source=generation_source,
+            passed=passed,
+            generated_files=generated_files,
+            validation_report=val_report,
+            adaptation_plan=None,
+            similar_specs_found=0,
+            best_match_score=avg_confidence,
+            files_from_retrieval=[],
+            files_from_template=list(template_contents.keys()) if "template_contents" in dir() else [],
+            warnings=all_warnings + [
+                f"LLM confidence: {avg_confidence:.2f}",
+                f"LLM warnings: {len(all_warnings)}",
+            ],
+            errors=[f"LLM errors: {total_errors}"] if total_errors > 0 else [],
+        )
+
+        return result
+
     def _try_retrieval_generation(
         self,
         similar: List[Any],

src/models/learning_module.py

@@ -0,0 +1,572 @@
+import logging
+from typing import List, Dict, Any, Optional, Tuple
+from dataclasses import dataclass, field
+from collections import defaultdict
+import json
+import os
+from datetime import datetime
+
+logger = logging.getLogger("uvmgen.ml.learning")
+
+
+@dataclass
+class ValidationFeedback:
+    design_name: str
+    file_name: str
+    file_type: str
+    passed: bool
+    errors: List[str]
+    warnings: List[str]
+    score: float
+    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "design_name": self.design_name,
+            "file_name": self.file_name,
+            "file_type": self.file_type,
+            "passed": self.passed,
+            "errors": self.errors,
+            "warnings": self.warnings,
+            "score": self.score,
+            "timestamp": self.timestamp,
+            "metadata": self.metadata,
+        }
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> "ValidationFeedback":
+        return cls(
+            design_name=d.get("design_name", "unknown"),
+            file_name=d.get("file_name", "unknown"),
+            file_type=d.get("file_type", "unknown"),
+            passed=d.get("passed", False),
+            errors=d.get("errors", []),
+            warnings=d.get("warnings", []),
+            score=d.get("score", 0.0),
+            timestamp=d.get("timestamp", datetime.now().isoformat()),
+            metadata=d.get("metadata", {}),
+        )
+
+
+@dataclass
+class GenerationHistory:
+    design_name: str
+    generation_source: str
+    spec_hash: str
+    feedback_list: List[ValidationFeedback]
+    success_rate: float = 0.0
+    avg_score: float = 0.0
+    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "design_name": self.design_name,
+            "generation_source": self.generation_source,
+            "spec_hash": self.spec_hash,
+            "feedback_list": [f.to_dict() for f in self.feedback_list],
+            "success_rate": self.success_rate,
+            "avg_score": self.avg_score,
+            "timestamp": self.timestamp,
+        }
+
+
+class PatternLearner:
+    def __init__(self):
+        self._error_patterns: Dict[str, int] = defaultdict(int)
+        self._success_patterns: Dict[str, int] = defaultdict(int)
+        self._file_type_stats: Dict[str, Dict[str, Any]] = defaultdict(
+            lambda: {"success": 0, "total": 0, "errors": defaultdict(int)}
+        )
+        self._protocol_stats: Dict[str, Dict[str, Any]] = defaultdict(
+            lambda: {"success": 0, "total": 0}
+        )
+
+    def record_error(self, error_msg: str, file_type: str = "unknown"):
+        patterns = self._extract_patterns(error_msg)
+        for p in patterns:
+            self._error_patterns[p] += 1
+        self._file_type_stats[file_type]["errors"][error_msg[:100]] += 1
+
+    def record_success(self, file_type: str = "unknown", protocol: str = "unknown"):
+        self._file_type_stats[file_type]["success"] += 1
+        self._file_type_stats[file_type]["total"] += 1
+        self._protocol_stats[protocol]["success"] += 1
+        self._protocol_stats[protocol]["total"] += 1
+
+    def record_attempt(self, file_type: str = "unknown", protocol: str = "unknown"):
+        self._file_type_stats[file_type]["total"] += 1
+        self._protocol_stats[protocol]["total"] += 1
+
+    def _extract_patterns(self, text: str) -> List[str]:
+        import re
+
+        patterns = []
+
+        uvm_patterns = [
+            (r"uvm_fatal", "uvm_fatal"),
+            (r"uvm_error", "uvm_error"),
+            (r"uvm_component_utils", "missing_uvm_macro"),
+            (r"uvm_object_utils", "missing_uvm_macro"),
+            (r"build_phase", "phase_issue"),
+            (r"connect_phase", "phase_issue"),
+            (r"run_phase", "phase_issue"),
+        ]
+
+        for pattern, name in uvm_patterns:
+            if re.search(pattern, text, re.IGNORECASE):
+                patterns.append(name)
+
+        syntax_patterns = [
+            (r"missing.*semicolon", "missing_semicolon"),
+            (r"unbalanced.*parenthes", "unbalanced_parentheses"),
+            (r"unbalanced.*brace", "unbalanced_braces"),
+            (r"unbalanced.*bracket", "unbalanced_brackets"),
+            (r"mismatch.*begin", "mismatched_blocks"),
+            (r"syntax error", "syntax_error"),
+        ]
+
+        for pattern, name in syntax_patterns:
+            if re.search(pattern, text, re.IGNORECASE):
+                patterns.append(name)
+
+        if not patterns:
+            patterns.append("unknown_error")
+
+        return patterns
+
+    def get_common_errors(self, top_n: int = 10) -> List[Tuple[str, int]]:
+        sorted_errors = sorted(
+            self._error_patterns.items(),
+            key=lambda x: x[1],
+            reverse=True,
+        )
+        return sorted_errors[:top_n]
+
+    def get_file_type_success_rate(self, file_type: str) -> float:
+        stats = self._file_type_stats.get(file_type, {})
+        total = stats.get("total", 0)
+        if total == 0:
+            return 0.5
+        return stats.get("success", 0) / total
+
+    def get_protocol_success_rate(self, protocol: str) -> float:
+        stats = self._protocol_stats.get(protocol, {})
+        total = stats.get("total", 0)
+        if total == 0:
+            return 0.5
+        return stats.get("success", 0) / total
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "error_patterns": dict(self._error_patterns),
+            "file_type_stats": {
+                ft: {
+                    "success": s["success"],
+                    "total": s["total"],
+                    "errors": dict(s["errors"]),
+                }
+                for ft, s in self._file_type_stats.items()
+            },
+            "protocol_stats": dict(self._protocol_stats),
+        }
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> "PatternLearner":
+        learner = cls()
+        learner._error_patterns = defaultdict(int, d.get("error_patterns", {}))
+
+        for ft, s in d.get("file_type_stats", {}).items():
+            learner._file_type_stats[ft] = {
+                "success": s.get("success", 0),
+                "total": s.get("total", 0),
+                "errors": defaultdict(int, s.get("errors", {})),
+            }
+
+        for proto, s in d.get("protocol_stats", {}).items():
+            learner._protocol_stats[proto] = {
+                "success": s.get("success", 0),
+                "total": s.get("total", 0),
+            }
+
+        return learner
+
+
+class ReinforcementLearner:
+    def __init__(self, learning_rate: float = 0.1, discount_factor: float = 0.9):
+        self._learning_rate = learning_rate
+        self._discount_factor = discount_factor
+        self._q_values: Dict[str, float] = defaultdict(lambda: 0.5)
+        self._visit_counts: Dict[str, int] = defaultdict(int)
+
+    def _get_state_key(
+        self,
+        protocol: str,
+        file_type: str,
+        generation_source: str,
+    ) -> str:
+        return f"{protocol}:{file_type}:{generation_source}"
+
+    def get_action_value(
+        self,
+        protocol: str,
+        file_type: str,
+        generation_source: str,
+    ) -> float:
+        key = self._get_state_key(protocol, file_type, generation_source)
+        return self._q_values[key]
+
+    def update(
+        self,
+        protocol: str,
+        file_type: str,
+        generation_source: str,
+        reward: float,
+    ):
+        key = self._get_state_key(protocol, file_type, generation_source)
+        old_value = self._q_values[key]
+        self._visit_counts[key] += 1
+        self._q_values[key] = (
+            old_value + self._learning_rate * (reward - old_value)
+        )
+
+    def select_best_action(
+        self,
+        protocol: str,
+        file_type: str,
+        available_sources: List[str],
+        epsilon: float = 0.1,
+    ) -> Tuple[str, float]:
+        import random
+
+        if random.random() < epsilon and len(available_sources) > 1:
+            chosen = random.choice(available_sources)
+            return chosen, self.get_action_value(protocol, file_type, chosen)
+
+        best_source = available_sources[0]
+        best_value = -1.0
+
+        for source in available_sources:
+            value = self.get_action_value(protocol, file_type, source)
+            if value > best_value:
+                best_value = value
+                best_source = source
+
+        return best_source, best_value
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "learning_rate": self._learning_rate,
+            "discount_factor": self._discount_factor,
+            "q_values": dict(self._q_values),
+            "visit_counts": dict(self._visit_counts),
+        }
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> "ReinforcementLearner":
+        learner = cls(
+            learning_rate=d.get("learning_rate", 0.1),
+            discount_factor=d.get("discount_factor", 0.9),
+        )
+        learner._q_values = defaultdict(lambda: 0.5)
+        learner._q_values.update(d.get("q_values", {}))
+        learner._visit_counts = defaultdict(int)
+        learner._visit_counts.update(d.get("visit_counts", {}))
+        return learner
+
+
+class LearningModule:
+    def __init__(self, storage_path: Optional[str] = None):
+        self._storage_path = storage_path
+        self._pattern_learner = PatternLearner()
+        self._rl_learner = ReinforcementLearner()
+        self._history: List[GenerationHistory] = []
+        self._total_generations = 0
+        self._successful_generations = 0
+
+        if storage_path:
+            self._load_from_storage()
+
+    def record_feedback(
+        self,
+        design_name: str,
+        generation_source: str,
+        spec_dict: Dict[str, Any],
+        validation_results: Dict[str, Any],
+    ):
+        import hashlib
+        import json
+
+        spec_str = json.dumps(spec_dict, sort_keys=True)
+        spec_hash = hashlib.md5(spec_str.encode()).hexdigest()[:12]
+
+        protocol = spec_dict.get("protocol", "unknown")
+
+        feedback_list = []
+
+        files_data = validation_results.get("files", [])
+
+        if isinstance(files_data, dict):
+            for file_name, file_info in files_data.items():
+                file_type = file_info.get("type", "unknown")
+                passed = file_info.get("passed", True)
+                errors = file_info.get("errors", [])
+                warnings = file_info.get("warnings", [])
+                score = file_info.get("score", 0.5)
+
+                feedback = ValidationFeedback(
+                    design_name=design_name,
+                    file_name=file_name,
+                    file_type=file_type,
+                    passed=passed,
+                    errors=errors,
+                    warnings=warnings,
+                    score=score,
+                )
+                feedback_list.append(feedback)
+
+                if passed:
+                    self._pattern_learner.record_success(file_type, protocol)
+                    reward = 1.0
+                else:
+                    for err in errors:
+                        self._pattern_learner.record_error(err, file_type)
+                    reward = -0.5
+
+                self._pattern_learner.record_attempt(file_type, protocol)
+                self._rl_learner.update(protocol, file_type, generation_source, reward)
+
+        elif isinstance(files_data, list):
+            for file_info in files_data:
+                file_name = file_info.get("filename", "unknown")
+                file_type = file_info.get("file_type", "unknown")
+                passed = file_info.get("passed", True)
+
+                issues = file_info.get("issues", [])
+                errors = []
+                warnings = []
+                for issue in issues:
+                    severity = issue.get("severity", "warning")
+                    message = issue.get("message", "")
+                    if severity == "error":
+                        errors.append(message)
+                    else:
+                        warnings.append(message)
+
+                error_count = file_info.get("error_count", 0)
+                warning_count = file_info.get("warning_count", 0)
+
+                if error_count > 0:
+                    passed = False
+
+                score = 1.0 if passed else 0.3
+                if passed and warning_count == 0:
+                    score = 1.0
+                elif passed and warning_count > 0:
+                    score = 0.7
+
+                feedback = ValidationFeedback(
+                    design_name=design_name,
+                    file_name=file_name,
+                    file_type=file_type,
+                    passed=passed,
+                    errors=errors,
+                    warnings=warnings,
+                    score=score,
+                )
+                feedback_list.append(feedback)
+
+                if passed:
+                    self._pattern_learner.record_success(file_type, protocol)
+                    reward = 1.0
+                else:
+                    for err in errors:
+                        self._pattern_learner.record_error(err, file_type)
+                    reward = -0.5
+
+                self._pattern_learner.record_attempt(file_type, protocol)
+                self._rl_learner.update(protocol, file_type, generation_source, reward)
+
+        all_passed = all(f.passed for f in feedback_list)
+        avg_score = sum(f.score for f in feedback_list) / len(feedback_list) if feedback_list else 0.0
+
+        history = GenerationHistory(
+            design_name=design_name,
+            generation_source=generation_source,
+            spec_hash=spec_hash,
+            feedback_list=feedback_list,
+            success_rate=1.0 if all_passed else 0.0,
+            avg_score=avg_score,
+        )
+        self._history.append(history)
+
+        self._total_generations += 1
+        if all_passed:
+            self._successful_generations += 1
+
+        if self._storage_path:
+            self._save_to_storage()
+
+    def select_best_generation_strategy(
+        self,
+        spec_dict: Dict[str, Any],
+        file_type: str,
+        available_sources: List[str],
+    ) -> Tuple[str, float]:
+        protocol = spec_dict.get("protocol", "unknown")
+
+        best_source, best_value = self._rl_learner.select_best_action(
+            protocol=protocol,
+            file_type=file_type,
+            available_sources=available_sources,
+            epsilon=0.05,
+        )
+
+        return best_source, best_value
+
+    def get_generation_hints(
+        self,
+        spec_dict: Dict[str, Any],
+        file_type: str,
+    ) -> Dict[str, Any]:
+        protocol = spec_dict.get("protocol", "unknown")
+
+        common_errors = self._pattern_learner.get_common_errors(5)
+        file_success_rate = self._pattern_learner.get_file_type_success_rate(file_type)
+        protocol_success_rate = self._pattern_learner.get_protocol_success_rate(protocol)
+
+        return {
+            "common_errors": common_errors,
+            "file_type_success_rate": file_success_rate,
+            "protocol_success_rate": protocol_success_rate,
+            "recommendations": self._generate_recommendations(
+                common_errors,
+                file_success_rate,
+                protocol_success_rate,
+            ),
+        }
+
+    def _generate_recommendations(
+        self,
+        common_errors: List[Tuple[str, int]],
+        file_success_rate: float,
+        protocol_success_rate: float,
+    ) -> List[str]:
+        recommendations = []
+
+        for error_pattern, count in common_errors[:3]:
+            if count > 0:
+                if "semicolon" in error_pattern:
+                    recommendations.append(
+                        "Ensure all statements end with semicolons"
+                    )
+                elif "parenthes" in error_pattern:
+                    recommendations.append(
+                        "Check for balanced parentheses"
+                    )
+                elif "brace" in error_pattern:
+                    recommendations.append(
+                        "Check for balanced begin/end blocks"
+                    )
+                elif "uvm_macro" in error_pattern:
+                    recommendations.append(
+                        "Add UVM factory registration macros (uvm_component_utils/uvm_object_utils)"
+                    )
+                elif "phase" in error_pattern:
+                    recommendations.append(
+                        "Ensure proper UVM phase implementation"
+                    )
+
+        if file_success_rate < 0.7:
+            recommendations.append(
+                "Consider using retrieval-based generation for this file type"
+            )
+
+        if protocol_success_rate < 0.7:
+            recommendations.append(
+                "Add protocol-specific templates may improve quality"
+            )
+
+        if not recommendations:
+            recommendations.append(
+                "No specific recommendations - generation should work well"
+            )
+
+        return recommendations
+
+    def get_stats(self) -> Dict[str, Any]:
+        return {
+            "total_generations": self._total_generations,
+            "successful_generations": self._successful_generations,
+            "success_rate": (
+                self._successful_generations / self._total_generations
+                if self._total_generations > 0
+                else 0.0
+            ),
+            "history_count": len(self._history),
+            "pattern_stats": self._pattern_learner.to_dict(),
+        }
+
+    def _save_to_storage(self):
+        if not self._storage_path:
+            return
+
+        try:
+            os.makedirs(os.path.dirname(self._storage_path), exist_ok=True)
+
+            data = {
+                "pattern_learner": self._pattern_learner.to_dict(),
+                "rl_learner": self._rl_learner.to_dict(),
+                "history": [h.to_dict() for h in self._history[-100:]],
+                "total_generations": self._total_generations,
+                "successful_generations": self._successful_generations,
+                "saved_at": datetime.now().isoformat(),
+            }
+
+            with open(self._storage_path, "w") as f:
+                json.dump(data, f, indent=2)
+
+            logger.debug("Learning module saved to: %s", self._storage_path)
+
+        except Exception as e:
+            logger.warning("Could not save learning module: %s", e)
+
+    def _load_from_storage(self):
+        if not self._storage_path or not os.path.exists(self._storage_path):
+            return
+
+        try:
+            with open(self._storage_path, "r") as f:
+                data = json.load(f)
+
+            self._pattern_learner = PatternLearner.from_dict(
+                data.get("pattern_learner", {})
+            )
+            self._rl_learner = ReinforcementLearner.from_dict(
+                data.get("rl_learner", {})
+            )
+
+            history_list = data.get("history", [])
+            for h_dict in history_list:
+                feedback_list = [
+                    ValidationFeedback.from_dict(f)
+                    for f in h_dict.get("feedback_list", [])
+                ]
+                history = GenerationHistory(
+                    design_name=h_dict.get("design_name", "unknown"),
+                    generation_source=h_dict.get("generation_source", "unknown"),
+                    spec_hash=h_dict.get("spec_hash", ""),
+                    feedback_list=feedback_list,
+                    success_rate=h_dict.get("success_rate", 0.0),
+                    avg_score=h_dict.get("avg_score", 0.0),
+                    timestamp=h_dict.get("timestamp", datetime.now().isoformat()),
+                )
+                self._history.append(history)
+
+            self._total_generations = data.get("total_generations", 0)
+            self._successful_generations = data.get("successful_generations", 0)
+
+            logger.info("Learning module loaded from: %s", self._storage_path)
+
+        except Exception as e:
+            logger.warning("Could not load learning module: %s", e)

src/models/llm_generator.py

@@ -0,0 +1,557 @@
+import logging
+from typing import List, Dict, Any, Optional, Tuple
+from dataclasses import dataclass, field
+from enum import Enum
+import json
+import re
+
+logger = logging.getLogger("uvmgen.ml.llm")
+
+
+class LLMType(Enum):
+    CODEGEN = "codegen"
+    CODET5 = "codet5"
+    CODEBERT = "codebert"
+    STARCODER = "starcoder"
+    LLAMA = "llama"
+    MISTRAL = "mistral"
+    FALLBACK = "fallback"
+
+
+@dataclass
+class LLMGenerationResult:
+    generated_code: str
+    prompt_used: str
+    model_name: str
+    tokens_generated: int
+    confidence: float = 0.5
+    metadata: Dict[str, Any] = field(default_factory=dict)
+    warnings: List[str] = field(default_factory=list)
+    errors: List[str] = field(default_factory=list)
+
+
+class LLMCodeGenerator:
+    _instance: Optional["LLMCodeGenerator"] = None
+    _model = None
+    _tokenizer = None
+    _model_name: str = "Salesforce/codegen-350M-mono"
+    _device: str = "cpu"
+    _initialized: bool = False
+    _llm_type: LLMType = LLMType.FALLBACK
+
+    UVM_PROMPT_TEMPLATE = """
+You are an expert in UVM (Universal Verification Methodology) and SystemVerilog.
+Generate production-quality UVM testbench code based on the following specification.
+
+SPECIFICATION:
+{spec_text}
+
+REQUIREMENTS:
+- Follow UVM 1.2 conventions and best practices
+- Use proper factory registration with `uvm_component_utils` or `uvm_object_utils`
+- Include appropriate phases (build_phase, connect_phase, run_phase)
+- Use TLM ports and exports for component communication
+- Include proper configuration database usage if needed
+- Generate synthesizable SystemVerilog code
+
+{context_examples}
+
+Generate the {file_type} for this specification. Return only the SystemVerilog code, no explanations.
+"""
+
+    FEW_SHOT_EXAMPLES = {
+        "driver": """
+EXAMPLE DRIVER:
+class my_driver extends uvm_driver #(my_seq_item);
+    `uvm_component_utils(my_driver)
+    
+    virtual my_if vif;
+    
+    function new(string name = "my_driver", uvm_component parent = null);
+        super.new(name, parent);
+    endfunction
+    
+    function void build_phase(uvm_phase phase);
+        super.build_phase(phase);
+        if (!uvm_config_db#(virtual my_if)::get(this, "", "vif", vif))
+            `uvm_fatal(get_type_name(), "Virtual interface not found")
+    endfunction
+    
+    task run_phase(uvm_phase phase);
+        forever begin
+            seq_item_port.get_next_item(req);
+            drive_item(req);
+            seq_item_port.item_done();
+        end
+    endtask
+    
+    task drive_item(my_seq_item item);
+        @(posedge vif.clk);
+        vif.valid <= 1'b1;
+        vif.data <= item.data;
+        @(posedge vif.clk);
+        vif.valid <= 1'b0;
+    endtask
+endclass
+""",
+        "monitor": """
+EXAMPLE MONITOR:
+class my_monitor extends uvm_monitor;
+    `uvm_component_utils(my_monitor)
+    
+    uvm_analysis_port #(my_seq_item) item_collected_port;
+    virtual my_if vif;
+    
+    function new(string name = "my_monitor", uvm_component parent = null);
+        super.new(name, parent);
+        item_collected_port = new("item_collected_port", this);
+    endfunction
+    
+    function void build_phase(uvm_phase phase);
+        super.build_phase(phase);
+        if (!uvm_config_db#(virtual my_if)::get(this, "", "vif", vif))
+            `uvm_fatal(get_type_name(), "Virtual interface not found")
+    endfunction
+    
+    task run_phase(uvm_phase phase);
+        my_seq_item item;
+        forever begin
+            @(posedge vif.clk);
+            if (vif.valid) begin
+                item = my_seq_item::type_id::create("item");
+                item.data = vif.data;
+                item_collected_port.write(item);
+            end
+        end
+    endtask
+endclass
+""",
+        "agent": """
+EXAMPLE AGENT:
+class my_agent extends uvm_agent;
+    `uvm_component_utils(my_agent)
+    
+    my_driver driver;
+    my_monitor monitor;
+    my_sequencer sequencer;
+    uvm_analysis_port #(my_seq_item) item_collected_port;
+    
+    function new(string name = "my_agent", uvm_component parent = null);
+        super.new(name, parent);
+        item_collected_port = new("item_collected_port", this);
+    endfunction
+    
+    function void build_phase(uvm_phase phase);
+        super.build_phase(phase);
+        
+        if (get_is_active() == UVM_ACTIVE) begin
+            driver = my_driver::type_id::create("driver", this);
+            sequencer = my_sequencer::type_id::create("sequencer", this);
+        end
+        
+        monitor = my_monitor::type_id::create("monitor", this);
+    endfunction
+    
+    function void connect_phase(uvm_phase phase);
+        super.connect_phase(phase);
+        
+        if (get_is_active() == UVM_ACTIVE) begin
+            driver.seq_item_port.connect(sequencer.seq_item_export);
+        end
+        
+        monitor.item_collected_port.connect(item_collected_port);
+    endfunction
+endclass
+""",
+    }
+
+    def __new__(cls, *args, **kwargs):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+        return cls._instance
+
+    def __init__(self, model_name: Optional[str] = None, device: Optional[str] = None):
+        if self._initialized:
+            return
+
+        if model_name:
+            self._model_name = model_name
+        if device:
+            self._device = device
+
+        self._initialized = False
+        self._model = None
+        self._tokenizer = None
+        self._detect_llm_type()
+
+    def _detect_llm_type(self):
+        name_lower = self._model_name.lower()
+        if "codegen" in name_lower:
+            self._llm_type = LLMType.CODEGEN
+        elif "codet5" in name_lower:
+            self._llm_type = LLMType.CODET5
+        elif "codebert" in name_lower:
+            self._llm_type = LLMType.CODEBERT
+        elif "starcoder" in name_lower or "starcoder" in name_lower:
+            self._llm_type = LLMType.STARCODER
+        elif "llama" in name_lower:
+            self._llm_type = LLMType.LLAMA
+        elif "mistral" in name_lower:
+            self._llm_type = LLMType.MISTRAL
+        else:
+            self._llm_type = LLMType.FALLBACK
+
+    def _load_model(self):
+        if self._initialized and self._model is not None:
+            return
+
+        if self._llm_type == LLMType.FALLBACK:
+            logger.info("LLMCodeGenerator using fallback mode (template-based)")
+            self._initialized = True
+            return
+
+        try:
+            import torch
+            from transformers import AutoTokenizer, AutoModelForCausalLM, AutoModelForSeq2SeqLM
+
+            if self._device == "auto":
+                self._device = "cuda" if torch.cuda.is_available() else "cpu"
+
+            logger.info("Loading LLM: %s on %s", self._model_name, self._device)
+
+            self._tokenizer = AutoTokenizer.from_pretrained(self._model_name)
+
+            if self._llm_type == LLMType.CODET5:
+                self._model = AutoModelForSeq2SeqLM.from_pretrained(
+                    self._model_name,
+                    torch_dtype=torch.float16 if self._device == "cuda" else torch.float32,
+                )
+            else:
+                self._model = AutoModelForCausalLM.from_pretrained(
+                    self._model_name,
+                    torch_dtype=torch.float16 if self._device == "cuda" else torch.float32,
+                )
+
+            self._model.to(self._device)
+            self._model.eval()
+
+            if self._tokenizer.pad_token is None:
+                self._tokenizer.pad_token = self._tokenizer.eos_token
+
+            self._initialized = True
+            logger.info("LLM loaded successfully")
+
+        except ImportError as e:
+            logger.warning(
+                "Could not load LLM (missing dependencies: %s). Using fallback mode.",
+                e,
+            )
+            self._llm_type = LLMType.FALLBACK
+            self._initialized = True
+        except Exception as e:
+            logger.warning(
+                "Could not load LLM (%s). Using fallback mode.",
+                e,
+            )
+            self._llm_type = LLMType.FALLBACK
+            self._initialized = True
+
+    def is_available(self) -> bool:
+        self._load_model()
+        return self._initialized and self._llm_type != LLMType.FALLBACK
+
+    def _spec_to_text(self, spec_dict: Dict[str, Any]) -> str:
+        lines = []
+
+        if "design_name" in spec_dict:
+            lines.append(f"Design Name: {spec_dict['design_name']}")
+
+        if "protocol" in spec_dict:
+            lines.append(f"Protocol: {spec_dict['protocol']}")
+
+        if "signals" in spec_dict:
+            lines.append("\nSignals:")
+            for sig in spec_dict["signals"]:
+                name = sig.get("name", "unknown")
+                direction = sig.get("direction", "inout")
+                width = sig.get("width", 1)
+                desc = sig.get("description", "")
+                lines.append(f"  - {name}: {direction}, width={width} {desc}")
+
+        if "registers" in spec_dict:
+            lines.append("\nRegisters:")
+            for reg in spec_dict["registers"]:
+                name = reg.get("name", "unknown")
+                addr = reg.get("address", "0x0")
+                width = reg.get("width", 32)
+                lines.append(f"  - {name}: addr={addr}, width={width}")
+
+        if "features" in spec_dict:
+            lines.append("\nFeatures:")
+            for feat in spec_dict["features"]:
+                lines.append(f"  - {feat}")
+
+        return "\n".join(lines)
+
+    def _build_prompt(
+        self,
+        spec_dict: Dict[str, Any],
+        file_type: str,
+        use_few_shot: bool = True,
+    ) -> str:
+        spec_text = self._spec_to_text(spec_dict)
+
+        context_examples = ""
+        if use_few_shot and file_type in self.FEW_SHOT_EXAMPLES:
+            context_examples = self.FEW_SHOT_EXAMPLES[file_type]
+
+        prompt = self.UVM_PROMPT_TEMPLATE.format(
+            spec_text=spec_text,
+            file_type=file_type,
+            context_examples=context_examples,
+        )
+
+        return prompt.strip()
+
+    def _extract_code(self, text: str) -> str:
+        code_block_patterns = [
+            r"```systemverilog\s+(.*?)```",
+            r"```verilog\s+(.*?)```",
+            r"```sv\s+(.*?)```",
+            r"```\s+(.*?)```",
+        ]
+
+        for pattern in code_block_patterns:
+            match = re.search(pattern, text, re.DOTALL | re.IGNORECASE)
+            if match:
+                return match.group(1).strip()
+
+        return text.strip()
+
+    def _fallback_generate(
+        self,
+        spec_dict: Dict[str, Any],
+        file_type: str,
+        templates: Optional[Dict[str, str]] = None,
+    ) -> LLMGenerationResult:
+        design_name = spec_dict.get("design_name", "unknown").lower()
+
+        fallback_templates = {
+            "driver": f"""
+class {design_name}_driver extends uvm_driver #({design_name}_seq_item);
+    `uvm_component_utils({design_name}_driver)
+    
+    virtual {design_name}_if vif;
+    
+    function new(string name = "{design_name}_driver", uvm_component parent = null);
+        super.new(name, parent);
+    endfunction
+    
+    function void build_phase(uvm_phase phase);
+        super.build_phase(phase);
+        if (!uvm_config_db#(virtual {design_name}_if)::get(this, "", "vif", vif))
+            `uvm_fatal(get_type_name(), "Virtual interface not found in config DB")
+    endfunction
+    
+    task run_phase(uvm_phase phase);
+        forever begin
+            seq_item_port.get_next_item(req);
+            drive_item(req);
+            seq_item_port.item_done();
+        end
+    endtask
+    
+    task drive_item({design_name}_seq_item item);
+        // Implement drive logic based on item
+        @(posedge vif.clk);
+    endtask
+endclass
+""",
+            "monitor": f"""
+class {design_name}_monitor extends uvm_monitor;
+    `uvm_component_utils({design_name}_monitor)
+    
+    uvm_analysis_port #({design_name}_seq_item) item_collected_port;
+    virtual {design_name}_if vif;
+    
+    function new(string name = "{design_name}_monitor", uvm_component parent = null);
+        super.new(name, parent);
+        item_collected_port = new("item_collected_port", this);
+    endfunction
+    
+    function void build_phase(uvm_phase phase);
+        super.build_phase(phase);
+        if (!uvm_config_db#(virtual {design_name}_if)::get(this, "", "vif", vif))
+            `uvm_fatal(get_type_name(), "Virtual interface not found in config DB")
+    endfunction
+    
+    task run_phase(uvm_phase phase);
+        {design_name}_seq_item item;
+        forever begin
+            @(posedge vif.clk);
+            // Sample signals and create item
+        end
+    endtask
+endclass
+""",
+            "agent": f"""
+class {design_name}_agent extends uvm_agent;
+    `uvm_component_utils({design_name}_agent)
+    
+    {design_name}_driver driver;
+    {design_name}_monitor monitor;
+    {design_name}_sequencer sequencer;
+    uvm_analysis_port #({design_name}_seq_item) item_collected_port;
+    
+    function new(string name = "{design_name}_agent", uvm_component parent = null);
+        super.new(name, parent);
+        item_collected_port = new("item_collected_port", this);
+    endfunction
+    
+    function void build_phase(uvm_phase phase);
+        super.build_phase(phase);
+        
+        if (get_is_active() == UVM_ACTIVE) begin
+            driver = {design_name}_driver::type_id::create("driver", this);
+            sequencer = {design_name}_sequencer::type_id::create("sequencer", this);
+        end
+        
+        monitor = {design_name}_monitor::type_id::create("monitor", this);
+    endfunction
+    
+    function void connect_phase(uvm_phase phase);
+        super.connect_phase(phase);
+        
+        if (get_is_active() == UVM_ACTIVE) begin
+            driver.seq_item_port.connect(sequencer.seq_item_export);
+        end
+        
+        monitor.item_collected_port.connect(item_collected_port);
+    endfunction
+endclass
+""",
+        }
+
+        if templates and file_type in templates:
+            code = templates[file_type]
+        elif file_type in fallback_templates:
+            code = fallback_templates[file_type]
+        else:
+            code = f"// {file_type} for {design_name} - template placeholder"
+
+        return LLMGenerationResult(
+            generated_code=code,
+            prompt_used=f"// Fallback generation for {file_type}",
+            model_name="fallback_template",
+            tokens_generated=len(code.split()),
+            confidence=0.3,
+            warnings=["Using fallback template generation (LLM not available)"],
+        )
+
+    def generate(
+        self,
+        spec_dict: Dict[str, Any],
+        file_type: str,
+        use_few_shot: bool = True,
+        max_tokens: int = 1024,
+        temperature: float = 0.2,
+        templates: Optional[Dict[str, str]] = None,
+    ) -> LLMGenerationResult:
+        self._load_model()
+
+        prompt = self._build_prompt(spec_dict, file_type, use_few_shot)
+
+        if self._llm_type == LLMType.FALLBACK or self._model is None:
+            return self._fallback_generate(spec_dict, file_type, templates)
+
+        try:
+            import torch
+
+            inputs = self._tokenizer(
+                prompt,
+                return_tensors="pt",
+                truncation=True,
+                max_length=1024,
+                padding=True,
+            )
+            inputs = {k: v.to(self._device) for k, v in inputs.items()}
+
+            with torch.no_grad():
+                if self._llm_type == LLMType.CODET5:
+                    outputs = self._model.generate(
+                        **inputs,
+                        max_new_tokens=max_tokens,
+                        temperature=temperature,
+                        do_sample=temperature > 0,
+                        num_return_sequences=1,
+                        pad_token_id=self._tokenizer.pad_token_id,
+                        eos_token_id=self._tokenizer.eos_token_id,
+                    )
+                else:
+                    outputs = self._model.generate(
+                        **inputs,
+                        max_new_tokens=max_tokens,
+                        temperature=temperature,
+                        do_sample=temperature > 0,
+                        num_return_sequences=1,
+                        pad_token_id=self._tokenizer.pad_token_id,
+                        eos_token_id=self._tokenizer.eos_token_id,
+                    )
+
+            generated_text = self._tokenizer.decode(outputs[0], skip_special_tokens=True)
+
+            if generated_text.startswith(prompt):
+                generated_text = generated_text[len(prompt) :].strip()
+
+            code = self._extract_code(generated_text)
+            tokens_generated = len(outputs[0]) - inputs["input_ids"].shape[1]
+
+            confidence = 0.7
+            if "uvm_component_utils" in code or "uvm_object_utils" in code:
+                confidence += 0.1
+            if "class" in code and "extends" in code:
+                confidence += 0.05
+            if "build_phase" in code or "run_phase" in code:
+                confidence += 0.05
+            if "endclass" in code:
+                confidence += 0.05
+
+            confidence = min(confidence, 0.95)
+
+            return LLMGenerationResult(
+                generated_code=code,
+                prompt_used=prompt,
+                model_name=self._model_name,
+                tokens_generated=tokens_generated,
+                confidence=confidence,
+                warnings=[],
+            )
+
+        except Exception as e:
+            logger.warning("Error during LLM generation: %s. Using fallback.", e)
+            result = self._fallback_generate(spec_dict, file_type, templates)
+            result.warnings.append(f"LLM generation failed: {str(e)}")
+            return result
+
+    def generate_batch(
+        self,
+        spec_dict: Dict[str, Any],
+        file_types: List[str],
+        use_few_shot: bool = True,
+        max_tokens: int = 1024,
+        temperature: float = 0.2,
+        templates: Optional[Dict[str, str]] = None,
+    ) -> Dict[str, LLMGenerationResult]:
+        results = {}
+
+        for file_type in file_types:
+            results[file_type] = self.generate(
+                spec_dict=spec_dict,
+                file_type=file_type,
+                use_few_shot=use_few_shot,
+                max_tokens=max_tokens,
+                temperature=temperature,
+                templates=templates.get(file_type) if templates else None,
+            )
+
+        return results

src/models/ml_generation_model.py

@@ -0,0 +1,204 @@
+import re
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional, Any
+
+
+@dataclass
+class MLModelConfig:
+    """Configuration for ML-based generation models."""
+    similarity_threshold: float = 0.75
+    auto_learn: bool = True
+    index_path: Optional[str] = None
+    top_k_retrieval: int = 3
+    fallback_to_templates: bool = True
+
+    use_llm: bool = True
+    llm_model_name: Optional[str] = None
+    llm_max_tokens: int = 1024
+    llm_temperature: float = 0.2
+    llm_use_few_shot: bool = True
+
+    use_semantic_encoder: bool = True
+    semantic_model_name: str = "microsoft/codebert-base"
+
+    use_learning: bool = True
+    learning_storage_path: Optional[str] = None
+    learning_rate: float = 0.1
+    reinforcement_discount: float = 0.9
+    exploration_epsilon: float = 0.05
+
+
+class RetrievalInfo:
+    """Information about last retrieval operation."""
+    def __init__(self, used_similarity: bool = True, similar_specs: int = 0, best_score: float = 0.0):
+        self.used_similarity = used_similarity
+        self.similar_specs = similar_specs
+        self.best_score = best_score
+
+
+class NameNormalizer:
+    """Utility for normalizing and adapting design names in filenames and code."""
+
+    DESIGN_NAME_PATTERN = re.compile(
+        r"([a-zA-Z_][a-zA-Z0-9_]*?)_(driver|monitor|agent|sequencer|sequence_item|sequence|scoreboard|coverage_collector|env|test|interface|testbench|ral_model|serial_monitor)",
+        re.IGNORECASE
+    )
+
+    @classmethod
+    def adapt_names(
+        cls,
+        filename: str,
+        old_design_name: str,
+        new_design_name: str,
+    ) -> str:
+        """
+        Adapt filenames and content from old design name to new design name.
+
+        Args:
+            filename: Original filename
+            old_design_name: Old design name to replace
+            new_design_name: New design name to use
+
+        Returns:
+            Adapted filename
+        """
+        if not old_design_name or not new_design_name:
+            return filename
+
+        old_lower = old_design_name.lower()
+        new_lower = new_design_name.lower()
+
+        base_name = filename
+        ext = ""
+
+        if "." in filename:
+            parts = filename.rsplit(".", 1)
+            base_name = parts[0]
+            ext = "." + parts[1] if len(parts) > 1 else ""
+
+        if old_lower in base_name.lower():
+            new_base = re.sub(
+                re.escape(old_design_name),
+                new_design_name,
+                base_name,
+                flags=re.IGNORECASE,
+            )
+            return new_base + ext
+
+        match = cls.DESIGN_NAME_PATTERN.match(base_name)
+        if match:
+            prefix = match.group(1)
+            suffix = match.group(2)
+            if prefix.lower() == old_lower:
+                return f"{new_design_name}_{suffix}{ext}"
+
+        return filename
+
+    @classmethod
+    def adapt_content(
+        cls,
+        content: str,
+        old_design_name: str,
+        new_design_name: str,
+    ) -> str:
+        """
+        Adapt SystemVerilog content from old design name to new design name.
+
+        Args:
+            content: Original SystemVerilog content
+            old_design_name: Old design name to replace
+            new_design_name: New design name to use
+
+        Returns:
+            Adapted content
+        """
+        if not old_design_name or not new_design_name or old_design_name == new_design_name:
+            return content
+
+        result = content
+
+        patterns = [
+            (
+                rf"\b{re.escape(old_design_name)}_([a-zA-Z_][a-zA-Z0-9_]*)\b",
+                f"{new_design_name}_\\1",
+            ),
+            (
+                rf"\bclass\s+{re.escape(old_design_name)}_",
+                f"class {new_design_name}_",
+            ),
+            (
+                rf"`uvm_component_utils\(\s*{re.escape(old_design_name)}_",
+                f"`uvm_component_utils({new_design_name}_",
+            ),
+            (
+                rf"`uvm_object_utils\(\s*{re.escape(old_design_name)}_",
+                f"`uvm_object_utils({new_design_name}_",
+            ),
+            (
+                rf"virtual\s+{re.escape(old_design_name)}_if\s+",
+                f"virtual {new_design_name}_if ",
+            ),
+            (
+                rf"{re.escape(old_design_name)}_if::type_id",
+                f"{new_design_name}_if::type_id",
+            ),
+        ]
+
+        for pattern, replacement in patterns:
+            result = re.sub(pattern, replacement, result, flags=re.IGNORECASE)
+
+        result = re.sub(
+            rf"\b{re.escape(old_design_name)}\b",
+            new_design_name,
+            result,
+        )
+
+        return result
+
+    @classmethod
+    def normalize_name(cls, name: str) -> str:
+        """
+        Normalize a design name to a standard format.
+
+        - Converts to snake_case
+        - Removes special characters
+        - Ensures valid SystemVerilog identifier
+
+        Args:
+            name: Original name
+
+        Returns:
+            Normalized name
+        """
+        if not name:
+            return "design"
+
+        result = name.strip()
+
+        result = re.sub(r"[^a-zA-Z0-9_]", "_", result)
+
+        result = re.sub(r"_+", "_", result)
+
+        result = result.strip("_")
+
+        if not result:
+            return "design"
+
+        if not result[0].isalpha() and result[0] != "_":
+            result = "_" + result
+
+        return result.lower()
+
+
+class MLGenerationModel:
+    """
+    ML-based generation model (legacy name for EnhancedMLGenerationModel).
+
+    This class exists for backward compatibility with tests and code
+    that imports MLGenerationModel. Use EnhancedMLGenerationModel directly
+    for new code.
+    """
+
+    def __new__(cls, *args, **kwargs):
+        from src.models.enhanced_ml_model import EnhancedMLGenerationModel
+        return EnhancedMLGenerationModel(*args, **kwargs)

src/models/semantic_encoder.py

@@ -0,0 +1,294 @@
+import logging
+from typing import List, Dict, Any, Optional, Tuple
+import numpy as np
+from dataclasses import dataclass, field
+
+logger = logging.getLogger("uvmgen.ml.semantic")
+
+
+@dataclass
+class SemanticEmbedding:
+    vector: np.ndarray
+    text: str
+    metadata: Dict[str, Any] = field(default_factory=dict)
+    embedding_type: str = "code"
+
+    @property
+    def dim(self) -> int:
+        return len(self.vector)
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "vector": self.vector.tolist(),
+            "text": self.text,
+            "metadata": self.metadata,
+            "embedding_type": self.embedding_type,
+            "dim": self.dim,
+        }
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> "SemanticEmbedding":
+        return cls(
+            vector=np.array(d["vector"], dtype=np.float32),
+            text=d["text"],
+            metadata=d.get("metadata", {}),
+            embedding_type=d.get("embedding_type", "code"),
+        )
+
+
+class SemanticCodeEncoder:
+    _instance: Optional["SemanticCodeEncoder"] = None
+    _model = None
+    _tokenizer = None
+    _model_name: str = "microsoft/codebert-base"
+    _device: str = "cpu"
+    _initialized: bool = False
+
+    def __new__(cls, *args, **kwargs):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+        return cls._instance
+
+    def __init__(self, model_name: Optional[str] = None, device: Optional[str] = None):
+        if self._initialized:
+            return
+
+        if model_name:
+            self._model_name = model_name
+        if device:
+            self._device = device
+
+        self._initialized = False
+        self._model = None
+        self._tokenizer = None
+
+    def _load_model(self):
+        if self._initialized and self._model is not None:
+            return
+
+        try:
+            import torch
+            from transformers import AutoTokenizer, AutoModel
+
+            if self._device == "auto":
+                self._device = "cuda" if torch.cuda.is_available() else "cpu"
+
+            logger.info("Loading semantic encoder: %s on %s", self._model_name, self._device)
+
+            self._tokenizer = AutoTokenizer.from_pretrained(self._model_name)
+            self._model = AutoModel.from_pretrained(self._model_name)
+            self._model.to(self._device)
+            self._model.eval()
+
+            self._initialized = True
+            logger.info("Semantic encoder loaded successfully")
+
+        except ImportError as e:
+            logger.warning(
+                "Could not load semantic encoder (missing dependencies: %s). "
+                "Using fallback TF-IDF-based similarity.",
+                e,
+            )
+            self._initialized = False
+            self._model = None
+            self._tokenizer = None
+        except Exception as e:
+            logger.warning(
+                "Could not load semantic encoder (%s). Using fallback similarity.",
+                e,
+            )
+            self._initialized = False
+            self._model = None
+            self._tokenizer = None
+
+    def is_available(self) -> bool:
+        self._load_model()
+        return self._initialized and self._model is not None
+
+    def encode(
+        self,
+        text: str,
+        embedding_type: str = "code",
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SemanticEmbedding:
+        self._load_model()
+
+        if not self.is_available():
+            return self._fallback_encode(text, embedding_type, metadata)
+
+        try:
+            import torch
+
+            inputs = self._tokenizer(
+                text,
+                return_tensors="pt",
+                truncation=True,
+                max_length=512,
+                padding=True,
+            )
+            inputs = {k: v.to(self._device) for k, v in inputs.items()}
+
+            with torch.no_grad():
+                outputs = self._model(**inputs)
+                embeddings = outputs.last_hidden_state[:, 0, :]
+                embeddings = embeddings.cpu().numpy().squeeze()
+
+            embeddings = embeddings / (np.linalg.norm(embeddings) + 1e-8)
+
+            return SemanticEmbedding(
+                vector=embeddings.astype(np.float32),
+                text=text,
+                metadata=metadata or {},
+                embedding_type=embedding_type,
+            )
+
+        except Exception as e:
+            logger.warning("Error encoding with neural model: %s. Using fallback.", e)
+            return self._fallback_encode(text, embedding_type, metadata)
+
+    def encode_batch(
+        self,
+        texts: List[str],
+        embedding_type: str = "code",
+        metadata_list: Optional[List[Dict[str, Any]]] = None,
+    ) -> List[SemanticEmbedding]:
+        self._load_model()
+
+        if not self.is_available():
+            return [
+                self._fallback_encode(text, embedding_type, metadata_list[i] if metadata_list else None)
+                for i, text in enumerate(texts)
+            ]
+
+        try:
+            import torch
+
+            inputs = self._tokenizer(
+                texts,
+                return_tensors="pt",
+                truncation=True,
+                max_length=512,
+                padding=True,
+            )
+            inputs = {k: v.to(self._device) for k, v in inputs.items()}
+
+            with torch.no_grad():
+                outputs = self._model(**inputs)
+                embeddings = outputs.last_hidden_state[:, 0, :]
+                embeddings = embeddings.cpu().numpy()
+
+            norms = np.linalg.norm(embeddings, axis=1, keepdims=True) + 1e-8
+            embeddings = embeddings / norms
+
+            results = []
+            for i, emb in enumerate(embeddings):
+                results.append(
+                    SemanticEmbedding(
+                        vector=emb.astype(np.float32),
+                        text=texts[i],
+                        metadata=metadata_list[i] if metadata_list else {},
+                        embedding_type=embedding_type,
+                    )
+                )
+            return results
+
+        except Exception as e:
+            logger.warning("Error batch encoding: %s. Using fallback.", e)
+            return [
+                self._fallback_encode(text, embedding_type, metadata_list[i] if metadata_list else None)
+                for i, text in enumerate(texts)
+            ]
+
+    def _fallback_encode(
+        self,
+        text: str,
+        embedding_type: str = "code",
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SemanticEmbedding:
+        words = text.lower().split()
+        vocab = sorted(set(words))
+        vec = np.zeros(len(vocab), dtype=np.float32)
+
+        for w in words:
+            if w in vocab:
+                vec[vocab.index(w)] += 1
+
+        norm = np.linalg.norm(vec)
+        if norm > 0:
+            vec = vec / norm
+
+        pad_size = 128 - len(vec)
+        if pad_size > 0:
+            vec = np.pad(vec, (0, pad_size), mode="constant")
+        elif pad_size < 0:
+            vec = vec[:128]
+
+        return SemanticEmbedding(
+            vector=vec.astype(np.float32),
+            text=text,
+            metadata=metadata or {},
+            embedding_type=embedding_type,
+        )
+
+    def similarity(self, emb1: SemanticEmbedding, emb2: SemanticEmbedding) -> float:
+        if len(emb1.vector) != len(emb2.vector):
+            min_len = min(len(emb1.vector), len(emb2.vector))
+            v1 = emb1.vector[:min_len]
+            v2 = emb2.vector[:min_len]
+        else:
+            v1 = emb1.vector
+            v2 = emb2.vector
+
+        norm1 = np.linalg.norm(v1)
+        norm2 = np.linalg.norm(v2)
+
+        if norm1 < 1e-8 or norm2 < 1e-8:
+            return 0.0
+
+        return float(np.dot(v1, v2) / (norm1 * norm2))
+
+    def batch_similarity(
+        self,
+        query_emb: SemanticEmbedding,
+        embeddings: List[SemanticEmbedding],
+    ) -> List[Tuple[int, float]]:
+        if not embeddings:
+            return []
+
+        q_vec = query_emb.vector
+        q_norm = np.linalg.norm(q_vec)
+
+        if q_norm < 1e-8:
+            return [(i, 0.0) for i in range(len(embeddings))]
+
+        results = []
+        for i, emb in enumerate(embeddings):
+            e_vec = emb.vector
+
+            if len(e_vec) != len(q_vec):
+                min_len = min(len(q_vec), len(e_vec))
+                qv = q_vec[:min_len]
+                ev = e_vec[:min_len]
+            else:
+                qv = q_vec
+                ev = e_vec
+
+            e_norm = np.linalg.norm(ev)
+            if e_norm < 1e-8:
+                results.append((i, 0.0))
+                continue
+
+            sim = float(np.dot(qv, ev) / (q_norm * e_norm))
+            results.append((i, sim))
+
+        return results
+
+
+def cosine_similarity(v1: np.ndarray, v2: np.ndarray) -> float:
+    norm1 = np.linalg.norm(v1)
+    norm2 = np.linalg.norm(v2)
+
+    if norm1 < 1e-8 or norm2 < 1e-8:
+        return 0.0
+
+    return float(np.dot(v1, v2) / (norm1 * norm2))

src/pipeline.py

@@ -55,7 +55,7 @@ class TBPipeline:
         model_type = ml_cfg.model_type
         self.logger.info("ML generation enabled, model_type=%s", model_type)
 
-        if model_type in ("ml", "hybrid"):
+        if model_type in ("ml", "hybrid", "llm", "semantic"):
             ml_model_config = MLModelConfig(
                 similarity_threshold=ml_cfg.similarity_threshold,
                 auto_learn=ml_cfg.auto_learn,
@@ -68,8 +68,19 @@ class TBPipeline:
                 config=ml_model_config,
                 templates_dir=self.cfg.generation.templates_dir,
                 strict_validation=True,
+                use_llm=ml_cfg.use_llm,
+                use_semantic_encoder=ml_cfg.use_semantic_encoder,
+                use_learning=ml_cfg.use_learning,
+                llm_model_name=ml_cfg.llm_model_name,
+                learning_storage_path=ml_cfg.learning_storage_path,
             )
             self.logger.info("Created EnhancedMLGenerationModel with index size: %d", len(model.index))
+
+            if model_type == "llm":
+                self.logger.info("LLM mode: will prioritize LLM generation")
+            elif model_type == "semantic":
+                self.logger.info("Semantic mode: will use semantic embeddings for similarity")
+
             return model
 
         self.logger.info("Falling back to template model")