core/tools: search a list of sources, DeepSeek's index then Bing

``search_sources`` (a provider list, ``["deepseek", "bing"]`` by default,
``[]`` to not offer the tool at all) names what the local ``web_search`` tool
searches with, in fallback order: ``search_in_order`` asks each source in turn
and keeps the first answer, so a source that errors — a rate limit, a rejected
key — or finds nothing hands over to the next instead of failing the tool. The
handler reports the failures only when every source came up empty.

The Bing leg answers with its RSS results (``?format=rss``), which need no key
and carry a description — the snippet DeepSeek's own index does not hand out —
and it goes through the ``fetch`` client, so the SSRF and host-grant policy
applies to it too (a host that binds tool instance state, as the CLI does).
This commit is contained in:
2026-10-07 00:09:46 +08:00
parent 19ab86f590
commit 2bb6e06171
9 changed files with 348 additions and 56 deletions
+14 -1
View File
@@ -21,10 +21,15 @@ def _effective(
convention: str = "",
url: str = "",
tools: list[dict[str, object]] | None = None,
sources: list[str] | None = None,
) -> EffectiveProvider:
provider = DeepseekProvider(access_key_or_token="sk-test", convention=convention, url=url)
effective = resolve_effective_provider(provider, model="deepseek-flash")
return effective if tools is None else replace(effective, provider_tools=tools)
if tools is not None:
effective = replace(effective, provider_tools=tools)
if sources is not None:
effective = replace(effective, search_sources=sources)
return effective
def test_a_deepseek_provider_asks_for_a_search_tool_like_openai_does() -> None:
@@ -45,6 +50,14 @@ def test_without_provider_tools_there_is_no_search() -> None:
assert local_web_search(_effective(convention="responses", tools=[])) is False
def test_without_search_sources_there_is_no_local_tool() -> None:
# ``search_sources = []`` leaves the tool out even though provider_tools ask
# for a search (the hosted path on the Anthropic convention is unaffected).
assert _effective(convention="responses", sources=[]).search_sources == []
assert local_web_search(_effective(convention="responses", sources=[])) is False
assert _effective(convention="responses").search_sources == ["deepseek", "bing"]
def test_search_base_url_prefers_an_anthropic_endpoint() -> None:
assert deepseek_search_base_url(_effective(convention="responses")) == "https://api.deepseek.com/anthropic"
assert (
+45
View File
@@ -0,0 +1,45 @@
"""Bing's RSS answers as a search source (parsing only — no network)."""
from __future__ import annotations
import pytest
from plyngent.lmproto.deepseek.anthropic.search import SearchHit
from plyngent.tools.net.bing import parse_bing_rss
DOCUMENT = """<?xml version="1.0" encoding="utf-8" ?>
<rss version="2.0"><channel>
<title>必应:DeepSeek Harness</title>
<item>
<title>DeepSeek Harness v0.2</title>
<link>https://pandaily.com/deepseek-harness-v0-2</link>
<description>Desktop installers and a plugin manager.</description>
<pubDate>Sat, 03 Oct 2026 12:06:00 GMT</pubDate>
</item>
<item>
<title>No link here</title>
<description>Dropped: a hit without a URL is nothing to read.</description>
</item>
<item>
<title> Spaced title </title>
<link> https://example.com/spaced </link>
</item>
</channel></rss>
"""
def test_parse_bing_rss_maps_items_to_hits() -> None:
hits = parse_bing_rss(DOCUMENT)
assert hits == [
SearchHit(
title="DeepSeek Harness v0.2",
url="https://pandaily.com/deepseek-harness-v0-2",
snippet="Desktop installers and a plugin manager.",
),
SearchHit(title="Spaced title", url="https://example.com/spaced", snippet=""),
]
def test_parse_bing_rss_rejects_a_non_rss_body() -> None:
with pytest.raises(RuntimeError, match="unparseable XML"):
_ = parse_bing_rss("<!doctype html><html><body>rate limited</body></html>")
+81 -22
View File
@@ -8,28 +8,34 @@ import pytest
from plyngent.agent.tools import ToolTag
from plyngent.lmproto.deepseek.anthropic.search import SearchHit
from plyngent.tools.net.search import WEB_SEARCH_TOOL_NAME, build_web_search_tool
from plyngent.tools.net.search import (
WEB_SEARCH_TOOL_NAME,
build_web_search_tool,
search_in_order,
)
class _StubSearch:
"""Stands in for :class:`DeepseekSearchClient`."""
class _StubBackend:
"""Stands in for a search source (raises, finds nothing, or answers)."""
def __init__(self, hits: list[SearchHit]) -> None:
self.hits = hits
self.queries: list[tuple[str, str]] = []
def __init__(self, *, hits: list[SearchHit] | None = None, error: str | None = None) -> None:
self.hits = hits or []
self.error = error
self.queries: list[str] = []
async def search(self, query: str, *, model: str, **kwargs: object) -> list[SearchHit]:
self.queries.append((query, model))
async def search(self, query: str) -> list[SearchHit]:
self.queries.append(query)
if self.error is not None:
raise RuntimeError(self.error)
return self.hits
def _definition(hits: list[SearchHit]) -> tuple[Any, _StubSearch]:
client = _StubSearch(hits)
return build_web_search_tool(client, model="deepseek-flash"), client
def _hits(*pairs: tuple[str, str]) -> list[SearchHit]:
return [SearchHit(title=title, url=url) for title, url in pairs]
def test_tool_definition_shape() -> None:
definition, _ = _definition([])
definition = build_web_search_tool([_StubBackend()])
assert definition.name == WEB_SEARCH_TOOL_NAME
assert definition.tags == ToolTag.LOCAL | ToolTag.READ_ONLY
assert "fetch" in definition.description
@@ -39,20 +45,73 @@ def test_tool_definition_shape() -> None:
@pytest.mark.asyncio
async def test_handler_returns_one_line_per_hit() -> None:
definition, client = _definition([SearchHit(title="DeepSeek-V4 preview", url="https://api-docs.deepseek.com/x")])
assert await definition.handler(query="DeepSeek-V4") == ("- DeepSeek-V4 preview — https://api-docs.deepseek.com/x")
assert client.queries == [("DeepSeek-V4", "deepseek-flash")]
async def test_handler_returns_one_entry_per_hit() -> None:
backend = _StubBackend(hits=_hits(("DeepSeek-V4 preview", "https://api-docs.deepseek.com/x")))
definition = build_web_search_tool([backend])
assert await definition.handler(query="DeepSeek-V4") == "- DeepSeek-V4 preview — https://api-docs.deepseek.com/x"
assert backend.queries == ["DeepSeek-V4"]
@pytest.mark.asyncio
async def test_handler_prints_the_snippet_under_its_hit() -> None:
backend = _StubBackend(hits=[SearchHit(title="T", url="https://example.com/1", snippet="One line of it.")])
definition = build_web_search_tool([backend])
assert await definition.handler(query="q") == "- T — https://example.com/1\n One line of it."
@pytest.mark.asyncio
async def test_the_first_source_that_answers_wins() -> None:
failing = _StubBackend(error="HTTP 503")
empty = _StubBackend()
answering = _StubBackend(hits=_hits(("Found", "https://example.com/found")))
hits, failures = await search_in_order([failing, empty, answering], "q")
assert [hit.url for hit in hits] == ["https://example.com/found"]
assert failures == ["_StubBackend: HTTP 503", "_StubBackend: no results"]
@pytest.mark.asyncio
async def test_handler_caps_the_hits_and_names_an_untitled_page() -> None:
definition, _ = _definition([SearchHit(title="", url=f"https://example.com/{index}") for index in range(5)])
lines = (await definition.handler(query="anything", hits=2)).splitlines()
assert lines == ["- (untitled) — https://example.com/0", "- (untitled) — https://example.com/1"]
backend = _StubBackend(hits=_hits(*[("", f"https://example.com/{index}") for index in range(5)]))
definition = build_web_search_tool([backend], max_hits=2)
assert (await definition.handler(query="anything")).splitlines() == [
"- (untitled) — https://example.com/0",
"- (untitled) — https://example.com/1",
]
@pytest.mark.asyncio
async def test_handler_reports_an_empty_search() -> None:
definition, _ = _definition([])
assert "no results" in await definition.handler(query="nothing matches this")
async def test_handler_reports_every_source_that_failed() -> None:
definition = build_web_search_tool([_StubBackend(error="timeout"), _StubBackend()])
message = await definition.handler(query="nothing matches this")
assert "found nothing" in message
assert "timeout" in message and "no results" in message
@pytest.mark.asyncio
async def test_handler_without_sources_says_so() -> None:
definition = build_web_search_tool([])
assert "no source configured" in await definition.handler(query="q")
def test_deepseek_backend_passes_the_model_through() -> None:
from plyngent.tools.net.search import DeepseekSearchBackend
class _Client:
def __init__(self) -> None:
self.seen: list[tuple[str, str]] = []
async def search(self, query: str, *, model: str, **kwargs: object) -> list[SearchHit]:
self.seen.append((query, model))
return _hits(("T", "https://example.com"))
client: Any = _Client()
backend = DeepseekSearchBackend(client, model="deepseek-flash")
async def run() -> list[SearchHit]:
return list(await backend.search("q"))
import asyncio
hits = asyncio.run(run())
assert [hit.url for hit in hits] == ["https://example.com"]
assert client.seen == [("q", "deepseek-flash")]