Merge pull request #117 from cloudera/release/SDS-1-0-4

Key Features
1. Configurable Concurrency Parameters
Added max_concurrent_topics parameter to SynthesisRequest (1-100, default: 5)
Added max_workers parameter to EvaluationRequest (1-100, default: 4)
Allows users to optimize performance based on infrastructure
2. Service Architecture Refactor
Split synthesis services:
synthesis_service.py - Freeform data generation only
synthesis_legacy_service.py - SFT and Custom Workflow
Split evaluator services:
evaluator_service.py - Freeform evaluation only
evaluator_legacy_service.py - SFT and Custom Workflow evaluation
Improved code maintainability and separation of concerns
Backward compatible - all existing endpoints work as before
This is a minor release to provide concurrent threads as a variable via backend API for advanced users

Author: Khauneesh-AI

.gitignore

@@ -55,6 +55,16 @@ qa_pairs*
 Khauneesh/
 *job_args*
 
+# Generated data files
+freeform_data_*.json
+row_data_*.json
+lending_*.json
+seeds_*.json
+SeedsInstructions.json
+*_example.json
+nm.json
+french_input.json
+
 # DB
 *metadata.db-shm
 *metadata.db-wal

app/client/src/pages/DataGenerator/Finish.tsx

@@ -255,7 +255,8 @@ const Finish = () => {
             title: 'Review Dataset',
             description: 'Review your dataset to ensure it properly fits your usecase.',
             icon: <GradingIcon/>,
-            href: getFilesURL(genDatasetResp?.export_path?.local || "")
+            href: getFilesURL(genDatasetResp?.export_path?.local || ""),
+            external: true
         },
         {
             avatar: '',
@@ -278,7 +279,8 @@ const Finish = () => {
             title: 'Review Dataset',
             description: 'Once your dataset finishes generating, you can review your dataset in the workbench files',
             icon: <GradingIcon/>,
-            href: getFilesURL('')
+            href: getFilesURL(''),
+            external: true
         },
         {
             avatar: '',
@@ -361,7 +363,18 @@ const Finish = () => {
             <List
                 itemLayout="horizontal"
                 dataSource={isDemo ? nextStepsListPreview : nextStepsListNonPreview}
-                renderItem={({ title, href, icon, description}, i) => (
+                renderItem={({ title, href, icon, description, external }, i) => (
+                    external ? 
+                    <Link to={href} target="_blank" rel="noopener noreferrer">
+                        <List.Item key={`${title}-${i}`}>
+                            <List.Item.Meta
+                                avatar={<Avatar style={{ backgroundColor: '#1677ff'}} icon={icon} />}
+                                title={title}
+                                description={description}
+                            />
+                        </List.Item>
+                    </Link> : 
+
                     <Link to={href}>
                         <List.Item key={`${title}-${i}`}>
                             <List.Item.Meta

app/client/src/pages/Home/TemplateCard.tsx

@@ -132,10 +132,10 @@ const TagsContainer = styled.div`
   }
 `;
 
-const StyledTag = styled(Tag)`
-  color: ${props => props.theme.color};
-  background-color: ${props => props.theme.backgroundColor};
-  border: 1px solid ${props => props.theme.borderColor};
+const StyledTag = styled(Tag)<{ $theme: { color: string; backgroundColor: string; borderColor: string } }>`
+  color: ${props => props.$theme.color} !important;
+  background-color: ${props => props.$theme.backgroundColor} !important;
+  border: 1px solid ${props => props.$theme.borderColor} !important;
 `;
 
 
@@ -150,7 +150,7 @@ const TemplateCard: React.FC<Props> = ({ template }) => {
         const { color, backgroundColor, borderColor } = getTemplateTagColors(theme as string);
 
         return (
-          <StyledTag key={tag} theme={{ color, backgroundColor, borderColor }}>
+          <StyledTag key={tag} $theme={{ color, backgroundColor, borderColor }}>
             <div className="tag-title" title={tag} style={{ maxWidth: '150px', color }}>
                 {tag}
             </div>

app/core/model_handlers.py

@@ -180,6 +180,8 @@ def generate_response(
             return self._handle_caii_request(prompt)
         if self.inference_type == "openai":
             return self._handle_openai_request(prompt)
+        if self.inference_type == "openai_compatible":
+            return self._handle_openai_compatible_request(prompt)
         if self.inference_type == "gemini":
             return self._handle_gemini_request(prompt)
         raise ModelHandlerError(f"Unsupported inference_type={self.inference_type}", 400)
@@ -342,6 +344,66 @@ def _handle_openai_request(self, prompt: str):
         except Exception as e:
             raise ModelHandlerError(f"OpenAI request failed: {e}", 500)
 
+    # ---------- OpenAI Compatible -------------------------------------------------------
+    def _handle_openai_compatible_request(self, prompt: str):
+        """Handle OpenAI compatible endpoints with proper timeout configuration"""
+        try:
+            import httpx
+            from openai import OpenAI
+            
+            # Get API key from environment variable (only credential needed)
+            api_key = os.getenv('OpenAI_Endpoint_Compatible_Key')
+            if not api_key:
+                raise ModelHandlerError("OpenAI_Endpoint_Compatible_Key environment variable not set", 500)
+            
+            # Base URL comes from caii_endpoint parameter (passed during initialization)
+            openai_compatible_endpoint = self.caii_endpoint
+            if not openai_compatible_endpoint:
+                raise ModelHandlerError("OpenAI compatible endpoint not provided", 500)
+            
+            # Configure timeout for OpenAI compatible client (same as OpenAI v1.57.2)
+            timeout_config = httpx.Timeout(
+                connect=self.OPENAI_CONNECT_TIMEOUT,
+                read=self.OPENAI_READ_TIMEOUT,
+                write=10.0,
+                pool=5.0
+            )
+            
+            # Configure httpx client with certificate verification for private cloud
+            if os.path.exists("/etc/ssl/certs/ca-certificates.crt"):
+                http_client = httpx.Client(
+                    verify="/etc/ssl/certs/ca-certificates.crt",
+                    timeout=timeout_config
+                )
+            else:
+                http_client = httpx.Client(timeout=timeout_config)
+            
+            # Remove trailing '/chat/completions' if present (similar to CAII handling)
+            openai_compatible_endpoint = openai_compatible_endpoint.removesuffix('/chat/completions')
+            
+            client = OpenAI(
+                api_key=api_key,
+                base_url=openai_compatible_endpoint,
+                http_client=http_client
+            )
+            
+            completion = client.chat.completions.create(
+                model=self.model_id,
+                messages=[{"role": "user", "content": prompt}],
+                max_tokens=self.model_params.max_tokens,
+                temperature=self.model_params.temperature,
+                top_p=self.model_params.top_p,
+                stream=False,
+            )
+            
+            print("generated via OpenAI Compatible endpoint")
+            response_text = completion.choices[0].message.content
+            
+            return self._extract_json_from_text(response_text) if not self.custom_p else response_text
+            
+        except Exception as e:
+            raise ModelHandlerError(f"OpenAI Compatible request failed: {str(e)}", status_code=500)
+
     # ---------- Gemini -------------------------------------------------------
     def _handle_gemini_request(self, prompt: str):
         if genai is None:

app/main.py

@@ -42,8 +42,10 @@
 sys.path.append(str(ROOT_DIR))
 
 from app.services.evaluator_service import EvaluatorService
+from app.services.evaluator_legacy_service import EvaluatorLegacyService
 from app.models.request_models import SynthesisRequest, EvaluationRequest, Export_synth, ModelParameters, CustomPromptRequest, JsonDataSize, RelativePath, Technique
 from app.services.synthesis_service import SynthesisService
+from app.services.synthesis_legacy_service import SynthesisLegacyService
 from app.services.export_results import Export_Service
 
 from app.core.prompt_templates import PromptBuilder, PromptHandler
@@ -66,8 +68,10 @@
 #****************************************Initialize************************************************
 
 # Initialize services
-synthesis_service = SynthesisService()
-evaluator_service = EvaluatorService()
+synthesis_service = SynthesisService()  # Freeform only
+synthesis_legacy_service = SynthesisLegacyService()  # SFT and Custom_Workflow
+evaluator_service = EvaluatorService()  # Freeform only
+evaluator_legacy_service = EvaluatorLegacyService()  # SFT and Custom_Workflow
 export_service = Export_Service()
 db_manager = DatabaseManager()
 
@@ -552,9 +556,11 @@ async def generate_examples(request: SynthesisRequest):
 
     if is_demo== True:
         if request.input_path:
-            return await synthesis_service.generate_result(request,is_demo, request_id=request_id)
+            # Custom_Workflow technique - route to legacy service
+            return await synthesis_legacy_service.generate_result(request,is_demo, request_id=request_id)
         else:
-            return await synthesis_service.generate_examples(request,is_demo, request_id=request_id)
+            # SFT technique - route to legacy service
+            return await synthesis_legacy_service.generate_examples(request,is_demo, request_id=request_id)
     else:
        return synthesis_job.generate_job(request, core, mem, request_id=request_id)
 
@@ -626,7 +632,8 @@ async def evaluate_examples(request: EvaluationRequest):
 
     is_demo = request.is_demo
     if is_demo:
-       return evaluator_service.evaluate_results(request, request_id=request_id)
+       # SFT and Custom_Workflow evaluation - route to legacy service
+       return evaluator_legacy_service.evaluate_results(request, request_id=request_id)
 
     else:
         return synthesis_job.evaluate_job(request, request_id=request_id)
@@ -1242,7 +1249,7 @@ def is_empty(self):
 async def health_check():
     """Get API health status"""
     #return {"status": "healthy"}
-    return synthesis_service.get_health_check()
+    return synthesis_legacy_service.get_health_check()
 
 @app.get("/{use_case}/example_payloads")
 async def get_example_payloads(use_case:UseCase):
@@ -1255,6 +1262,7 @@ async def get_example_payloads(use_case:UseCase):
                     "technique": "sft",
                     "topics": ["python_basics", "data_structures"],
                     "is_demo": True,
+                    "max_concurrent_topics": 5,
                     "examples":  [
                 {
                     "question": "How do you create a list in Python and add elements to it?",
@@ -1281,6 +1289,7 @@ async def get_example_payloads(use_case:UseCase):
                     "technique": "sft",
                     "topics": ["basic_queries", "joins"],
                     "is_demo": True,
+                    "max_concurrent_topics": 5,
                     "schema": "CREATE TABLE users (id INT PRIMARY KEY, name VARCHAR(100), email VARCHAR(255));\nCREATE TABLE orders (id INT PRIMARY KEY, user_id INT, amount DECIMAL(10,2), FOREIGN KEY (user_id) REFERENCES users(id));",
                     "examples":[
                                 {
@@ -1309,6 +1318,7 @@ async def get_example_payloads(use_case:UseCase):
             "topics": ["topic 1", "topic 2"],
             "custom_prompt": "Give your instructions here",
             "is_demo": True,
+            "max_concurrent_topics": 5,
 
             "examples":[
                         {

app/models/request_models.py

@@ -123,6 +123,7 @@ class SynthesisRequest(BaseModel):
     # Optional fields that can override defaults
     inference_type: Optional[str] = "aws_bedrock"
     caii_endpoint: Optional[str] = None
+    openai_compatible_endpoint: Optional[str] = None
     topics: Optional[List[str]] = None
     doc_paths: Optional[List[str]] = None
     input_path: Optional[List[str]] = None
@@ -137,7 +138,13 @@ class SynthesisRequest(BaseModel):
     example_path: Optional[str] = None
     schema: Optional[str] = None  # Added schema field
     custom_prompt: Optional[str] = None 
-    display_name: Optional[str] = None 
+    display_name: Optional[str] = None
+    max_concurrent_topics: Optional[int] = Field(
+        default=5, 
+        ge=1, 
+        le=100, 
+        description="Maximum number of concurrent topics to process (1-100)"
+    ) 
 
     # Optional model parameters with defaults
     model_params: Optional[ModelParameters] = Field(
@@ -155,7 +162,7 @@ class SynthesisRequest(BaseModel):
                 "technique": "sft",
                 "topics": ["python_basics", "data_structures"],
                 "is_demo": True,
-                
+                "max_concurrent_topics": 5
 
             }
         }
@@ -208,6 +215,12 @@ class EvaluationRequest(BaseModel):
     display_name: Optional[str] = None 
     output_key: Optional[str] = 'Prompt'
     output_value: Optional[str] = 'Completion'
+    max_workers: Optional[int] = Field(
+        default=4, 
+        ge=1, 
+        le=100, 
+        description="Maximum number of worker threads for parallel evaluation (1-100)"
+    )
 
     # Export configuration
     export_type: str = "local"  # "local" or "s3"
@@ -226,7 +239,8 @@ class EvaluationRequest(BaseModel):
                 "inference_type": "aws_bedrock",
                 "import_path": "qa_pairs_llama3-1-70b-instruct-v1:0_20241114_212837_test.json",
                 "import_type": "local",
-                "export_type":"local"
+                "export_type":"local",
+                "max_workers": 4
 
             }
         }

app/run_eval_job.py

@@ -31,6 +31,7 @@
 from app.models.request_models import EvaluationRequest, ModelParameters
 
 from app.services.evaluator_service import EvaluatorService
+from app.services.evaluator_legacy_service import EvaluatorLegacyService
 import asyncio
 import nest_asyncio  
 
@@ -40,7 +41,7 @@
 async def run_eval(request, job_name, request_id):
     try:
 
-        job = EvaluatorService()
+        job = EvaluatorLegacyService()
         result = job.evaluate_results(request,job_name, is_demo=False, request_id=request_id)
         return result
     except Exception as e:

app/run_job.py

@@ -32,6 +32,7 @@
 import json
 from app.models.request_models import SynthesisRequest
 from app.services.synthesis_service import SynthesisService
+from app.services.synthesis_legacy_service import SynthesisLegacyService
 import asyncio
 import nest_asyncio  # Add this import
 
@@ -41,7 +42,7 @@
 async def run_synthesis(request, job_name, request_id):
     """Run standard synthesis job for question-answer pairs"""
     try:
-        job = SynthesisService()
+        job = SynthesisLegacyService()
         if request.input_path:
             result = await job.generate_result(request, job_name, is_demo=False, request_id=request_id)
         else: