butler/tools/network_search_policy.py
Rama main · réplica de lectura del repositorio original.
"""Per-turn network search policy: web_search-first gate + Firecrawl search quota."""
from __future__ import annotations
import json
import re
from contextlib import contextmanager
from contextvars import ContextVar
from typing import Any, Iterator
from butler.utilities.env_parse import env_truthy, int_env
_CTX: ContextVar[dict[str, Any] | None] = ContextVar("network_search_turn", default=None)
_SEARCH_INTENT_RE = re.compile(
r"(搜(索|一下)?|查(一下|询)?|竞品|竟品|调研|检索|对比|排行|"
r"competitor|research|lookup|\bsearch\b)",
re.IGNORECASE,
)
def network_search_gate_enabled() -> bool:
return bool(env_truthy("BUTLER_NETWORK_SEARCH_GATE", default=True))
def max_firecrawl_search_per_turn() -> int:
try:
return max(0, int(int_env("BUTLER_FIRECRAWL_SEARCH_MAX_PER_TURN", 3, min=0)))
except ValueError:
return 3
def max_web_search_empty_per_turn() -> int:
try:
return max(0, int(int_env("BUTLER_WEB_SEARCH_EMPTY_MAX_PER_TURN", 2, min=0)))
except ValueError:
return 2
def max_firecrawl_agent_per_turn() -> int:
try:
return max(0, int(int_env("BUTLER_FIRECRAWL_AGENT_MAX_PER_TURN", 0, min=0)))
except ValueError:
return 0
def max_firecrawl_feedback_per_turn() -> int:
try:
return max(0, int(int_env("BUTLER_FIRECRAWL_FEEDBACK_MAX_PER_TURN", 0, min=0)))
except ValueError:
return 0
def is_firecrawl_agent_tool(tool_name: str) -> bool:
name = str(tool_name or "").lower()
return "firecrawl" in name and "agent" in name
def is_firecrawl_feedback_tool(tool_name: str) -> bool:
name = str(tool_name or "").lower()
return "firecrawl" in name and "feedback" in name
def is_firecrawl_search_tool(tool_name: str) -> bool:
name = str(tool_name or "").lower()
if "feedback" in name or "agent" in name:
return False
return "firecrawl" in name and "search" in name
def is_todoist_mcp_intent(query: str) -> bool:
"""Todoist 列项目/任务应走 MCP，不走 web 检索或 API 文档臆造。"""
text = str(query or "").strip().lower()
if "todoist" not in text:
return False
hints = (
"项目", "待办", "任务", "inbox", "列出", "哪些", "今天", "明天",
"project", "task", "list",
)
return any(h in text for h in hints)
def is_github_mcp_intent(query: str) -> bool:
"""GitHub 列仓库/issues 应走 MCP，不走 web 检索。"""
from butler.tools.network_search_policy_ops import is_github_mcp_intent_safe
return bool(is_github_mcp_intent_safe(query))
def _github_mcp_block(tool_name: str) -> dict[str, Any] | None:
if not is_github_mcp_intent(_turn_user_query()):
return None
name = str(tool_name or "").strip().lower()
if is_web_search_tool(name) or name == "web_fetch":
return {
"ok": False,
"code": "GITHUB_USE_MCP",
"error": (
"GitHub 仓库/issues 须调用 mcp_github_lst_repos_authenticated_usr / "
"mcp_github_lst_repo_issues 等 MCP 工具，禁止 web_search 或 web_fetch。"
),
"hint": "mcp_github_lst_repos_authenticated_usr · mcp_github_lst_repo_issues",
}
return None
def _todoist_mcp_block(tool_name: str) -> dict[str, Any] | None:
if not is_todoist_mcp_intent(_turn_user_query()):
return None
name = str(tool_name or "").strip().lower()
if is_web_search_tool(name) or name == "web_fetch":
return {
"ok": False,
"code": "TODOIST_USE_MCP",
"error": (
"Todoist 查询须调用 mcp_todoist_lst_projects / mcp_todoist_lst_tasks 等 MCP 工具，"
"禁止 web_search 或 web_fetch 查 API 文档。"
),
"hint": "mcp_todoist_lst_projects · mcp_todoist_lst_tasks",
}
return None
def is_web_search_tool(tool_name: str) -> bool:
return str(tool_name or "").strip().lower() == "web_search"
def is_web_search_intent(query: str) -> bool:
text = str(query or "").strip()
if not text:
return False
if is_todoist_mcp_intent(text):
return False
if _SEARCH_INTENT_RE.search(text):
return True
if "## 相关知识" in text:
tail = text.split("## 相关知识", 1)[0].strip()
if tail and _SEARCH_INTENT_RE.search(tail):
return True
return False
@contextmanager
def turn_network_search_scope(inbound_user_text: str = "") -> Iterator[None]:
"""Reset per-turn counters (agent loop); gateway also resets bridge fields."""
token = _CTX.set(
{
"web_search_used": False,
"firecrawl_search_count": 0,
"firecrawl_agent_count": 0,
"firecrawl_feedback_count": 0,
"web_search_empty_count": 0,
"inbound_user_text": str(inbound_user_text or "").strip(),
}
)
try:
yield
finally:
_CTX.reset(token)
def _bridge_state() -> Any | None:
from butler.tools.network_search_policy_ops import get_turn_bridge_safe
return get_turn_bridge_safe()
def _ctx_state() -> dict[str, Any]:
bucket = _CTX.get()
if bucket is None:
bucket = {
"web_search_used": False,
"firecrawl_search_count": 0,
"firecrawl_agent_count": 0,
"firecrawl_feedback_count": 0,
"web_search_empty_count": 0,
"inbound_user_text": "",
}
_CTX.set(bucket)
return bucket
def _web_search_used() -> bool:
bridge = _bridge_state()
if bridge is not None:
return bool(getattr(bridge, "network_search_web_used", False))
return bool(_ctx_state().get("web_search_used"))
def _set_web_search_used() -> None:
bridge = _bridge_state()
if bridge is not None:
bridge.network_search_web_used = True
_ctx_state()["web_search_used"] = True
def _firecrawl_search_count() -> int:
bridge = _bridge_state()
if bridge is not None:
return int(getattr(bridge, "firecrawl_search_count", 0) or 0)
return int(_ctx_state().get("firecrawl_search_count") or 0)
def _inc_firecrawl_search_count() -> int:
bridge = _bridge_state()
if bridge is not None:
bridge.firecrawl_search_count = int(getattr(bridge, "firecrawl_search_count", 0) or 0) + 1
return int(bridge.firecrawl_search_count)
bucket = _ctx_state()
bucket["firecrawl_search_count"] = int(bucket.get("firecrawl_search_count") or 0) + 1
return int(bucket["firecrawl_search_count"])
def _firecrawl_agent_count() -> int:
bridge = _bridge_state()
if bridge is not None:
return int(getattr(bridge, "firecrawl_agent_count", 0) or 0)
return int(_ctx_state().get("firecrawl_agent_count") or 0)
def _inc_firecrawl_agent_count() -> int:
bridge = _bridge_state()
if bridge is not None:
bridge.firecrawl_agent_count = int(getattr(bridge, "firecrawl_agent_count", 0) or 0) + 1
return int(bridge.firecrawl_agent_count)
bucket = _ctx_state()
bucket["firecrawl_agent_count"] = int(bucket.get("firecrawl_agent_count") or 0) + 1
return int(bucket["firecrawl_agent_count"])
def _firecrawl_feedback_count() -> int:
bridge = _bridge_state()
if bridge is not None:
return int(getattr(bridge, "firecrawl_feedback_count", 0) or 0)
return int(_ctx_state().get("firecrawl_feedback_count") or 0)
def _inc_firecrawl_feedback_count() -> int:
bridge = _bridge_state()
if bridge is not None:
bridge.firecrawl_feedback_count = int(getattr(bridge, "firecrawl_feedback_count", 0) or 0) + 1
return int(bridge.firecrawl_feedback_count)
bucket = _ctx_state()
bucket["firecrawl_feedback_count"] = int(bucket.get("firecrawl_feedback_count") or 0) + 1
return int(bucket["firecrawl_feedback_count"])
def _web_search_empty_count() -> int:
bridge = _bridge_state()
if bridge is not None:
return int(getattr(bridge, "web_search_empty_count", 0) or 0)
return int(_ctx_state().get("web_search_empty_count") or 0)
def _inc_web_search_empty_count() -> int:
bridge = _bridge_state()
if bridge is not None:
bridge.web_search_empty_count = int(getattr(bridge, "web_search_empty_count", 0) or 0) + 1
return int(bridge.web_search_empty_count)
bucket = _ctx_state()
bucket["web_search_empty_count"] = int(bucket.get("web_search_empty_count") or 0) + 1
return int(bucket["web_search_empty_count"])
def _turn_user_query() -> str:
bridge = _bridge_state()
if bridge is not None:
text = str(getattr(bridge, "inbound_user_text", "") or "").strip()
if text:
return text
text = str(_ctx_state().get("inbound_user_text") or "").strip()
if text:
return text
from butler.tools.network_search_policy_ops import epoch_user_query_safe
return str(epoch_user_query_safe())
def _web_search_in_current_toolset() -> bool:
from butler.tools.web_search import web_search_enabled
from butler.tools.network_search_policy_ops import web_search_in_current_toolset_safe
if not web_search_enabled():
return False
return bool(web_search_in_current_toolset_safe(fallback=web_search_enabled()))
def _gate_active() -> bool:
return network_search_gate_enabled() and _web_search_in_current_toolset()
def _firecrawl_aux_block(tool_name: str, *, disabled_code: str, disabled_msg: str, quota_code: str) -> dict[str, Any] | None:
cap = max_firecrawl_agent_per_turn() if is_firecrawl_agent_tool(tool_name) else max_firecrawl_feedback_per_turn()
if cap <= 0:
return {
"ok": False,
"code": disabled_code,
"error": disabled_msg,
"hint": "勿向用户提及审批流程；用 firecrawl_search 结果的 URL 做 scrape 或直接总结",
}
count = _firecrawl_agent_count() if is_firecrawl_agent_tool(tool_name) else _firecrawl_feedback_count()
if count >= cap:
return {
"ok": False,
"code": quota_code,
"error": f"本轮该 Firecrawl 工具已达上限（{cap} 次）；请 scrape 已有 URL 或直接总结。",
"hint": "勿向用户提及审批流程",
}
return None
def _firecrawl_gate_block(tool_name: str) -> dict[str, Any] | None:
"""Shared gate for Firecrawl search/agent/feedback on retrieval-intent turns."""
if not is_web_search_intent(_turn_user_query()):
return None
if not _web_search_used():
return {
"ok": False,
"code": "WEB_SEARCH_REQUIRED",
"error": (
"本轮须先用 web_search 检索链接，再调用 Firecrawl search/scrape；"
"请改用 web_search(query=...) 获取 URL 列表。"
),
"hint": "deep-research：web_search → web_fetch / mcp_firecrawl_scrape",
}
if is_firecrawl_feedback_tool(tool_name):
return _firecrawl_aux_block(
tool_name,
disabled_code="FIRECRAWL_FEEDBACK_DISABLED",
disabled_msg=(
"检索任务禁止 mcp_firecrawl_*_feedback；"
"请根据 firecrawl_search 结果 scrape URL 或直接总结。"
),
quota_code="FIRECRAWL_FEEDBACK_QUOTA",
)
if is_firecrawl_agent_tool(tool_name):
return _firecrawl_aux_block(
tool_name,
disabled_code="FIRECRAWL_AGENT_DISABLED",
disabled_msg=(
"检索任务禁止 mcp_firecrawl_*_agent（高成本）；"
"请根据 firecrawl_search 结果 scrape 已有 URL 或直接总结。"
),
quota_code="FIRECRAWL_AGENT_QUOTA",
)
cap = max_firecrawl_search_per_turn()
if cap >= 0 and _firecrawl_search_count() >= cap:
return {
"ok": False,
"code": "FIRECRAWL_SEARCH_QUOTA",
"error": (
f"本轮 mcp_firecrawl_*_search 已达上限（{cap} 次）；"
"请根据已有搜索结果选用 URL 做 scrape，或直接总结。"
),
"hint": "优先 mcp_firecrawl_scrape 已有 URL，勿调 feedback/agent",
}
return None
def check_network_search_tool_block(tool_name: str, args: dict[str, Any] | None = None) -> dict[str, Any] | None:
"""Return error payload if tool call violates network search policy."""
_ = args
name = str(tool_name or "").strip()
github_block = _github_mcp_block(name)
if github_block is not None:
return github_block
todoist_block = _todoist_mcp_block(name)
if todoist_block is not None:
return todoist_block
if not name or not _gate_active():
return None
if is_web_search_tool(name):
if is_web_search_intent(_turn_user_query()):
cap = max_web_search_empty_per_turn()
if cap >= 0 and _web_search_empty_count() >= cap:
return {
"ok": False,
"code": "WEB_SEARCH_EXHAUSTED",
"error": (
f"本轮 web_search 已连续空结果 {cap} 次；"
"请改用 mcp_firecrawl_firecrawl_search / scrape 或直接总结。"
),
"hint": "勿再重复 web_search；Firecrawl 已在 web_search 之后可用",
}
return None
if (
not is_firecrawl_search_tool(name)
and not is_firecrawl_agent_tool(name)
and not is_firecrawl_feedback_tool(name)
):
return None
return _firecrawl_gate_block(name)
def record_network_search_tool(tool_name: str) -> None:
"""Record web_search / firecrawl search/agent/feedback usage for the current turn."""
name = str(tool_name or "").strip()
if is_web_search_tool(name):
_set_web_search_used()
elif is_firecrawl_feedback_tool(name):
_inc_firecrawl_feedback_count()
elif is_firecrawl_agent_tool(name):
_inc_firecrawl_agent_count()
elif is_firecrawl_search_tool(name):
_inc_firecrawl_search_count()
def note_web_search_outcome(result_text: str) -> None:
"""Track empty web_search results to cap pointless retries."""
text = str(result_text or "").strip()
if not text:
_inc_web_search_empty_count()
return
try:
payload = json.loads(text)
except json.JSONDecodeError:
return
if not isinstance(payload, dict):
return
if payload.get("error"):
_inc_web_search_empty_count()
return
results = payload.get("results")
if isinstance(results, list) and len(results) == 0:
_inc_web_search_empty_count()
def should_run_network_search(query: str) -> bool:
"""Check if network search should be performed for the given query."""
return network_search_gate_enabled() and is_web_search_intent(query)
def run_network_search(query: str) -> list[dict[str, Any]]:
"""Run web search for the given query."""
from butler.tools.web_search import web_search
try:
result = web_search(query=query)
if not result:
return []
try:
payload = json.loads(result)
if isinstance(payload, dict) and isinstance(payload.get("results"), list):
return payload["results"]
except json.JSONDecodeError:
pass
except Exception:
pass
return []
__all__ = [
"check_network_search_tool_block",
"is_firecrawl_agent_tool",
"is_firecrawl_feedback_tool",
"is_firecrawl_search_tool",
"is_github_mcp_intent",
"is_web_search_intent",
"is_web_search_tool",
"max_firecrawl_agent_per_turn",
"max_firecrawl_feedback_per_turn",
"max_firecrawl_search_per_turn",
"max_web_search_empty_per_turn",
"network_search_gate_enabled",
"note_web_search_outcome",
"record_network_search_tool",
"run_network_search",
"should_run_network_search",
"turn_network_search_scope",
]