Stream SSE chunks from Fireworks API.
(self, headers: dict, payload: dict, display_model_name: str, actual_model_id: str)
| 186 | return response_data |
| 187 | |
| 188 | async def _stream_fireworks_response(self, headers: dict, payload: dict, display_model_name: str, actual_model_id: str): |
| 189 | """Stream SSE chunks from Fireworks API.""" |
| 190 | try: |
| 191 | client = get_http_client() |
| 192 | async with client.stream("POST", self.FIREWORKS_API_URL, headers=headers, json=payload) as response: |
| 193 | response.raise_for_status() |
| 194 | async for line in response.aiter_lines(): |
| 195 | if line.startswith("data: "): |
| 196 | # Replace actual model ID with display name in streaming chunks |
| 197 | chunk_data = line[6:] # Remove "data: " prefix |
| 198 | if chunk_data != "[DONE]": |
| 199 | try: |
| 200 | chunk_json = json.loads(chunk_data) |
| 201 | if "model" in chunk_json: |
| 202 | chunk_json["model"] = display_model_name |
| 203 | yield f"data: {json.dumps(chunk_json)}\n\n" |
| 204 | except json.JSONDecodeError: |
| 205 | # If we can't parse, just forward as-is |
| 206 | yield f"data: {chunk_data}\n\n" |
| 207 | else: |
| 208 | yield f"data: {chunk_data}\n\n" |
| 209 | except httpx.RequestError as exc: |
| 210 | logger.error(f"Fireworks streaming API request failed: {exc}") |
| 211 | yield f"data: {json.dumps({'error': f'Connection error: {exc}'})}\n\n" |
| 212 | except httpx.HTTPStatusError as exc: |
| 213 | error_body = exc.response.text |
| 214 | logger.error(f"Fireworks streaming API error {exc.response.status_code}: {error_body[:500]}") |
| 215 | yield f"data: {json.dumps({'error': f'API error ({exc.response.status_code}): {error_body}'})}\n\n" |
| 216 | except Exception as exc: |
| 217 | logger.exception(f"Unexpected error in Fireworks streaming for model {actual_model_id}") |
| 218 | yield f"data: {json.dumps({'error': f'Unexpected error: {exc}'})}\n\n" |
no test coverage detected