Merge pull request #540 from MervinPraison/claude/issue-240-20250528_161140

fix: Add DeepSeek compatibility fallback for hierarchical process mode

Author: MervinPraison

src/praisonai-agents/praisonaiagents/process/process.py

@@ -1,12 +1,14 @@
 import logging
 import asyncio
+import json
 from typing import Dict, Optional, List, Any, AsyncGenerator
 from pydantic import BaseModel, ConfigDict
 from ..agent.agent import Agent
 from ..task.task import Task
 from ..main import display_error, client
 import csv
 import os
+from openai import AsyncOpenAI
 
 class LoopItems(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True)
@@ -106,6 +108,126 @@ def _find_next_not_started_task(self) -> Optional[Task]:
                 logging.debug(f"Fallback attempt {fallback_attempts}: No 'not started' task found within retry limit.")
         return None # Return None if no task found after all attempts
 
+    async def _get_manager_instructions_with_fallback_async(self, manager_task, manager_prompt, ManagerInstructions):
+        """Async version of getting manager instructions with fallback"""
+        try:
+            # First try structured output (OpenAI compatible)
+            logging.info("Attempting structured output...")
+            return await self._get_structured_response_async(manager_task, manager_prompt, ManagerInstructions)
+        except Exception as e:
+            logging.info(f"Structured output failed: {e}, falling back to JSON mode...")
+            # Fallback to regular JSON mode
+            try:
+                # Generate JSON structure description from Pydantic model
+                try:
+                    schema = ManagerInstructions.model_json_schema()
+                    props_desc = ", ".join([f'"{k}": <{v.get("type", "any")}>' for k, v in schema.get('properties', {}).items()])
+                    required_props = schema.get('required', [])
+                    required_desc = f" (required: {', '.join(f'\"{p}\"' for p in required_props)})" if required_props else ""
+                    json_structure_desc = "{" + props_desc + "}"
+                    enhanced_prompt = manager_prompt + f"\n\nIMPORTANT: Respond with valid JSON only, using this exact structure: {json_structure_desc}{required_desc}"
+                except Exception as schema_error:
+                    logging.warning(f"Could not generate schema for ManagerInstructions: {schema_error}. Using hardcoded prompt.")
+                    # Fallback to hardcoded prompt if schema generation fails
+                    enhanced_prompt = manager_prompt + "\n\nIMPORTANT: Respond with valid JSON only, using this exact structure: {\"task_id\": <int>, \"agent_name\": \"<string>\", \"action\": \"<execute or stop>\"}"
+                
+                return await self._get_json_response_async(manager_task, enhanced_prompt, ManagerInstructions)
+            except Exception as fallback_error:
+                error_msg = f"Both structured output and JSON fallback failed: {fallback_error}"
+                logging.error(error_msg, exc_info=True)
+                raise Exception(error_msg) from fallback_error
+
+    def _get_manager_instructions_with_fallback(self, manager_task, manager_prompt, ManagerInstructions):
+        """Sync version of getting manager instructions with fallback"""
+        try:
+            # First try structured output (OpenAI compatible)
+            logging.info("Attempting structured output...")
+            manager_response = client.beta.chat.completions.parse(
+                model=self.manager_llm,
+                messages=[
+                    {"role": "system", "content": manager_task.description},
+                    {"role": "user", "content": manager_prompt}
+                ],
+                temperature=0.7,
+                response_format=ManagerInstructions
+            )
+            return manager_response.choices[0].message.parsed
+        except Exception as e:
+            logging.info(f"Structured output failed: {e}, falling back to JSON mode...")
+            # Fallback to regular JSON mode
+            try:
+                # Generate JSON structure description from Pydantic model
+                try:
+                    schema = ManagerInstructions.model_json_schema()
+                    props_desc = ", ".join([f'"{k}": <{v.get("type", "any")}>' for k, v in schema.get('properties', {}).items()])
+                    required_props = schema.get('required', [])
+                    required_desc = f" (required: {', '.join(f'\"{p}\"' for p in required_props)})" if required_props else ""
+                    json_structure_desc = "{" + props_desc + "}"
+                    enhanced_prompt = manager_prompt + f"\n\nIMPORTANT: Respond with valid JSON only, using this exact structure: {json_structure_desc}{required_desc}"
+                except Exception as schema_error:
+                    logging.warning(f"Could not generate schema for ManagerInstructions: {schema_error}. Using hardcoded prompt.")
+                    # Fallback to hardcoded prompt if schema generation fails
+                    enhanced_prompt = manager_prompt + "\n\nIMPORTANT: Respond with valid JSON only, using this exact structure: {\"task_id\": <int>, \"agent_name\": \"<string>\", \"action\": \"<execute or stop>\"}"
+                
+                manager_response = client.chat.completions.create(
+                    model=self.manager_llm,
+                    messages=[
+                        {"role": "system", "content": manager_task.description},
+                        {"role": "user", "content": enhanced_prompt}
+                    ],
+                    temperature=0.7,
+                    response_format={"type": "json_object"}
+                )
+                
+                # Parse JSON and validate with Pydantic
+                try:
+                    json_content = manager_response.choices[0].message.content
+                    parsed_json = json.loads(json_content)
+                    return ManagerInstructions(**parsed_json)
+                except (json.JSONDecodeError, ValueError) as e:
+                    raise Exception(f"Failed to parse JSON response: {json_content}") from e
+            except Exception as fallback_error:
+                error_msg = f"Both structured output and JSON fallback failed: {fallback_error}"
+                logging.error(error_msg, exc_info=True)
+                raise Exception(error_msg) from fallback_error
+
+    async def _get_structured_response_async(self, manager_task, manager_prompt, ManagerInstructions):
+        """Async version of structured response"""
+        # Create an async client instance for this async method
+        async_client = AsyncOpenAI()
+        manager_response = await async_client.beta.chat.completions.parse(
+            model=self.manager_llm,
+            messages=[
+                {"role": "system", "content": manager_task.description},
+                {"role": "user", "content": manager_prompt}
+            ],
+            temperature=0.7,
+            response_format=ManagerInstructions
+        )
+        return manager_response.choices[0].message.parsed
+
+    async def _get_json_response_async(self, manager_task, enhanced_prompt, ManagerInstructions):
+        """Async version of JSON fallback response"""
+        # Create an async client instance for this async method
+        async_client = AsyncOpenAI()
+        manager_response = await async_client.chat.completions.create(
+            model=self.manager_llm,
+            messages=[
+                {"role": "system", "content": manager_task.description},
+                {"role": "user", "content": enhanced_prompt}
+            ],
+            temperature=0.7,
+            response_format={"type": "json_object"}
+        )
+        
+        # Parse JSON and validate with Pydantic
+        try:
+            json_content = manager_response.choices[0].message.content
+            parsed_json = json.loads(json_content)
+            return ManagerInstructions(**parsed_json)
+        except (json.JSONDecodeError, ValueError) as e:
+            raise Exception(f"Failed to parse JSON response: {json_content}") from e
+
 
     async def aworkflow(self) -> AsyncGenerator[str, None]:
         """Async version of workflow method"""
@@ -496,26 +618,13 @@ class ManagerInstructions(BaseModel):
             try:
                 logging.info("Requesting manager instructions...")
                 if manager_task.async_execution:
-                    manager_response = await client.beta.chat.completions.parse(
-                        model=self.manager_llm,
-                        messages=[
-                            {"role": "system", "content": manager_task.description},
-                            {"role": "user", "content": manager_prompt}
-                        ],
-                        temperature=0.7,
-                        response_format=ManagerInstructions
+                    parsed_instructions = await self._get_manager_instructions_with_fallback_async(
+                        manager_task, manager_prompt, ManagerInstructions
                     )
                 else:
-                    manager_response = client.beta.chat.completions.parse(
-                        model=self.manager_llm,
-                        messages=[
-                            {"role": "system", "content": manager_task.description},
-                            {"role": "user", "content": manager_prompt}
-                        ],
-                        temperature=0.7,
-                        response_format=ManagerInstructions
+                    parsed_instructions = self._get_manager_instructions_with_fallback(
+                        manager_task, manager_prompt, ManagerInstructions
                     )
-                parsed_instructions = manager_response.choices[0].message.parsed
                 logging.info(f"Manager instructions: {parsed_instructions}")
             except Exception as e:
                 display_error(f"Manager parse error: {e}")
@@ -1110,16 +1219,9 @@ class ManagerInstructions(BaseModel):
 
             try:
                 logging.info("Requesting manager instructions...")
-                manager_response = client.beta.chat.completions.parse(
-                    model=self.manager_llm,
-                    messages=[
-                        {"role": "system", "content": manager_task.description},
-                        {"role": "user", "content": manager_prompt}
-                    ],
-                    temperature=0.7,
-                    response_format=ManagerInstructions
+                parsed_instructions = self._get_manager_instructions_with_fallback(
+                    manager_task, manager_prompt, ManagerInstructions
                 )
-                parsed_instructions = manager_response.choices[0].message.parsed
                 logging.info(f"Manager instructions: {parsed_instructions}")
             except Exception as e:
                 display_error(f"Manager parse error: {e}")