mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 07:17:54 +08:00
lmproto: search through DeepSeek's Anthropic-compatible endpoint
DeepSeek runs ``web_search`` inside a turn — the request declares the server tool, the response carries a ``server_tool_use`` per query and the ``web_search_tool_result`` it produced — so a search is one ``/messages`` call whose hits are read out of the content blocks. The turn stops as soon as it has searched (a live probe with 64 output tokens spent all of them before the search and returned nothing, 512 answers with the hits). Hits carry a title, a URL and an ``encrypted_content`` only the index can read, so the hit list is all a caller gets: the page bodies stay server-side.
This commit is contained in:
@@ -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",
|
||||
]
|
||||
@@ -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")
|
||||
Reference in New Issue
Block a user