Generate responses for multiple prompts using batch inference if available. Args: prompts: List of user prompts system_prompt: Optional system prompt (can be single string for all, or list of strings per prompt) use_tuner_model: If True,
(
self,
prompts: List[str],
system_prompt: Optional[Union[str, List[str]]] = None,
use_tuner_model: bool = False,
temperature: float = 0.7,
max_tokens: Optional[int] = None,
)
| 130 | return None |
| 131 | |
| 132 | def batch_generate( |
| 133 | self, |
| 134 | prompts: List[str], |
| 135 | system_prompt: Optional[Union[str, List[str]]] = None, |
| 136 | use_tuner_model: bool = False, |
| 137 | temperature: float = 0.7, |
| 138 | max_tokens: Optional[int] = None, |
| 139 | ) -> List[str]: |
| 140 | """ |
| 141 | Generate responses for multiple prompts using batch inference if available. |
| 142 | |
| 143 | Args: |
| 144 | prompts: List of user prompts |
| 145 | system_prompt: Optional system prompt (can be single string for all, or list of strings per prompt) |
| 146 | use_tuner_model: If True, use tuner_model_fn; otherwise use tuning_model_fn |
| 147 | temperature: Sampling temperature |
| 148 | max_tokens: Maximum tokens to generate |
| 149 | |
| 150 | Returns: |
| 151 | List of generated text responses |
| 152 | """ |
| 153 | if not prompts: |
| 154 | return [] |
| 155 | |
| 156 | # Normalize system_prompts to handle both single string and list |
| 157 | if isinstance(system_prompt, list): |
| 158 | if len(system_prompt) != len(prompts): |
| 159 | raise ValueError(f"Length of system_prompt list ({len(system_prompt)}) must match length of prompts ({len(prompts)})") |
| 160 | system_prompts = system_prompt |
| 161 | else: |
| 162 | # Single system prompt for all |
| 163 | system_prompts = system_prompt |
| 164 | |
| 165 | # Use batch inference function if available |
| 166 | batch_fn = self.tuner_model_batch_fn if use_tuner_model else self.tuning_model_batch_fn |
| 167 | |
| 168 | if batch_fn is not None: |
| 169 | # Use batch inference for better performance |
| 170 | for attempt in range(self.max_retries): |
| 171 | try: |
| 172 | responses = batch_fn(prompts, system_prompts, temperature, max_tokens) |
| 173 | if len(responses) != len(prompts): |
| 174 | raise ValueError(f"Batch function returned {len(responses)} responses for {len(prompts)} prompts") |
| 175 | return responses |
| 176 | except Exception as e: |
| 177 | if attempt < self.max_retries - 1: |
| 178 | time.sleep(self.retry_delay * (attempt + 1)) |
| 179 | else: |
| 180 | raise Exception(f"Batch generation failed after {self.max_retries} attempts: {str(e)}") |
| 181 | else: |
| 182 | # Fallback to sequential generation |
| 183 | responses = [] |
| 184 | # Convert to list if it's a single system_prompt |
| 185 | if not isinstance(system_prompts, list): |
| 186 | system_prompts = [system_prompts] * len(prompts) |
| 187 | |
| 188 | for idx, (prompt, sys_prompt) in enumerate(zip(prompts, system_prompts)): |
| 189 | response = self.generate( |
no test coverage detected