mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-04-07 19:53:40 +00:00
89 lines
3.0 KiB
Python
89 lines
3.0 KiB
Python
"""Runtime-specific helper functions and constants."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from loguru import logger
|
|
|
|
from nanobot.utils.helpers import stringify_text_blocks
|
|
|
|
_MAX_REPEAT_EXTERNAL_LOOKUPS = 2
|
|
|
|
EMPTY_FINAL_RESPONSE_MESSAGE = (
|
|
"I completed the tool steps but couldn't produce a final answer. "
|
|
"Please try again or narrow the task."
|
|
)
|
|
|
|
FINALIZATION_RETRY_PROMPT = (
|
|
"You have already finished the tool work. Do not call any more tools. "
|
|
"Using only the conversation and tool results above, provide the final answer for the user now."
|
|
)
|
|
|
|
|
|
def empty_tool_result_message(tool_name: str) -> str:
|
|
"""Short prompt-safe marker for tools that completed without visible output."""
|
|
return f"({tool_name} completed with no output)"
|
|
|
|
|
|
def ensure_nonempty_tool_result(tool_name: str, content: Any) -> Any:
|
|
"""Replace semantically empty tool results with a short marker string."""
|
|
if content is None:
|
|
return empty_tool_result_message(tool_name)
|
|
if isinstance(content, str) and not content.strip():
|
|
return empty_tool_result_message(tool_name)
|
|
if isinstance(content, list):
|
|
if not content:
|
|
return empty_tool_result_message(tool_name)
|
|
text_payload = stringify_text_blocks(content)
|
|
if text_payload is not None and not text_payload.strip():
|
|
return empty_tool_result_message(tool_name)
|
|
return content
|
|
|
|
|
|
def is_blank_text(content: str | None) -> bool:
|
|
"""True when *content* is missing or only whitespace."""
|
|
return content is None or not content.strip()
|
|
|
|
|
|
def build_finalization_retry_message() -> dict[str, str]:
|
|
"""A short no-tools-allowed prompt for final answer recovery."""
|
|
return {"role": "user", "content": FINALIZATION_RETRY_PROMPT}
|
|
|
|
|
|
def external_lookup_signature(tool_name: str, arguments: dict[str, Any]) -> str | None:
|
|
"""Stable signature for repeated external lookups we want to throttle."""
|
|
if tool_name == "web_fetch":
|
|
url = str(arguments.get("url") or "").strip()
|
|
if url:
|
|
return f"web_fetch:{url.lower()}"
|
|
if tool_name == "web_search":
|
|
query = str(arguments.get("query") or arguments.get("search_term") or "").strip()
|
|
if query:
|
|
return f"web_search:{query.lower()}"
|
|
return None
|
|
|
|
|
|
def repeated_external_lookup_error(
|
|
tool_name: str,
|
|
arguments: dict[str, Any],
|
|
seen_counts: dict[str, int],
|
|
) -> str | None:
|
|
"""Block repeated external lookups after a small retry budget."""
|
|
signature = external_lookup_signature(tool_name, arguments)
|
|
if signature is None:
|
|
return None
|
|
count = seen_counts.get(signature, 0) + 1
|
|
seen_counts[signature] = count
|
|
if count <= _MAX_REPEAT_EXTERNAL_LOOKUPS:
|
|
return None
|
|
logger.warning(
|
|
"Blocking repeated external lookup {} on attempt {}",
|
|
signature[:160],
|
|
count,
|
|
)
|
|
return (
|
|
"Error: repeated external lookup blocked. "
|
|
"Use the results you already have to answer, or try a meaningfully different source."
|
|
)
|