Extract routing features from the current request step.
(
body: dict,
*,
step_type: str,
tool_names: list[str] | tuple[str, ...] | None = None,
prompt: str = "",
max_output_tokens: int = 4096,
session_id: str | None = None,
)
| 1686 | |
| 1687 | |
| 1688 | def _extract_routing_features( |
| 1689 | body: dict, |
| 1690 | *, |
| 1691 | step_type: str, |
| 1692 | tool_names: list[str] | tuple[str, ...] | None = None, |
| 1693 | prompt: str = "", |
| 1694 | max_output_tokens: int = 4096, |
| 1695 | session_id: str | None = None, |
| 1696 | ) -> RoutingFeatures: |
| 1697 | """Extract routing features from the current request step.""" |
| 1698 | messages = body.get("messages", []) |
| 1699 | raw_tools = body.get("tools") or body.get("customTools") or [] |
| 1700 | normalized_tool_names = tuple(tool_names or ()) |
| 1701 | if not normalized_tool_names: |
| 1702 | derived_tool_names: list[str] = [] |
| 1703 | for tool in raw_tools: |
| 1704 | fn = tool.get("function") or tool.get("definition") or {} |
| 1705 | name = str(fn.get("name") or tool.get("name") or "").strip() |
| 1706 | if name: |
| 1707 | derived_tool_names.append(name) |
| 1708 | normalized_tool_names = tuple(derived_tool_names) |
| 1709 | |
| 1710 | has_vision = any(_has_vision_content(msg.get("content")) for msg in messages if isinstance(msg, dict)) |
| 1711 | needs_tool_calling = bool(raw_tools) |
| 1712 | has_tool_results = step_type == "tool-result-followup" |
| 1713 | suggestion_mode = _is_suggestion_mode_prompt(prompt) |
| 1714 | title_generation_sidechannel = _body_has_title_generation_sidechannel(messages) |
| 1715 | |
| 1716 | response_format = body.get("response_format") |
| 1717 | response_format_name: str | None = None |
| 1718 | wants_structured_output = False |
| 1719 | if isinstance(response_format, dict): |
| 1720 | response_format_name = str(response_format.get("type") or "json_schema").strip().lower() or "json_schema" |
| 1721 | wants_structured_output = True |
| 1722 | elif isinstance(response_format, str): |
| 1723 | response_format_name = response_format.strip().lower() or None |
| 1724 | wants_structured_output = response_format_name in {"json", "json_schema"} |
| 1725 | if ( |
| 1726 | not title_generation_sidechannel |
| 1727 | and not wants_structured_output |
| 1728 | and _body_has_structured_output_directive(messages) |
| 1729 | ): |
| 1730 | response_format_name = "system" |
| 1731 | wants_structured_output = True |
| 1732 | |
| 1733 | step_risk = ( |
| 1734 | "low" |
| 1735 | if title_generation_sidechannel |
| 1736 | else _estimate_step_risk( |
| 1737 | messages=messages, |
| 1738 | step_type=step_type, |
| 1739 | tool_names=normalized_tool_names, |
| 1740 | prompt=prompt, |
| 1741 | needs_tool_calling=needs_tool_calling, |
| 1742 | wants_structured_output=wants_structured_output, |
| 1743 | ) |
| 1744 | ) |
| 1745 | contextual_floor = ( |