Spaces:
Running
Running
File size: 3,667 Bytes
4036608 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | #!/usr/bin/env python3
"""
patch_web_search_fallback.py — DuckDuckGo 免费 fallback
当所有付费搜索后端（Firecrawl / Tavily / Exa / Parallel）都不可用时，
自动 fallback 到 DuckDuckGo 搜索，无需任何 API key。
Monkey-patch 目标（tools/web_tools.py）：
- _get_backend() : 无可用后端时返回 "duckduckgo"
- web_search_tool() : 分发 duckduckgo 后端
- _is_backend_available(): duckduckgo 始终返回 True
"""
import json
import logging
logger = logging.getLogger(__name__)
def _ddg_search(query: str, limit: int = 5) -> dict:
"""通过 DuckDuckGo 搜索，返回标准格式结果。"""
try:
from duckduckgo_search import DDGS
results = []
with DDGS() as ddgs:
for i, r in enumerate(ddgs.text(query, max_results=limit)):
results.append({
"title": r.get("title", ""),
"url": r.get("href", ""),
"description": r.get("body", ""),
"position": i + 1,
})
return {"success": True, "data": {"web": results}}
except Exception as e:
logger.error("DuckDuckGo search failed: %s", e)
return {"success": False, "error": f"DuckDuckGo search failed: {e}"}
def patch():
"""Apply the DuckDuckGo fallback patch to web_tools."""
import tools.web_tools as wt
# ── 1. 让 _get_backend() 在无可用后端时返回 "duckduckgo" ──
_original_get_backend = wt._get_backend
def _get_backend_patched():
configured = (wt._load_web_config().get("backend") or "").lower().strip()
# 用户显式指定了某个后端 → 尊重配置（即使 key 可能无效）
if configured in ("parallel", "firecrawl", "tavily", "exa"):
return configured
# 按优先级检查哪个后端可用
for backend in ("firecrawl", "parallel", "tavily", "exa"):
if wt._is_backend_available(backend):
return backend
# 全都不可用 → DuckDuckGo 兜底
return "duckduckgo"
wt._get_backend = _get_backend_patched
# ── 2. 在 web_search_tool() 中添加 duckduckgo 分发 ──
_original_web_search = wt.web_search_tool
def web_search_patched(query: str, limit: int = 5) -> str:
backend = wt._get_backend()
if backend == "duckduckgo":
logger.info("DuckDuckGo search (free fallback): '%s' (limit: %d)", query, limit)
response_data = _ddg_search(query, limit)
return json.dumps(response_data, indent=2, ensure_ascii=False)
# 其他后端走原逻辑
return _original_web_search(query, limit)
wt.web_search_tool = web_search_patched
# ── 3. 让 _is_backend_available("duckduckgo") 始终返回 True ──
_original_is_backend_available = wt._is_backend_available
def _is_backend_available_patched(backend: str) -> bool:
if backend == "duckduckgo":
return True
return _original_is_backend_available(backend)
wt._is_backend_available = _is_backend_available_patched
# ── 4. 让 web_extract 也能在 duckduckgo 后端下工作 ──
# web_extract 不依赖搜索后端（它直接抓 URL），但如果 _get_backend
# 返回 "duckduckgo" 且 web_extract 内部也调用 _get_backend，需要兼容。
# 目前 web_extract 不调用 _get_backend，所以不需要 patch。
logger.info(
"Web search: DuckDuckGo free fallback enabled "
"(no API key required, auto-activates when paid backends unavailable)"
)
if __name__ == "__main__":
patch()
|