| 314 | |
| 315 | |
| 316 | def parse_stream_usage_metrics( |
| 317 | chunks: list[bytes], |
| 318 | model: str, |
| 319 | pricing: dict[str, ModelPricing], |
| 320 | ) -> UsageMetrics | None: |
| 321 | if not chunks: |
| 322 | return None |
| 323 | try: |
| 324 | text = b"".join(chunks).decode("utf-8", errors="replace") |
| 325 | except Exception: |
| 326 | return None |
| 327 | |
| 328 | anthropic_usage: dict[str, Any] = {} |
| 329 | latest: UsageMetrics | None = None |
| 330 | fallback_output_tokens = 0 |
| 331 | fallback_prompt_tokens = 0 |
| 332 | fallback_has_content = False |
| 333 | |
| 334 | for line in text.splitlines(): |
| 335 | if not line.startswith("data:"): |
| 336 | continue |
| 337 | payload = line[5:].strip() |
| 338 | if not payload or payload == "[DONE]": |
| 339 | continue |
| 340 | try: |
| 341 | data = json.loads(payload) |
| 342 | except json.JSONDecodeError: |
| 343 | continue |
| 344 | if isinstance(data, dict) and isinstance(data.get("usage"), dict): |
| 345 | anthropic_usage.update(data["usage"]) |
| 346 | parsed = parse_usage_metrics(json.dumps({"usage": anthropic_usage}).encode("utf-8"), model, pricing) |
| 347 | if parsed is not None: |
| 348 | latest = parsed |
| 349 | continue |
| 350 | if not isinstance(data, dict): |
| 351 | continue |
| 352 | message = data.get("message") |
| 353 | if isinstance(message, dict) and isinstance(message.get("usage"), dict): |
| 354 | anthropic_usage.update(message["usage"]) |
| 355 | if data.get("type") == "message_delta" and isinstance(data.get("usage"), dict): |
| 356 | anthropic_usage.update(data["usage"]) |
| 357 | if anthropic_usage: |
| 358 | parsed = parse_usage_metrics( |
| 359 | json.dumps({"usage": anthropic_usage}).encode("utf-8"), |
| 360 | model, |
| 361 | pricing, |
| 362 | ) |
| 363 | if parsed is not None: |
| 364 | latest = parsed |
| 365 | choices = data.get("choices") |
| 366 | if isinstance(choices, list) and choices: |
| 367 | first_choice = choices[0] |
| 368 | delta = first_choice.get("delta") if isinstance(first_choice, dict) else None |
| 369 | if isinstance(delta, dict): |
| 370 | content = delta.get("content") or delta.get("reasoning_content") or "" |
| 371 | if isinstance(content, str) and content: |
| 372 | fallback_has_content = True |
| 373 | fallback_output_tokens += max(1, len(content) // 4) |