Merge pull request #3539 from xinnan-tech/py-Outcrawl

Py outcrawl

docs/web-search-integration.md

@@ -2,14 +2,15 @@
 
 ## 功能简介
 
-联网搜索插件 `web_search` 支持在对话过程中实时联网搜索信息并返回结果。插件支持三个搜索源：秘塔（Metaso）、Tavily和Serply，用户可根据需要选择其中一个。
+联网搜索插件 `web_search` 支持在对话过程中实时联网搜索信息并返回结果。插件支持四个搜索源：秘塔（Metaso）、Tavily、Serply和Outcrawl，用户可根据需要选择其中一个。
 
 ## API Key申请指南
 
-目前我们适配了`秘塔搜索`、`Tavily搜索`和`Serply搜索`。
+目前我们适配了`秘塔搜索`、`Tavily搜索`、`Serply搜索`和`Outcrawl搜索`。
 - Tavily搜索：每个月1000次免费额度。
 - 秘塔搜索：拥有较为优质的国内数据源。
 - Serply搜索：返回谷歌搜索结果，注册赠送2500次免费额度，额度用完后需要付费。
+- Outcrawl搜索：面向AI智能体的网页搜索接口，每月1000 credits免费额度。
 
 ## API Key申请指南
 
@@ -32,6 +33,13 @@
 - 复制生成的API Key，这是配置所需的关键信息
 - 接口说明见 [Serply文档](https://serply.io/docs)
 
+### 方式四：使用Outcrawl搜索
+
+- 访问 [Outcrawl](https://outcrawl.ai)，注册并登录账号
+- 在控制台中创建API Key
+- 复制生成的API Key，这是配置所需的关键信息
+- 接口说明见 [Outcrawl文档](https://outcrawl.ai/docs/search/overview)
+
 ## 配置方式
 
 ### 方式1. 使用智控台部署（推荐）
@@ -40,14 +48,14 @@
 - 进入"配置角色"页面，选择要配置的智能体
 - 点击"编辑功能"按钮，在右侧参数配置区域找到"联网搜索"插件
 - 勾选"联网搜索"
-- 填入搜索源（`metaso`、`tavily`或`serply`），并将对应的`API Key`填入配置项
+- 填入搜索源（`metaso`、`tavily`、`serply`或`outcrawl`），并将对应的`API Key`填入配置项
 - 保存配置，再保存智能体配置
 
 ### 方式2. 单模块xiaozhi-server部署
 
 在 `data/.config.yaml` 中配置：
 
-- 将搜索源填入 `provider`，可选值为 `metaso`、`tavily` 或 `serply`
+- 将搜索源填入 `provider`，可选值为 `metaso`、`tavily`、`serply` 或 `outcrawl`
 - 将申请到的API Key填入 `api_key`
 
 ```yaml

main/manager-api/src/main/resources/db/changelog/202611021200.sql

@@ -0,0 +1,9 @@
+-- 联网搜索插件新增Outcrawl搜索源
+
+UPDATE `ai_model_provider`
+SET `fields` = JSON_SET(
+    `fields`,
+    '$[0].label',
+    '搜索源：metaso / tavily / serply / outcrawl'
+)
+WHERE `id` = 'SYSTEM_PLUGIN_WEB_SEARCH';

main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml

@@ -760,3 +760,10 @@ databaseChangeLog:
         - sqlFile:
             encoding: utf8
             path: classpath:db/changelog/202610302051.sql
+  - changeSet:
+      id: 202611021200
+      author: heydryft
+      changes:
+        - sqlFile:
+            encoding: utf8
+            path: classpath:db/changelog/202611021200.sql

main/xiaozhi-server/config.yaml

@@ -172,18 +172,19 @@ plugins:
     # ragflow知识库id
     dataset_ids: ["123456789"]
   # 联网搜索插件配置
-  # 支持秘塔(metaso)、Tavily和Serply三个搜索源
+  # 支持秘塔(metaso)、Tavily、Serply和Outcrawl四个搜索源
   # 秘塔API Key申请：https://metaso.cn/search-api/api-keys
   # Tavily API Key申请：https://app.tavily.com/home
   # Serply API Key申请：https://serply.io
+  # Outcrawl API Key申请：https://outcrawl.ai
   web_search:
-    # 搜索源：metaso / tavily / serply
+    # 搜索源：metaso / tavily / serply / outcrawl
     provider: "metaso"
     # 工具描述，用于告知LLM何时调用此工具，可根据需要自定义
     description: "联网搜索工具。当用户明确需要联网搜索问题时使用此工具。"
     # 返回结果数量（默认5）
     max_results: 5
-    # 秘塔API Key（mk-前缀） / Tavily API Key（tvly-前缀） / Serply API Key
+    # 秘塔API Key（mk-前缀） / Tavily API Key（tvly-前缀） / Serply API Key / Outcrawl API Key
     api_key: "mk-xxx"
 # 声纹识别配置
 voiceprint:

main/xiaozhi-server/plugins_func/functions/web_search.py

@@ -149,6 +149,54 @@ async def _search_serply(api_key: str, query: str, max_results: int) -> str:
     return "\n".join(lines)
 
 
+async def _search_outcrawl(api_key: str, query: str, max_results: int) -> str:
+    """调用Outcrawl搜索API"""
+    url = "https://api.outcrawl.ai/v1/search"
+    headers = {
+        "Authorization": f"Bearer {api_key}",
+        "Content-Type": "application/json",
+        "User-Agent": "xiaozhi-esp32-server",
+    }
+    # Outcrawl单次最多返回20条
+    payload = {"query": query, "limit": max(1, min(max_results, 20))}
+    logger.bind(tag=TAG).debug(f"Outcrawl搜索请求 | URL: {url} | payload: {payload}")
+    async with httpx.AsyncClient(timeout=httpx.Timeout(15.0, connect=3.0)) as client:
+        response = await client.post(url, json=payload, headers=headers)
+    data = response.json()
+    logger.bind(tag=TAG).debug(f"Outcrawl搜索响应 | status: {response.status_code}")
+
+    if response.status_code != 200:
+        # 错误信封为 {"code": "...", "message": "...", "details": {}}，
+        # code 是官方承诺的封闭枚举
+        code = data.get("code", "") if isinstance(data, dict) else ""
+        detail = data.get("message", "") if isinstance(data, dict) else ""
+        logger.bind(tag=TAG).error(
+            f"Outcrawl搜索失败 | status: {response.status_code} | code: {code} | detail: {detail}"
+        )
+        if code == "quota_exceeded":
+            # 该code覆盖两种情况：每分钟请求数限流，或月度credits用尽
+            return "联网搜索被限流或额度已用完，请稍后重试或检查Outcrawl账户余额。"
+        if code == "unauthorized":
+            return "联网搜索请求失败，请检查API Key是否正确。"
+        # 其余code（bad_request、no_capacity、search_refused、internal等）
+        # 多为输入问题或服务端暂时性故障，统一引导重试
+        return "联网搜索请求失败，请稍后重试。"
+
+    hits = data.get("hits", [])
+    if not hits:
+        return "未找到相关搜索结果。"
+
+    lines = ["【联网搜索结果】"]
+    for i, item in enumerate(hits[:max_results], 1):
+        title = item.get("title") or "无标题"
+        snippet = item.get("snippet", "")
+        lines.append(f"{i}. 标题：{title}")
+        if snippet:
+            lines.append(f"   摘要：{snippet}")
+
+    return "\n".join(lines)
+
+
 @register_function("web_search", WEB_SEARCH_FUNCTION_DESC, ToolType.SYSTEM_CTL)
 async def web_search(conn: "ConnectionHandler", query: str = None):
     logger.bind(tag=TAG).info(f"web_search 被调用 | query={query}")
@@ -175,6 +223,8 @@ async def web_search(conn: "ConnectionHandler", query: str = None):
             result_text = await _search_tavily(api_key, query, max_results)
         elif provider == "serply":
             result_text = await _search_serply(api_key, query, max_results)
+        elif provider == "outcrawl":
+            result_text = await _search_outcrawl(api_key, query, max_results)
         else:
             return ActionResponse(
                 Action.REQLLM,

main/xiaozhi-server/tests/plugins_func/test_web_search_outcrawl.py

@@ -0,0 +1,122 @@
+"""Tests for the Outcrawl source of the web_search plugin.
+
+The HTTP layer is replaced with a fake client and every request is recorded.
+Nothing here touches the network.
+"""
+import httpx
+import pytest
+
+from plugins_func.functions import web_search as web_search_module
+from plugins_func.register import Action
+
+
+class FakeResponse:
+    def __init__(self, status_code=200, payload=None):
+        self.status_code = status_code
+        self.payload = payload
+
+    def json(self):
+        return self.payload
+
+
+class FakeAsyncClient:
+    """Stands in for httpx.AsyncClient and records what was requested."""
+
+    requests = []
+    response = None
+
+    def __init__(self, *args, **kwargs):
+        pass
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc_info):
+        return False
+
+    async def post(self, url, json=None, headers=None):
+        FakeAsyncClient.requests.append({"url": url, "json": json, "headers": headers})
+        return FakeAsyncClient.response
+
+
+@pytest.fixture
+def fake_http(monkeypatch):
+    """Replace httpx.AsyncClient with the recording fake."""
+    FakeAsyncClient.requests = []
+    FakeAsyncClient.response = FakeResponse(payload={"hits": []})
+    monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)
+    return FakeAsyncClient
+
+
+class FakeConn:
+    def __init__(self, web_search_config):
+        self.config = {"plugins": {"web_search": web_search_config}}
+
+
+async def test_outcrawl_sends_bearer_key_and_limit(fake_http):
+    await web_search_module._search_outcrawl("oc-key", "上海天气", 5)
+
+    [request] = fake_http.requests
+    assert request["url"] == "https://api.outcrawl.ai/v1/search"
+    assert request["json"] == {"query": "上海天气", "limit": 5}
+    assert request["headers"]["Authorization"] == "Bearer oc-key"
+    assert request["headers"]["User-Agent"] == "xiaozhi-esp32-server"
+
+
+@pytest.mark.parametrize(("max_results", "limit"), [(0, 1), (50, 20)])
+async def test_outcrawl_clamps_limit(fake_http, max_results, limit):
+    await web_search_module._search_outcrawl("oc-key", "q", max_results)
+
+    assert fake_http.requests[0]["json"]["limit"] == limit
+
+
+async def test_outcrawl_formats_titles_and_snippets_without_urls(fake_http):
+    fake_http.response = FakeResponse(
+        payload={
+            "hits": [
+                {"url": "https://a.example/1", "title": "标题一", "snippet": "摘要一", "rank": 1},
+                {"url": "https://a.example/2", "title": None, "snippet": "", "rank": 2},
+                {"url": "https://a.example/3", "title": "标题三", "snippet": "摘要三", "rank": 3},
+            ],
+            "usage": {"credits": "1.5"},
+        }
+    )
+
+    result = await web_search_module._search_outcrawl("oc-key", "q", 2)
+
+    assert result == "【联网搜索结果】\n1. 标题：标题一\n   摘要：摘要一\n2. 标题：无标题"
+    assert "https://" not in result
+
+
+async def test_outcrawl_empty_hits(fake_http):
+    fake_http.response = FakeResponse(payload={"hits": []})
+
+    assert await web_search_module._search_outcrawl("oc-key", "q", 3) == "未找到相关搜索结果。"
+
+
+@pytest.mark.parametrize(
+    ("code", "message"),
+    [
+        ("unauthorized", "联网搜索请求失败，请检查API Key是否正确。"),
+        ("quota_exceeded", "联网搜索被限流或额度已用完，请稍后重试或检查Outcrawl账户余额。"),
+        ("bad_request", "联网搜索请求失败，请稍后重试。"),
+        ("internal", "联网搜索请求失败，请稍后重试。"),
+    ],
+)
+async def test_outcrawl_error_code(fake_http, code, message):
+    fake_http.response = FakeResponse(
+        status_code=429, payload={"code": code, "message": "denied", "details": {}}
+    )
+
+    assert await web_search_module._search_outcrawl("oc-key", "q", 3) == message
+
+
+async def test_web_search_dispatches_to_outcrawl(fake_http):
+    fake_http.response = FakeResponse(payload={"hits": [{"url": "https://a.example", "title": "T", "snippet": "S"}]})
+    conn = FakeConn({"provider": "Outcrawl", "api_key": "oc-key", "max_results": 3})
+
+    response = await web_search_module.web_search(conn, "q")
+
+    assert response.action == Action.REQLLM
+    assert response.result == "【联网搜索结果】\n1. 标题：T\n   摘要：S"
+    assert fake_http.requests[0]["json"] == {"query": "q", "limit": 3}