base_agent/tools/web_search.py

"""
tools/web_search.py
网络搜索工具 —— google搜索
配置通过 settings.tools['web_search'] 读取
"""

from dataclasses import dataclass
from serpapi import SerpApiClient
from agent.config.settings import settings
from agent.tools.base_tool import BaseTool
from agent.utils.logger import get_logger

logger = get_logger("TOOL.WebSearch")


def _cfg(key: str, fallback=None):
    return settings.tools['web_search'].get(key, fallback)


@dataclass
class SearchResult:
    title:   str
    url:     str
    snippet: str
    rank:    int = 0

    def __str__(self) -> str:
        return f"[{self.rank}] {self.title}\n    {self.url}\n    {self.snippet}"


class Tool(BaseTool):
    name        = "web_search"
    description = (
        "在互联网上搜索信息，返回相关网页的标题、链接和摘要。"
        "适用于需要实时信息、最新资讯或不确定的知识查询。"
    )
    parameters  = {
        "type": "object",
        "properties": {
            "query": {
                "type":        "string",
                "description": "搜索关键词或问题，例如: 'Python 3.12 新特性'",
            }
        },
        "required": ["query"],
    }

    def execute(self, query: str = "", **_) -> str:
        """
            一个基于SerpApi的实战网页搜索引擎工具。
            它会智能地解析搜索结果，优先返回直接答案或知识图谱信息。
            """
        print(f"🔍 正在执行 [SerpApi] 网页搜索: {query}")
        try:
            api_key = _cfg("api_key")
            if not api_key:
                return "错误:ApiKey未配置。"

            params = {
                "engine": "google",
                "q": query,
                "api_key": api_key,
                "gl": "cn",  # 国家代码
                "hl": "zh-cn",  # 语言代码
            }

            client = SerpApiClient(params)
            results = client.get_dict()

            # 智能解析:优先寻找最直接的答案
            if "answer_box_list" in results:
                return "\n".join(results["answer_box_list"])
            if "answer_box" in results and "answer" in results["answer_box"]:
                return results["answer_box"]["answer"]
            if "knowledge_graph" in results and "description" in results["knowledge_graph"]:
                return results["knowledge_graph"]["description"]
            if "organic_results" in results and results["organic_results"]:
                # 如果没有直接答案，则返回前三个有机结果的摘要
                snippets = [
                    f"[{i + 1}] {res.get('title', '')}\n{res.get('snippet', '')}"
                    for i, res in enumerate(results["organic_results"][:3])
                ]
                return "\n\n".join(snippets)

            return f"对不起，没有找到关于 '{query}' 的信息。"

        except Exception as e:
            return f"搜索时发生错误: {e}"