vent_deep_agent/tools/web_search_tools.py

# -*- coding: utf-8 -*-
"""
联网搜索工具模块

提供两大能力：
- web_search: 通过 DuckDuckGo 搜索互联网公开信息
- web_fetch:   抓取指定网页并提取正文文本

所有工具函数供 DeepAgent 调用，返回结构化 JSON 文本，由 LLM 解析并生成自然语言回复。
"""

import asyncio
import json
import re
import time

import httpx
from ddgs import DDGS

# ── 搜索重试配置 ──
_MAX_RETRIES = 3            # 最大重试次数
_RETRY_BASE_DELAY = 2.0     # 重试基础延迟（秒），按指数增长：2 → 4 → 8
_SEARCH_TIMEOUT = 20.0      # 单次搜索超时（秒）


async def web_search(query: str, max_results: int = 5) -> str:
    """搜索互联网获取公开信息。当用户询问的问题超出煤矿通风专业知识库范围、
    或需要了解最新政策/新闻/行业动态时调用此工具。
    
    优先策略：
    - 煤矿安全规程、技术标准 → 优先使用 query_knowledge_base（内部知识库）
    - 最新政策动态、行业新闻、通用知识 → 使用 web_search
    
    Args:
        query: 搜索关键词，建议使用简洁明确的中文词组
        max_results: 最大返回结果数，默认 5 条
    
    Returns:
        JSON 格式的搜索结果，包含 query 和 results 列表。
        每条结果含 title（标题）、href（链接）、body（摘要）。
    """
    if not query or not query.strip():
        return json.dumps({"error": "搜索关键词不能为空"}, ensure_ascii=False)

    q = query.strip()
    last_error = ""

    for attempt in range(1, _MAX_RETRIES + 1):
        try:
            # 在线程池中运行同步 DDGS 调用（避免阻塞事件循环），并设置超时
            results = await asyncio.wait_for(
                asyncio.to_thread(_do_search, q, max_results),
                timeout=_SEARCH_TIMEOUT,
            )

            formatted = []
            for r in results:
                formatted.append({
                    "title": r.get("title", ""),
                    "href": r.get("href", ""),
                    "body": r.get("body", ""),
                })

            return json.dumps(
                {"query": q, "results": formatted},
                ensure_ascii=False,
            )

        except asyncio.TimeoutError:
            last_error = f"搜索超时（>{_SEARCH_TIMEOUT}s）"
        except Exception as e:
            last_error = str(e)

        if attempt < _MAX_RETRIES:
            delay = _RETRY_BASE_DELAY ** attempt
            await asyncio.sleep(delay)

    return json.dumps(
        {"error": f"搜索失败（已重试{_MAX_RETRIES}次）: {last_error}", "query": q},
        ensure_ascii=False,
    )


def _do_search(query: str, max_results: int) -> list[dict]:
    """同步搜索辅助函数（在线程池中执行）。"""
    with DDGS() as ddgs:
        return list(ddgs.text(query, max_results=max_results))


async def web_fetch(url: str) -> str:
    """抓取指定网页并提取正文文本。通常在 web_search 获取到相关链接后，
    需要查看页面详细内容时调用。
    
    Args:
        url: 要抓取的网页完整 URL（含 https://）
    
    Returns:
        JSON 格式，包含 url、title（页面标题）、content（提取的正文文本）、
        content_length（正文字符数）。
        正文最多保留 8000 字符，超出部分截断。
    """
    if not url or not url.strip():
        return json.dumps({"error": "URL 不能为空"}, ensure_ascii=False)

    url = url.strip()
    try:
        async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
            resp = await client.get(
                url,
                headers={
                    "User-Agent": (
                        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                        "AppleWebKit/537.36 (KHTML, like Gecko) "
                        "Chrome/125.0.0.0 Safari/537.36"
                    ),
                    "Accept": "text/html,application/xhtml+xml",
                    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
                },
            )
            resp.raise_for_status()
            html = resp.text

    except httpx.HTTPStatusError as e:
        return json.dumps(
            {"error": f"请求失败 HTTP {e.response.status_code}", "url": url},
            ensure_ascii=False,
        )
    except httpx.TimeoutException:
        return json.dumps(
            {"error": "请求超时（15s）", "url": url},
            ensure_ascii=False,
        )
    except Exception as e:
        return json.dumps(
            {"error": f"抓取失败: {str(e)}", "url": url},
            ensure_ascii=False,
        )

    # ── 提取标题 ──
    title = ""
    title_match = re.search(r"<title[^>]*>(.*?)</title>", html, re.IGNORECASE | re.DOTALL)
    if title_match:
        title = title_match.group(1).strip()

    # ── 去除不需要的标签及其内容 ──
    for tag in ("script", "style", "nav", "footer", "header", "noscript", "iframe"):
        html = re.sub(
            rf"<{tag}[^>]*>.*?</{tag}>",
            "",
            html,
            flags=re.IGNORECASE | re.DOTALL,
        )

    # ── 去除所有 HTML 标签，提取纯文本 ──
    text = re.sub(r"<[^>]+>", " ", html)

    # ── 清理 HTML 实体 ──
    text = re.sub(r"&nbsp;", " ", text)
    text = re.sub(r"&amp;", "&", text)
    text = re.sub(r"&lt;", "<", text)
    text = re.sub(r"&gt;", ">", text)
    text = re.sub(r"&quot;", '"', text)
    text = re.sub(r"&#?\w+;", " ", text)
    text = re.sub(r"\s+", " ", text).strip()

    # ── 截断到 8000 字符 ──
    max_chars = 8000
    if len(text) > max_chars:
        text = text[:max_chars] + "…（内容已截断）"

    return json.dumps(
        {
            "url": url,
            "title": title,
            "content": text,
            "content_length": len(text),
        },
        ensure_ascii=False,
    )