(
*,
messages: list[Any],
step_type: str,
tool_names: tuple[str, ...],
prompt: str,
needs_tool_calling: bool,
wants_structured_output: bool,
)
| 1616 | |
| 1617 | |
| 1618 | def _estimate_step_risk( |
| 1619 | *, |
| 1620 | messages: list[Any], |
| 1621 | step_type: str, |
| 1622 | tool_names: tuple[str, ...], |
| 1623 | prompt: str, |
| 1624 | needs_tool_calling: bool, |
| 1625 | wants_structured_output: bool, |
| 1626 | ) -> str: |
| 1627 | tool_result_text, tool_result_is_error, tool_command = _current_step_tool_result_context(messages, step_type) |
| 1628 | previous_tool_result_text, previous_tool_result_is_error = _latest_tool_result_signal(messages) |
| 1629 | prompt_text = str(prompt or "") |
| 1630 | if _is_suggestion_mode_prompt(prompt_text): |
| 1631 | return "low" |
| 1632 | prompt_has_high_risk_shape = ( |
| 1633 | needs_tool_calling |
| 1634 | and text_substance_score(prompt_text) |
| 1635 | >= DEFAULT_SIGNAL_TUNING.tool_prompt_high_risk_substance_score |
| 1636 | ) |
| 1637 | |
| 1638 | if step_type == "tool-result-followup": |
| 1639 | if tool_result_is_error: |
| 1640 | return "high" |
| 1641 | if _contains_tool_failure_signal(tool_result_text, tool_command): |
| 1642 | return "high" |
| 1643 | if _tool_result_is_routine_success(tool_result_text, tool_result_is_error, tool_command): |
| 1644 | return "low" |
| 1645 | if _tool_result_is_short_success_observation(tool_result_text, tool_result_is_error, tool_command): |
| 1646 | return "normal" |
| 1647 | return "normal" |
| 1648 | |
| 1649 | retrying_previous_tool = ( |
| 1650 | _contains_risk_marker(prompt_text, _RETRY_PROMPT_MARKERS) |
| 1651 | and ( |
| 1652 | previous_tool_result_is_error |
| 1653 | or _contains_tool_failure_signal(previous_tool_result_text) |
| 1654 | ) |
| 1655 | ) |
| 1656 | if retrying_previous_tool or prompt_has_high_risk_shape: |
| 1657 | return "high" |
| 1658 | |
| 1659 | if step_type == "tool-selection" and len(prompt_text) <= 40 and len(tool_names) <= 6: |
| 1660 | return "low" |
| 1661 | if not needs_tool_calling and len(prompt_text) <= 80: |
| 1662 | return "low" |
| 1663 | return "normal" |
| 1664 | |
| 1665 | |
| 1666 | def _agent_state_pressure(messages: list[Any], step_risk: str) -> tuple[int, float]: |
no test coverage detected