refactor(datasetgen): split generator module into focused components

fix: resolve taxonomy dataset generation issues and enhance system initialization

Author: sun

.env.example

@@ -19,7 +19,10 @@ COMPILEO_API_KEYS=default-api-key
 # REDIS CONFIGURATION
 # ==========================================
 # Redis connection URL (required for job queuing)
-REDIS_URL=redis://redis:6379/0
+# For local Python development: use localhost
+# For Docker deployment: use redis container name (set by docker-compose.yml)
+# If not set, defaults to redis://localhost:6379/0 for local development
+# REDIS_URL=redis://localhost:6379/0
 
 # ==========================================
 # API ENDPOINT CONFIGURATION

requirements-docker.txt

@@ -32,7 +32,7 @@ pikepdf
 opencv-python==4.12.0.88
 pdfminer.six==20250506
 pypdf==6.0.0
-pdf2image
+
 PyMuPDF==1.26.4
 unstructured[docx,pptx,xlsx,md]==0.18.15
 unstructured-client==0.42.3

requirements.txt

@@ -31,7 +31,7 @@ pikepdf
 opencv-python==4.12.0.88
 pdfminer.six==20250506
 pypdf==6.0.0
-pdf2image
+
 PyMuPDF==1.26.4
 unstructured[docx,pptx,xlsx,md]==0.18.15
 unstructured-client==0.42.3

src/compileo/features/datasetgen/data_loader.py

@@ -39,8 +39,8 @@ def build_prompt(self, project_id: int, prompt_name: str, chunk_text: Optional[s
         prompt_record = self.prompt_repository.get_by_name(prompt_name)
         if prompt_record:
             prompt_template = prompt_record[2]  # content is the third column (index 2)
-        elif prompt_name == "default":
-            # Hardcoded default prompt template for dataset generation
+        elif prompt_name == "default" or prompt_name == "qa_pairs":
+            # Hardcoded default prompt template for Q&A dataset generation
             prompt_template = """Generate {datasets_per_chunk} question-answer pairs from the following content.
 
 Content: {chunk}
@@ -52,6 +52,8 @@ def build_prompt(self, project_id: int, prompt_name: str, chunk_text: Optional[s
 - Vary question types (factual, analytical, applicative)
 - Focus on key concepts, important details, and relationships in the content
 
+{taxonomy_instructions}
+
 {purpose_context}{audience_context}{complexity_level_context}
 
 Format your response as a JSON array of objects with this structure:
@@ -84,6 +86,12 @@ def build_prompt(self, project_id: int, prompt_name: str, chunk_text: Optional[s
         # Ensure datasets_per_chunk is in param_dict
         param_dict['datasets_per_chunk'] = datasets_per_chunk
 
+        # Ensure taxonomy-related parameters exist
+        param_dict.setdefault('taxonomy_instructions', '')
+        param_dict.setdefault('taxonomy_name', '')
+        param_dict.setdefault('taxonomy_description', '')
+        param_dict.setdefault('taxonomy_categories', '')
+
         # Load taxonomy if specified
         taxonomy_info = None
         if taxonomy_project and taxonomy_name:
@@ -108,6 +116,9 @@ def build_prompt(self, project_id: int, prompt_name: str, chunk_text: Optional[s
         if taxonomy_info:
             param_dict.update(taxonomy_info)
 
+        # Handle legacy {Na} placeholder (maps to taxonomy_name)
+        param_dict.setdefault('Na', param_dict.get('taxonomy_name', ''))
+
         # Replace {chunk} with actual chunk text if provided
         if chunk_text is not None:
             if "{chunk}" in prompt_template:
@@ -244,12 +255,39 @@ def build_question_generation_prompt(self, project_id: int, chunk_text: str,
             except Exception as e:
                 logger.warning(f"Could not load taxonomy {taxonomy_project}/{taxonomy_name}: {e}")
 
+        # Ensure taxonomy-related parameters exist even when no taxonomy is loaded
+        param_dict.setdefault('taxonomy_instructions', '')
+        param_dict.setdefault('taxonomy_name', '')
+        param_dict.setdefault('taxonomy_description', '')
+        param_dict.setdefault('taxonomy_categories', '')
+
         prompt_record = self.prompt_repository.get_by_name("question_generation")
         if prompt_record:
             prompt_template = prompt_record[2]
         else:
             # Fallback template
-            prompt_template = "Generate a question based on the following content: {chunk_text}"
+            prompt_template = """Generate {datasets_per_chunk} questions from the following content.
+
+Content: {chunk_text}
+
+Instructions:
+- Create diverse, high-quality questions that test understanding of the content
+- Ensure questions are answerable using only the provided information
+- Vary question types (factual, analytical, applicative)
+- Focus on key concepts, important details, and relationships in the content
+
+{taxonomy_instructions}
+
+{purpose_context}{audience_context}{complexity_level_context}
+
+Format your response as a JSON array of objects with this structure:
+[
+  {{
+    "question": "Your question here",
+    "category": "content_category",
+    "difficulty": "easy|medium|hard"
+  }}
+]"""
 
         # Format the prompt with the parameters and the chunk text
         final_prompt = prompt_template.format(
@@ -319,12 +357,39 @@ def build_answer_generation_prompt(self, project_id: int, chunk_text: str,
             except Exception as e:
                 logger.warning(f"Could not load taxonomy {taxonomy_project}/{taxonomy_name}: {e}")
 
+        # Ensure taxonomy-related parameters exist even when no taxonomy is loaded
+        param_dict.setdefault('taxonomy_instructions', '')
+        param_dict.setdefault('taxonomy_name', '')
+        param_dict.setdefault('taxonomy_description', '')
+        param_dict.setdefault('taxonomy_categories', '')
+
         prompt_record = self.prompt_repository.get_by_name("answer_generation")
         if prompt_record:
             prompt_template = prompt_record[2]
         else:
             # Fallback template
-            prompt_template = "Generate an answer for the following content: {chunk_text}"
+            prompt_template = """Generate {datasets_per_chunk} answers for the following question based on the content.
+
+Question: {question}
+Content: {chunk_text}
+
+Instructions:
+- Provide accurate, comprehensive answers based only on the given content
+- Ensure answers are supported by the provided information
+- Include relevant details and context from the content
+
+{taxonomy_instructions}
+
+{purpose_context}{audience_context}{complexity_level_context}
+
+Format your response as a JSON array of objects with this structure:
+[
+  {{
+    "answer": "Your answer here",
+    "category": "content_category",
+    "confidence": 0.0-1.0
+  }}
+]"""
 
         # Format the prompt with the parameters and the chunk text
         final_prompt = prompt_template.format(
@@ -394,12 +459,38 @@ def build_summarization_prompt(self, project_id: int, chunk_text: str,
             except Exception as e:
                 logger.warning(f"Could not load taxonomy {taxonomy_project}/{taxonomy_name}: {e}")
 
+        # Ensure taxonomy-related parameters exist even when no taxonomy is loaded
+        param_dict.setdefault('taxonomy_instructions', '')
+        param_dict.setdefault('taxonomy_name', '')
+        param_dict.setdefault('taxonomy_description', '')
+        param_dict.setdefault('taxonomy_categories', '')
+
         prompt_record = self.prompt_repository.get_by_name("summarization")
         if prompt_record:
             prompt_template = prompt_record[2]
         else:
             # Fallback template
-            prompt_template = "Summarize the following content in a concise manner: {chunk_text}"
+            prompt_template = """Generate {datasets_per_chunk} summaries from the following content.
+
+Content: {chunk_text}
+
+Instructions:
+- Create concise, accurate summaries that capture the main points
+- Focus on key concepts and important details
+- Maintain factual accuracy from the source content
+
+{taxonomy_instructions}
+
+{purpose_context}{audience_context}{complexity_level_context}
+
+Format your response as a JSON array of objects with this structure:
+[
+  {{
+    "summary": "Your summary here",
+    "key_points": ["Point 1", "Point 2", "Point 3"],
+    "category": "content_category"
+  }}
+]"""
 
         # Format the prompt with the parameters and the chunk text
         final_prompt = prompt_template.format(

src/compileo/features/datasetgen/generator.py

@@ -0,0 +1,14 @@
+from typing import Any, Dict, List, Optional
+from ..prompt_builder import PromptBuilder
+from ..llm_interaction import LLMInteraction
+
+class BaseGenerationStrategy:
+    """
+    Base class for dataset generation strategies.
+    """
+    def __init__(self, prompt_builder: PromptBuilder, llm_interaction: LLMInteraction):
+        self.prompt_builder = prompt_builder
+        self.llm_interaction = llm_interaction
+
+    def generate(self, *args, **kwargs) -> List[Dict[str, Any]]:
+        raise NotImplementedError("Subclasses must implement generate method.")

src/compileo/features/datasetgen/prompt_builder.py

@@ -0,0 +1,126 @@
+from typing import List, Dict, Any, Optional
+from src.compileo.core.logging import get_logger
+from .base_strategy import BaseGenerationStrategy
+
+logger = get_logger(__name__)
+
+class ChunkGenerationStrategy(BaseGenerationStrategy):
+    """
+    Strategy for generating datasets from raw text chunks.
+    """
+
+    def generate_qa(self, project_id: int, prompt_name: str, chunk: str,
+                    taxonomy_project: Optional[str] = None, taxonomy_name: Optional[str] = None,
+                    datasets_per_chunk: int = 3) -> List[Dict[str, Any]]:
+        """
+        Generates multiple question-answer pairs for a single chunk.
+        """
+        results = []
+        for i in range(datasets_per_chunk):
+            final_prompt = self.prompt_builder.build_prompt(project_id, prompt_name, chunk,
+                                                            taxonomy_project=taxonomy_project, taxonomy_name=taxonomy_name,
+                                                            datasets_per_chunk=datasets_per_chunk)
+            # Modify prompt to request multiple items if needed
+            if datasets_per_chunk > 1:
+                final_prompt += f"\n\nGenerate {datasets_per_chunk} different question-answer pairs from this text chunk. Return them as a JSON array."
+
+            result = self.llm_interaction.generate(final_prompt)
+
+            # Handle multiple results
+            if isinstance(result, list):
+                for item in result[:datasets_per_chunk]:
+                    results.append(item)
+            else:
+                results.append(result)
+
+            if len(results) >= datasets_per_chunk:
+                break
+
+        return results[:datasets_per_chunk]
+
+    def generate_questions(self, project_id: int, chunk: str,
+                           taxonomy_project: Optional[str] = None, taxonomy_name: Optional[str] = None,
+                           datasets_per_chunk: int = 3) -> List[Dict[str, Any]]:
+        """
+        Generates multiple questions for a single chunk.
+        """
+        results = []
+        for i in range(datasets_per_chunk):
+            final_prompt = self.prompt_builder.build_question_generation_prompt(project_id, chunk, taxonomy_project, taxonomy_name)
+            # Modify prompt to request multiple items if needed
+            if datasets_per_chunk > 1:
+                final_prompt += f"\n\nGenerate {datasets_per_chunk} different questions from this text chunk. Return them as a JSON array."
+
+            result = self.llm_interaction.generate(final_prompt)
+
+            # Handle multiple results
+            if isinstance(result, list):
+                for item in result[:datasets_per_chunk]:
+                    item["result_type"] = "questions"
+                    results.append(item)
+            else:
+                result["result_type"] = "questions"
+                results.append(result)
+
+            if len(results) >= datasets_per_chunk:
+                break
+
+        return results[:datasets_per_chunk]
+
+    def generate_answers(self, project_id: int, chunk: str,
+                         taxonomy_project: Optional[str] = None, taxonomy_name: Optional[str] = None,
+                         datasets_per_chunk: int = 3) -> List[Dict[str, Any]]:
+        """
+        Generates multiple answers for a single chunk.
+        """
+        results = []
+        for i in range(datasets_per_chunk):
+            final_prompt = self.prompt_builder.build_answer_generation_prompt(project_id, chunk, taxonomy_project, taxonomy_name)
+            # Modify prompt to request multiple items if needed
+            if datasets_per_chunk > 1:
+                final_prompt += f"\n\nGenerate {datasets_per_chunk} different answers from this text chunk. Return them as a JSON array."
+
+            result = self.llm_interaction.generate(final_prompt)
+
+            # Handle multiple results
+            if isinstance(result, list):
+                for item in result[:datasets_per_chunk]:
+                    item["result_type"] = "answers"
+                    results.append(item)
+            else:
+                result["result_type"] = "answers"
+                results.append(result)
+
+            if len(results) >= datasets_per_chunk:
+                break
+
+        return results[:datasets_per_chunk]
+
+    def generate_summaries(self, project_id: int, chunk: str,
+                           taxonomy_project: Optional[str] = None, taxonomy_name: Optional[str] = None,
+                           datasets_per_chunk: int = 3) -> List[Dict[str, Any]]:
+        """
+        Generates multiple summaries for a single chunk.
+        """
+        results = []
+        for i in range(datasets_per_chunk):
+            final_prompt = self.prompt_builder.build_summarization_prompt(project_id, chunk, taxonomy_project, taxonomy_name)
+            # Modify prompt to request multiple items if needed
+            if datasets_per_chunk > 1:
+                final_prompt += f"\n\nGenerate {datasets_per_chunk} different summaries from this text chunk. Return them as a JSON array."
+
+            result = self.llm_interaction.generate(final_prompt)
+
+            # Handle multiple results
+            if isinstance(result, list):
+                for item in result[:datasets_per_chunk]:
+                    item["result_type"] = "summary"
+                    results.append(item)
+            else:
+                result["result_type"] = "summary"
+                results.append(result)
+
+            if len(results) >= datasets_per_chunk:
+                break
+
+        return results[:datasets_per_chunk]