Extract last user message, system prompt, and max_tokens from request body.
(body: dict)
| 453 | |
| 454 | |
| 455 | def _extract_prompt(body: dict) -> tuple[str, str | None, int]: |
| 456 | """Extract last user message, system prompt, and max_tokens from request body.""" |
| 457 | messages = body.get("messages", []) |
| 458 | raw_max_tokens = body.get("max_tokens", body.get("max_completion_tokens", 4096)) |
| 459 | if raw_max_tokens is None: |
| 460 | raw_max_tokens = body.get("max_completion_tokens", 4096) |
| 461 | try: |
| 462 | max_tokens = max(1, int(raw_max_tokens)) if raw_max_tokens is not None else 4096 |
| 463 | except (TypeError, ValueError): |
| 464 | max_tokens = 4096 |
| 465 | |
| 466 | prompt = "" |
| 467 | system_prompt: str | None = None |
| 468 | |
| 469 | if _debug_log.isEnabledFor(logging.DEBUG): |
| 470 | _debug_log.debug( |
| 471 | "=== REQUEST STRUCTURE === messages=%d roles=%s", |
| 472 | len(messages), |
| 473 | [m.get("role") for m in messages], |
| 474 | ) |
| 475 | for i, msg in enumerate(messages): |
| 476 | role = msg.get("role", "?") |
| 477 | content = msg.get("content", "") |
| 478 | raw_text = _content_text(content) if not isinstance(content, str) else content |
| 479 | _debug_log.debug( |
| 480 | " msg[%d] role=%s content_type=%s len=%d preview=%.200s", |
| 481 | i, role, type(content).__name__, len(raw_text), raw_text[:200], |
| 482 | ) |
| 483 | |
| 484 | for msg in reversed(messages): |
| 485 | if msg.get("role") == "user" and not prompt: |
| 486 | raw_content = msg.get("content", "") |
| 487 | candidate = _extract_user_prompt_text(raw_content) |
| 488 | if _debug_log.isEnabledFor(logging.DEBUG): |
| 489 | raw_text = _content_text(raw_content) if not isinstance(raw_content, str) else raw_content |
| 490 | _debug_log.debug( |
| 491 | " LAST USER MSG: raw_len=%d extracted_len=%d raw_preview=%.300s extracted=%.300s", |
| 492 | len(raw_text), len(candidate), raw_text[:300], candidate[:300], |
| 493 | ) |
| 494 | if candidate: |
| 495 | prompt = candidate |
| 496 | if msg.get("role") == "system" and system_prompt is None: |
| 497 | text = _content_text(msg.get("content", "")) |
| 498 | system_prompt = text if text else None |
| 499 | |
| 500 | if _debug_log.isEnabledFor(logging.DEBUG): |
| 501 | _debug_log.debug( |
| 502 | " EXTRACTED: prompt_len=%d prompt=%.200s system_len=%d", |
| 503 | len(prompt), prompt[:200], |
| 504 | len(system_prompt) if system_prompt else 0, |
| 505 | ) |
| 506 | |
| 507 | return prompt, system_prompt, max_tokens |
| 508 | |
| 509 | |
| 510 | def _build_debug_response(prompt: str, system_prompt: str | None, routing_config=DEFAULT_CONFIG) -> dict: |