Make a streaming request and yield chunks.
(
self, payload: dict
)
| 419 | )) |
| 420 | |
| 421 | def _process_stream( |
| 422 | self, payload: dict |
| 423 | ) -> Generator[Union[str, LLMResponse], None, None]: |
| 424 | """Make a streaming request and yield chunks.""" |
| 425 | headers = { |
| 426 | 'Content-Type': 'application/json' |
| 427 | } |
| 428 | |
| 429 | url = f'{self._api_url}/engines/v1/chat/completions' |
| 430 | |
| 431 | request = urllib.request.Request( |
| 432 | url, |
| 433 | data=json.dumps(payload).encode('utf-8'), |
| 434 | headers=headers, |
| 435 | method='POST' |
| 436 | ) |
| 437 | |
| 438 | try: |
| 439 | response = urllib.request.urlopen( |
| 440 | request, timeout=300, context=SSL_CONTEXT |
| 441 | ) |
| 442 | except urllib.error.HTTPError as e: |
| 443 | error_body = e.read().decode('utf-8') |
| 444 | try: |
| 445 | error_data = json.loads(error_body) |
| 446 | error_msg = error_data.get( |
| 447 | 'error', {} |
| 448 | ).get('message', str(e)) |
| 449 | except json.JSONDecodeError: |
| 450 | error_msg = error_body or str(e) |
| 451 | raise LLMClientError(LLMError( |
| 452 | message=error_msg, |
| 453 | code=str(e.code), |
| 454 | provider=self.provider_name, |
| 455 | retryable=e.code in (429, 500, 502, 503, 504) |
| 456 | )) |
| 457 | except urllib.error.URLError as e: |
| 458 | raise LLMClientError(LLMError( |
| 459 | message=f"Connection error: {e.reason}. " |
| 460 | f"Is Docker Model Runner running at " |
| 461 | f"{self._api_url}?", |
| 462 | provider=self.provider_name, |
| 463 | retryable=True |
| 464 | )) |
| 465 | except socket.timeout: |
| 466 | raise LLMClientError(LLMError( |
| 467 | message="Request timed out.", |
| 468 | code='timeout', |
| 469 | provider=self.provider_name, |
| 470 | retryable=True |
| 471 | )) |
| 472 | |
| 473 | try: |
| 474 | yield from self._read_openai_stream(response) |
| 475 | finally: |
| 476 | response.close() |
| 477 | |
| 478 | def _read_openai_stream( |
no test coverage detected