Spaces:
Running
Running
| #!/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() | |