大模型技术日新月异的今天,开发者在构建智能代理(Agent)或检索增强生成(RAG)流水线时,往往会依赖商业化的搜索引擎API,如SerpAPI或Bing Search。这些API功能强大,但同时也面临着诸多限制,如按量计费、速率限制以及最关键的:数据泄露风险。随着本地大模型的普及,我们的搜索基础设施也应该尽可能地靠近模型,实现数据的本地化和私有化。

本文将带你一步步搭建本地化的 SearxNG 容器——一款开源的元搜索引擎。通过它,你的应用可以进行实时的网络搜索,而无需将任何数据泄露到外部世界。同时,我们还将展示如何直接从 Python 中消费这些数据,以便你的 Agent 和 大模型 能够安全、离线且高效地使用。

SerpAPI 的替代方案:为什么选择本地化 SearxNG?

SerpAPI 及其它类似的商业服务,虽然方便易用,但却存在着无法忽视的缺点:

  • 成本高昂:按查询次数收费,或需要订阅服务,长期使用成本不菲。
  • 限制颇多:受限于其配额、API 端点以及服务可用性,灵活性不足。
  • 缺乏隐私:查询数据发送到第三方服务,打破了本地优先的隐私模型。尤其对于处理敏感数据的应用来说,这是不可接受的。
  • 数据不透明:你无法得知数据是如何被过滤或跟踪的,存在潜在的安全风险。

举例来说,一个金融行业的应用需要利用网络数据进行风险评估。如果使用 SerpAPI,每次查询客户信息都将暴露给第三方,这显然违反了金融行业的合规要求。而使用本地化的 SearxNG,则可以确保所有数据都留在本地服务器上,满足严格的隐私保护需求。

案例:某医疗机构使用 大模型 进行辅助诊断,需要检索大量的医学文献。使用 SerpAPI 会将患者的关键词暴露给第三方,存在泄露隐私的风险。而本地部署 SearxNG 后,所有的检索行为都发生在机构内部网络,有效保护了患者的隐私。

SearxNG:开源元搜索引擎的强大力量

SearxNG 是一款尊重隐私的元搜索引擎,它可以聚合来自多个搜索引擎(如 Google、Bing、Brave、Wikipedia 等)的结果,并提供干净、可定制的 JSON API。这意味着你可以根据自己的需求,选择不同的搜索引擎作为数据来源,并对结果进行自定义排序和过滤。

数据:SearxNG 支持超过 70 种不同的搜索引擎,覆盖了新闻、学术、社交媒体等各个领域。你可以根据自己的需求,选择最合适的搜索引擎组合,获得更全面、更准确的搜索结果。

与商业搜索引擎API相比,SearxNG 拥有以下优势:

  • 完全免费:无需支付任何费用,即可享受强大的搜索功能。
  • 开源透明:代码完全开源,你可以自由地查看、修改和定制,完全掌控你的数据。
  • 尊重隐私:不会跟踪你的搜索行为,保护你的隐私。
  • 高度可定制:可以自定义搜索引擎列表、结果排序方式、API 输出格式等,满足不同的应用场景。

搭建本地 SearxNG 容器:一步到位

下面我们将演示如何在 Docker 中运行 SearxNG。

Step 1: 使用 Docker 部署 SearxNG

保存以下 docker-compose-searxng.yml 文件:

services:
  searxng:
    image: searxng/searxng:latest
    restart: unless-stopped
    ports:
      - "0.0.0.0:8089:8080"  # Access via http://localhost:8089
    networks:
      - app_network
    volumes:
      - ${SEARXNG_DATA}:/etc/searxng:rw
      - ${SEARXNG_SETTINGS}:/etc/searxng/settings.yml:ro
    environment:
      - BASE_URL=http://localhost:8080
      - REDIS_URL=redis://host.docker.internal:6379/0
      - UWSGI_WORKERS=4
      - UWSGI_THREADS=4
    cap_drop: ["ALL"]
    cap_add: ["CHOWN", "SETGID", "SETUID"]
    ulimits:
      nofile:
        soft: 65535
        hard: 65535
networks:
  app_network:
    external: true

你可以通过 VSCode 等 GUI 工具点击文件 > Compose Up 来运行代码,或者通过 CMD 运行:

# 创建 docker 网络 (如果不存在):
docker network create --driver bridge app_network

# 运行容器
docker-compose up -d

使用 curl 命令检查容器是否启动成功:

# 从主机
curl http://localhost:8089/search?q=rag+agents&format=json

# 或者从容器内部
curl http://host.docker.internal:8089/search

使用浏览器在主机上打开 http://localhost:8089,即可访问 SearxNG 的 Web 界面。

Step 2: 从 Python 访问 SearxNG

以下是一个简单的异步包装器,可以获取多个页面的结果,并将它们作为结构化数据返回,供你的 Agent 流水线或 大模型 输入使用:

import os
import httpx
import asyncio
from datetime import datetime
from pydantic import BaseModel, Field
from typing import List, Optional, Union

class WebsiteData(BaseModel):
    """Represents structured data extracted from a website."""
    url: str
    title: Optional[str] = None
    snippet: Optional[str] = None
    content: Optional[str] = None
    author: Optional[str] = None
    date: Optional[str] = None
    language: Optional[str] = None
    engine: Optional[str] = None
    engines: Optional[List[str]] = None
    score: Optional[float] = None
    category: Optional[str] = None
    published_date: Optional[str] = None
    extracted_at: Optional[str] = None
    success: bool = True

class WebsiteError(BaseModel):
    """Represents an error encountered during website data extraction."""
    query: str
    error: str
    extracted_at: datetime = Field(default_factory=datetime.now)
    success: bool = False

class WebsiteSearch(BaseModel):
    """Performs website search and result extraction using a SearxNG instance."""
    class Config:
        arbitrary_types_allowed = True

    async def _fetch_page(
        self,
        client: httpx.AsyncClient,
        base_url: str,
        query: str,
        page: int,
        timeout: int,
    ):
        """
        Fetch a single page of search results from the SearxNG instance.
        :param client: httpx.AsyncClient instance for making HTTP requests.
        :param base_url: Base URL of the SearxNG search endpoint.
        :param query: Search query string.
        :param page: Page number to fetch.
        :param timeout: Timeout in seconds for the HTTP request.
        :return: A list of search result items or an error message string.
        :rtype: Union[List[dict], str]
        """
        try:
            params = {"q": query, "format": "json", "pageno": page}
            response = await client.get(base_url, params=params, timeout=timeout)
            response.raise_for_status()
            page_json = response.json()
            return page_json.get("results", [])
        except Exception as e:
            return f"Page {page} failed: {e}"

    async def _async_search(
        self,
        query: str,
        max_page_number: int,
        timeout: int,
        max_results: Optional[int],
        output_for_llm: bool,
    ) -> Union[List[WebsiteData] | str, WebsiteError | str]:
        """
        Perform an asynchronous search across multiple pages and return structured results.
        :param query: Search query string.
        :param max_page_number: Number of result pages to fetch.
        :param timeout: Timeout in seconds for each HTTP request.
        :param max_results: Maximum number of results to return (after sorting by score).
        :param output_for_llm: Whether to format results as newline-separated JSON strings.
        :return: A list of WebsiteData or a WebsiteError, or formatted string if output_for_llm is True.
        :rtype: Union[List[WebsiteData], str, WebsiteError]
        :raises ValueError if SEARXNG_BASE_URL is not set.
        """
        base_url: str = os.getenv("SEARXNG_BASE_URL")
        if not base_url:
            raise ValueError("SEARXNG_BASE_URL environment variable is not set.")

        all_data: List[WebsiteData] = []
        async with httpx.AsyncClient() as client:
            tasks = [
                self._fetch_page(client, base_url, query, page, timeout)
                for page in range(1, max_page_number + 1)
            ]
            pages_results = await asyncio.gather(*tasks)

        for result in pages_results:
            if isinstance(result, str):
                return (
                    result
                    if output_for_llm
                    else WebsiteError(query=query, error=result)
                )
            for item in result:
                all_data.append(
                    WebsiteData(
                        url=item.get("url"),
                        title=item.get("title"),
                        snippet=item.get("snippet"),
                        content=item.get("content"),
                        author=item.get("author"),
                        date=f"{item.get('date')}",
                        language=item.get("language"),
                        engine=item.get("engine"),
                        engines=item.get("engines"),
                        score=item.get("score"),
                        category=item.get("category"),
                        published_date=f"{item.get('published_date')}",
                        extracted_at=datetime.now().isoformat(),
                        success=True,
                    )
                )

        if max_results:
            all_data.sort(
                key=lambda d: d.score if d.score is not None else 0, reverse=True
            )
            all_data = all_data[:max_results]

        if output_for_llm:
            return "\n".join(d.model_dump_json() for d in all_data)

        return all_data

    def search(
        self,
        query: str,
        timeout: int = 10,
        max_page_number: int = 5,
        output_for_llm: bool = False,
        max_results: Optional[int] = None
    ) -> Union[List[WebsiteData] | str, WebsiteError | str]:
        """
        Synchronously performs a search query using asyncio and returns results.
        :param query: Search query string.
        :param max_page_number: Number of result pages to retrieve (default is 5).
        :param timeout: Timeout in seconds for each request (default is 10).
        :param max_results: Maximum number of results to return (optional).
        :param output_for_llm: If True, returns newline-separated JSON strings for LLM input.
        :return: A list of WebsiteData, or WebsiteError on failure; or formatted string if output_for_llm is True.
        :rtype: Union[List[WebsiteData], str, WebsiteError]
        """
        return asyncio.run(
            self._async_search(
                query, max_page_number, timeout, max_results, output_for_llm
            )
        )

result = WebsiteSearch().search(query="RAG for agents", output_for_llm=True, max_results=2)
print(result)

这段代码的背后,它会向本地 SearxNG 实例发送多个分页请求,使用 Pydantic 包装数据,并将数据格式化为 JSON 对象或换行符分隔的字符串,非常适合提供给 大模型 使用。

实际应用案例

假设你正在开发一个智能客服机器人,需要能够回答用户关于产品信息的问题。你可以使用本地 SearxNG 检索公司网站和产品文档,并将结果传递给 大模型,让它生成自然流畅的回答。

具体步骤如下:

  1. 用户提出问题:“你们公司的最新产品是什么?”
  2. 智能客服机器人使用 SearxNG 在公司网站和产品文档中搜索相关信息。
  3. SearxNG 返回搜索结果,包括产品名称、功能介绍、用户评价等。
  4. 机器人将搜索结果传递给 大模型
  5. 大模型 分析搜索结果,提取关键信息,生成简洁明了的回答:“我们公司最新产品是XXX,它具有XXX功能,深受用户好评。”

通过这种方式,智能客服机器人可以利用本地数据和 大模型 的能力,为用户提供准确、及时的解答,提升用户体验。

适用场景:本地化 SearxNG 的最佳实践

本地化 SearxNG 适用于以下场景:

  • 本地运行大模型:如果你正在本地运行 大模型(例如,通过 Ollama 或 LM Studio),那么本地化的 SearxNG 可以与你的模型无缝集成,避免数据泄露的风险。
  • 构建 RAG 或检索增强 Agent:RAG 和检索增强 Agent 需要从外部知识库中检索信息。使用本地 SearxNG 可以确保检索过程的隐私性和安全性。
  • 对隐私要求严格的工具原型设计:如果你正在开发对隐私要求非常严格的工具,例如医疗健康、金融等领域的应用,那么本地 SearxNG 是一个理想的选择。
  • 摆脱 SerpAPI 的速率限制:如果你受够了 SerpAPI 的速率限制,想要更自由地进行网络搜索,那么本地 SearxNG 可以让你摆脱这些限制。

最终思考:掌控你的数据,掌控你的AI

现代 AI 工作流程应尊重开发者的隐私、控制权和成本效益。虽然 SerpAPI 和其他商业服务在原型设计阶段可能很有用,但它们最终只是一个拐杖。使用 SearxNG,你可以完全消除这种依赖。

它开放、快速、易于修改,并且可以与你的本地模型完美配合。

数据安全的重要性

大模型 时代,数据安全至关重要。如果你的数据泄露,可能会面临以下风险:

  • 商业机密泄露:竞争对手可能会窃取你的商业机密,导致经济损失。
  • 用户隐私泄露:用户的个人信息可能会被滥用,损害他们的权益。
  • 法律风险:违反数据保护法规可能会导致巨额罚款。

因此,保护数据安全是每个开发者和企业的责任。本地化 SearxNG 可以帮助你构建更安全、更可靠的 AI 应用。

展望未来

在下一篇文章中,我们将把 OpenWebUI 与本地 Ollama 模型集成,并将自定义工具(如这个搜索包装器)公开为模型可以使用的原生函数,例如 LLaMA 3.2。一切都在本地。没有 API 成本。敬请期待!

如果你有兴趣深入了解如何将此设置与 LangChain 或 Ollama 等工具集成,请关注我的下一篇文章。

你还可以关注我 @gabrielrodewald,了解更多关于 LLM 运维和开源 AI 工作流程的信息。通过 本地化 SearxNG,构建完全自主可控的 大模型 应用,拥抱 私有化AI 的未来。