Read and parse an OpenAI-format SSE stream. Uses readline() for incremental reading.
(
self, response
)
| 476 | response.close() |
| 477 | |
| 478 | def _read_openai_stream( |
| 479 | self, response |
| 480 | ) -> Generator[Union[str, LLMResponse], None, None]: |
| 481 | """Read and parse an OpenAI-format SSE stream. |
| 482 | |
| 483 | Uses readline() for incremental reading. |
| 484 | """ |
| 485 | content_parts = [] |
| 486 | tool_calls_data = {} |
| 487 | finish_reason = None |
| 488 | model_name = self._model |
| 489 | usage = Usage() |
| 490 | |
| 491 | while True: |
| 492 | line_bytes = response.readline() |
| 493 | if not line_bytes: |
| 494 | break |
| 495 | |
| 496 | line = line_bytes.decode('utf-8', errors='replace').strip() |
| 497 | |
| 498 | if not line or line.startswith(':'): |
| 499 | continue |
| 500 | |
| 501 | if line == 'data: [DONE]': |
| 502 | continue |
| 503 | |
| 504 | if not line.startswith('data: '): |
| 505 | continue |
| 506 | |
| 507 | try: |
| 508 | data = json.loads(line[6:]) |
| 509 | except json.JSONDecodeError: |
| 510 | continue |
| 511 | |
| 512 | if 'usage' in data and data['usage']: |
| 513 | u = data['usage'] |
| 514 | usage = Usage( |
| 515 | input_tokens=u.get('prompt_tokens', 0), |
| 516 | output_tokens=u.get('completion_tokens', 0), |
| 517 | total_tokens=u.get('total_tokens', 0) |
| 518 | ) |
| 519 | |
| 520 | if 'model' in data: |
| 521 | model_name = data['model'] |
| 522 | |
| 523 | choices = data.get('choices', []) |
| 524 | if not choices: |
| 525 | continue |
| 526 | |
| 527 | choice = choices[0] |
| 528 | delta = choice.get('delta', {}) |
| 529 | |
| 530 | if choice.get('finish_reason'): |
| 531 | finish_reason = choice['finish_reason'] |
| 532 | |
| 533 | text_chunk = delta.get('content') |
| 534 | if text_chunk: |
| 535 | content_parts.append(text_chunk) |
no test coverage detected