diff --git a/src/plyngent/lmproto/deepseek/anthropic/search.py b/src/plyngent/lmproto/deepseek/anthropic/search.py new file mode 100644 index 0000000..c6da305 --- /dev/null +++ b/src/plyngent/lmproto/deepseek/anthropic/search.py @@ -0,0 +1,142 @@ +"""Server-side web search on DeepSeek's Anthropic-compatible endpoint. + +DeepSeek runs ``web_search`` *inside* a turn: the request carries the server +tool, and the response comes back with a ``server_tool_use`` block per query +plus the ``web_search_tool_result`` that query produced — nothing is executed +here. Results only carry a title, a URL and an ``encrypted_content`` the index +keeps the page body in, which is why the hit list is all a caller can pass on. + +Only this surface searches: the Responses and chat-completions surfaces ignore a +hosted search tool (a live probe answers "I can't browse the web" whether the +tool arrives as ``{"type": "web_search"}`` or in the server-tool shape), so a +DeepSeek provider that talks them searches through +:func:`~plyngent.tools.net.search.build_web_search_tool` instead. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, cast + +import msgspec +import niquests + +if TYPE_CHECKING: + from plyngent.lmproto.anthropic.config import AnthropicConfig + +# A search turn stops as soon as the search came back: the answer that follows is +# thrown away, so the budget only has to cover the thinking that picks the query +# and the results. A live probe with 64 tokens spent all of it before searching +# and returned no results at all. +DEFAULT_SEARCH_MAX_TOKENS = 512 +DEFAULT_SEARCH_MAX_USES = 3 +_HTTP_ERROR = 400 +_SEARCH_TOOL: dict[str, Any] = {"type": "web_search_20250305", "name": "web_search"} + + +class SearchHit(msgspec.Struct, frozen=True, omit_defaults=True): + """One page the search index matched.""" + + title: str = "" + url: str = "" + + +def hits_from_content(content: object) -> list[SearchHit]: + """Collect the search hits from a response's content blocks (deduplicated).""" + if not isinstance(content, list): + return [] + hits: list[SearchHit] = [] + seen: set[str] = set() + for block_obj in cast("list[object]", content): + if not isinstance(block_obj, dict): + continue + block = cast("dict[str, object]", block_obj) + if block.get("type") != "web_search_tool_result": + continue + entries = block.get("content") + if not isinstance(entries, list): + continue + for entry_obj in cast("list[object]", entries): + if not isinstance(entry_obj, dict): + continue + entry = cast("dict[str, object]", entry_obj) + if entry.get("type") != "web_search_result": + continue + url = entry.get("url") + if not isinstance(url, str) or not url or url in seen: + continue + seen.add(url) + title = entry.get("title") + hits.append(SearchHit(title=title if isinstance(title, str) else "", url=url)) + return hits + + +class DeepseekSearchClient: + """``POST /messages`` with a hosted ``web_search`` tool, read for its hits.""" + + session: niquests.AsyncSession + encoder: msgspec.json.Encoder + _api_key: str + _api_version: str + + def __init__(self, config: AnthropicConfig) -> None: + self.session = niquests.AsyncSession(base_url=config.base_url, timeout=config.timeout) + self.encoder = msgspec.json.Encoder() + self._api_key = config.api_key + self._api_version = config.anthropic_version + + def _headers(self) -> dict[str, str]: + return { + "Content-Type": "application/json", + "x-api-key": self._api_key, + "anthropic-version": self._api_version, + } + + async def search( + self, + query: str, + *, + model: str, + max_uses: int = DEFAULT_SEARCH_MAX_USES, + max_tokens: int = DEFAULT_SEARCH_MAX_TOKENS, + ) -> list[SearchHit]: + """Run *query* through the server's search; return the pages it matched.""" + payload: dict[str, Any] = { + "model": model, + "max_tokens": max_tokens, + "messages": [{"role": "user", "content": query}], + "tools": [{**_SEARCH_TOOL, "max_uses": max_uses}], + } + resp = await self.session.post( + "/messages", + data=self.encoder.encode(payload), + headers=self._headers(), + stream=False, + ) + body = await self._read_body(resp) + parsed = msgspec.json.decode(body, type=dict[str, object]) + return hits_from_content(parsed.get("content")) + + async def _read_body(self, resp: object) -> bytes: + from plyngent.lmproto.openai_compatible.client import ( + http_error_message, + read_response_body, + ) + + body = await read_response_body(resp) + status = int(getattr(resp, "status_code", 0) or 0) + if status >= _HTTP_ERROR: + detail = http_error_message(status, body, what="search") or f"search HTTP {status}" + raise RuntimeError(detail) + if body is None: + msg = "search response body is empty" + raise RuntimeError(msg) + return body.encode() if isinstance(body, str) else bytes(body) + + +__all__ = [ + "DEFAULT_SEARCH_MAX_TOKENS", + "DEFAULT_SEARCH_MAX_USES", + "DeepseekSearchClient", + "SearchHit", + "hits_from_content", +] diff --git a/tests/test_lmproto/test_deepseek_anthropic_search.py b/tests/test_lmproto/test_deepseek_anthropic_search.py new file mode 100644 index 0000000..bc170a2 --- /dev/null +++ b/tests/test_lmproto/test_deepseek_anthropic_search.py @@ -0,0 +1,115 @@ +"""DeepSeek server-side search backend (Anthropic-compatible endpoint).""" + +from __future__ import annotations + +import msgspec +import pytest + +from plyngent.lmproto.anthropic.config import AnthropicConfig +from plyngent.lmproto.deepseek.anthropic.search import ( + DeepseekSearchClient, + SearchHit, + hits_from_content, +) + + +def test_hits_from_content_dedupes_and_skips_noise() -> None: + content = [ + {"type": "thinking", "thinking": "search"}, + { + "type": "web_search_tool_result", + "tool_use_id": "call_1", + "content": [ + {"type": "web_search_result", "title": "First", "url": "https://a.example/x"}, + {"type": "web_search_result", "title": "First again", "url": "https://a.example/x"}, + {"type": "web_search_result", "url": "https://b.example/y"}, + {"type": "web_search_result", "title": "no url"}, + ], + }, + { + "type": "web_search_tool_result", + "tool_use_id": "call_2", + "content": [{"type": "web_search_tool_result_error", "error_code": "max_uses_exceeded"}], + }, + ] + hits = hits_from_content(content) + assert hits == [ + SearchHit(title="First", url="https://a.example/x"), + SearchHit(title="", url="https://b.example/y"), + ] + assert hits_from_content("not a list") == [] + + +class _Resp: + def __init__(self, status_code: int, body: bytes) -> None: + self.status_code = status_code + self.content = body + + +def _search_body() -> bytes: + return msgspec.json.encode( + { + "id": "msg_1", + "content": [ + { + "type": "server_tool_use", + "id": "call_00_1", + "name": "web_search", + "input": {"query": "DeepSeek-V4"}, + }, + { + "type": "web_search_tool_result", + "tool_use_id": "call_00_1", + "content": [ + { + "type": "web_search_result", + "title": "DeepSeek-V4 preview", + "url": "https://api-docs.deepseek.com/news/news260424/", + "encrypted_content": "opaque", + } + ], + }, + ], + "usage": {"input_tokens": 9932, "output_tokens": 377}, + } + ) + + +@pytest.mark.asyncio +async def test_search_sends_the_hosted_tool_and_reads_its_hits(monkeypatch: pytest.MonkeyPatch) -> None: + client = DeepseekSearchClient(AnthropicConfig(api_key="sk-test", base_url="https://api.deepseek.com/anthropic")) + sent: dict[str, object] = {} + + async def fake_post(path: str, **kwargs: object) -> _Resp: + sent["path"] = path + sent.update(kwargs) + return _Resp(200, _search_body()) + + monkeypatch.setattr(client.session, "post", fake_post) + hits = await client.search("DeepSeek-V4", model="deepseek-flash") + + assert hits == [SearchHit(title="DeepSeek-V4 preview", url="https://api-docs.deepseek.com/news/news260424/")] + assert sent["path"] == "/messages" + assert sent["stream"] is False + data = sent["data"] + assert isinstance(data, bytes) + payload = msgspec.json.decode(data, type=dict[str, object]) + assert payload["tools"] == [{"type": "web_search_20250305", "name": "web_search", "max_uses": 3}] + # A search turn stops as soon as it has searched; the answer is not needed. + assert payload["max_tokens"] == 512 + assert payload["messages"] == [{"role": "user", "content": "DeepSeek-V4"}] + headers = sent["headers"] + assert isinstance(headers, dict) + assert headers["x-api-key"] == "sk-test" + + +@pytest.mark.asyncio +async def test_search_raises_with_the_endpoint_detail(monkeypatch: pytest.MonkeyPatch) -> None: + client = DeepseekSearchClient(AnthropicConfig(api_key="sk-test", base_url="https://api.deepseek.com/anthropic")) + + async def fake_post(path: str, **kwargs: object) -> _Resp: + return _Resp(422, b'{"error": {"message": "tools[0]: unknown variant"}}') + + monkeypatch.setattr(client.session, "post", fake_post) + with pytest.raises(RuntimeError, match="unknown variant"): + _ = await client.search("DeepSeek-V4", model="deepseek-flash")