From fe812c916febbeffe85bc64df44a95ed192309b2 Mon Sep 17 00:00:00 2001 From: jinliyl <6469360+jinliyl@users.noreply.github.com> Date: Wed, 4 Nov 2026 19:51:14 +0800 Subject: [PATCH] feat: add DashScope web search job (#710) --- docs/en/configuration.md | 15 ++ docs/zh/configuration.md | 15 ++ example.env | 3 + pyproject.toml | 1 + reme/config/default.yaml | 20 +++ reme/steps/common/__init__.py | 2 + reme/steps/common/dashscope_search.py | 71 +++++++++ tests/unit/test_dashscope_search_step.py | 182 +++++++++++++++++++++++ 8 files changed, 309 insertions(+) create mode 100644 reme/steps/common/dashscope_search.py create mode 100644 tests/unit/test_dashscope_search_step.py diff --git a/docs/en/configuration.md b/docs/en/configuration.md index fedf7ac0..8b4a2e78 100644 --- a/docs/en/configuration.md +++ b/docs/en/configuration.md @@ -113,6 +113,21 @@ Built-in registrations include `openai`, `anthropic`, `dashscope`, `deepseek`, ` File operations, BM25 search, wikilink traversal, and `proactive_read` do not require an LLM. Evolution workflows such as `auto_memory`, `auto_resource`, `auto_dream`, and proactive refresh do. +## Web search + +With `reme-ai[core]` installed, `dashscope_search` calls `qwen3.8-flash` through the DashScope SDK, independently of LLM settings. +Set the key and start the service in one terminal: + +```bash +export DASHSCOPE_API_KEY=sk-xxx +reme start config=default +``` + +From another terminal, call `reme dashscope_search query="Latest Qwen models"`. +A missing key does not prevent service startup; calls to this Job return a failure. +The result's `search_info.search_results` preserves provider source fields; `content` is separate model-generated text. +Page bodies and snippets are not guaranteed. See `reme/config/default.yaml` for the model, search strategy, and timeout. + ## Decision Model Configuration `as_decision_model` is a provider-independent interface for probabilistic decisions. The current `agentscope` adapter diff --git a/docs/zh/configuration.md b/docs/zh/configuration.md index 6cec4614..7053a74c 100644 --- a/docs/zh/configuration.md +++ b/docs/zh/configuration.md @@ -117,6 +117,21 @@ components: 基础文件操作、BM25 检索、wikilink 遍历和 `proactive_read` 不需要 LLM。`auto_memory`、`auto_resource`、 `auto_dream` 和 proactive refresh 等演化流程需要可用 LLM。 +## 联网搜索 + +安装 `reme-ai[core]` 后，默认 `dashscope_search` Job 使用 DashScope SDK 调用 `qwen3.8-flash`，独立于上述 LLM 配置。 +在一个终端设置 Key 并启动服务： + +```bash +export DASHSCOPE_API_KEY=sk-xxx +reme start config=default +``` + +在另一个终端调用 `reme dashscope_search query="Qwen 最新模型"`。 +未配置 Key 不影响服务启动，但此 Job 调用会返回失败。 +结果的 `search_info.search_results` 保留供应商返回的来源字段，`content` 是单独的模型生成正文； +网页正文和摘要不保证提供。模型、搜索策略和超时配置见 `reme/config/default.yaml`。 + ## 决策模型配置 `as_decision_model` 是不依赖供应商的概率决策模型接口；当前的 `agentscope` 适配器通过 AgentScope `2.0.11` 调用 TypeSafe API。 diff --git a/example.env b/example.env index b8e9e3d9..fa78dd5a 100644 --- a/example.env +++ b/example.env @@ -3,6 +3,9 @@ EMBEDDING_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1 LLM_API_KEY=sk-xxx LLM_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1 +# Optional dashscope_search job (qwen3.8-flash web search). +# DASHSCOPE_API_KEY=sk-xxx + # Optional decision models: install reme-ai[core,decision]. # Setting these variables alone does not enable an as_decision_model component. # DECISION_MODEL_API_KEY=your-typesafe-api-key diff --git a/pyproject.toml b/pyproject.toml index e508f170..b00a962f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,6 +54,7 @@ web = [ ] core = [ "reme-ai[as]", + "dashscope>=1.26.0", "pillow>=13.0.0", "claude-agent-sdk>=0.2.174", "openai-codex>=0.163.0", diff --git a/reme/config/default.yaml b/reme/config/default.yaml index 32233ce4..d2d481ba 100644 --- a/reme/config/default.yaml +++ b/reme/config/default.yaml @@ -502,6 +502,26 @@ jobs: expand_links: true max_links_per_direction: 10 + dashscope_search: + backend: base + description: >- + Search the web with qwen3.8-flash; requires DASHSCOPE_API_KEY. + Returns raw search_info sources and separate model-generated content; page bodies and snippets are not guaranteed. + parameters: + type: object + properties: + query: + type: string + description: "web search query" + required: + - query + steps: + - backend: dashscope_search_step + model: qwen3.8-flash + api_key: ${DASHSCOPE_API_KEY:-} + search_strategy: turbo + timeout: 60 + node_search: backend: base description: "Digest node recall — given a candidate abstraction's name+description, surface existing digest nodes similar enough to either dedup against or link to as related." diff --git a/reme/steps/common/__init__.py b/reme/steps/common/__init__.py index 18d2c7b0..86240fe4 100644 --- a/reme/steps/common/__init__.py +++ b/reme/steps/common/__init__.py @@ -3,6 +3,7 @@ from .add import AddStep from .app_config import AppConfigStep from .chat import ChatStep +from .dashscope_search import DashScopeSearchStep from .decision_demo import DecisionDemoStep from .demo import DemoEchoStep1, DemoEchoStep2 from .health_check import HealthCheckStep @@ -19,6 +20,7 @@ __all__ = [ "AddStep", "AppConfigStep", "ChatStep", + "DashScopeSearchStep", "DecisionDemoStep", "DemoEchoStep1", "DemoEchoStep2", diff --git a/reme/steps/common/dashscope_search.py b/reme/steps/common/dashscope_search.py new file mode 100644 index 00000000..230aac83 --- /dev/null +++ b/reme/steps/common/dashscope_search.py @@ -0,0 +1,71 @@ +"""Web search through DashScope, preserving provider search results.""" + +import asyncio +import os +from contextlib import aclosing +from http import HTTPStatus + +from ..base_step import BaseStep +from ...components import R + + +@R.register("dashscope_search_step") +class DashScopeSearchStep(BaseStep): + """Return raw search sources and separate model-generated content.""" + + async def execute(self): + assert self.context is not None + response = self.context.response + query = self.context.get("query") + query = "" if query is None else str(query).strip() + api_key = (self.kwargs.get("api_key") or os.getenv("DASHSCOPE_API_KEY", "")).strip() + if not query or not api_key: + response.success = False + response.answer = "query is required" if not query else "DASHSCOPE_API_KEY is required" + return response + + from aiohttp import ClientSession + from dashscope import AioMultiModalConversation + + model = self.kwargs.get("model", "qwen3.8-flash") + search_info, sources, content, usage = {}, {}, [], {} + request_id = "" + async with asyncio.timeout(self.kwargs.get("timeout", 60)), ClientSession(trust_env=True) as session: + stream = await AioMultiModalConversation.call( + api_key=api_key, + session=session, + model=model, + messages=[{"role": "user", "content": [{"text": query}]}], + enable_search=True, + enable_thinking=False, + stream=True, + incremental_output=True, + search_options={ + "forced_search": True, + "enable_source": True, + "search_strategy": self.kwargs.get("search_strategy", "turbo"), + }, + ) + async with aclosing(stream): + async for chunk in stream: + if chunk.status_code != HTTPStatus.OK: + raise RuntimeError(f"DashScope {chunk.code}: {chunk.message} (request_id={chunk.request_id})") + request_id = chunk.request_id or request_id + if any((chunk.usage or {}).values()): + usage = dict(chunk.usage) + output = chunk.output or {} + info = output.get("search_info") or {} + search_info.update({key: value for key, value in info.items() if key != "search_results"}) + for item in info.get("search_results") or []: + sources.setdefault(item["url"], item) + for choice in output.get("choices") or []: + text = (choice.get("message") or {}).get("content") or [] + if isinstance(text, str): + content.append(text) + else: + content.extend(part["text"] for part in text if "text" in part) + + search_info["search_results"] = list(sources.values()) + response.answer = {"search_info": search_info, "content": "".join(content)} + response.metadata.update(model=model, request_id=request_id, usage=usage) + return response diff --git a/tests/unit/test_dashscope_search_step.py b/tests/unit/test_dashscope_search_step.py new file mode 100644 index 00000000..04bc0c20 --- /dev/null +++ b/tests/unit/test_dashscope_search_step.py @@ -0,0 +1,182 @@ +"""DashScope web search contracts, with no provider requests.""" + +import asyncio +from unittest.mock import AsyncMock + +import pytest +from dashscope import AioMultiModalConversation +from dashscope.api_entities.dashscope_response import DashScopeAPIResponse, MultiModalConversationResponse + +from reme.application import Application +from reme.components.job import BaseJob +from reme.config.config_parser import _load_config +from reme.steps.common.dashscope_search import DashScopeSearchStep + + +@pytest.fixture(name="provider") +def mock_provider(monkeypatch): + """Supply SDK frames and record stream cleanup.""" + frames, closed = [], [] + + async def stream(): + try: + for frame in frames: + yield frame + finally: + closed.append(True) + + call = AsyncMock(side_effect=lambda **_kwargs: stream()) + monkeypatch.setattr(AioMultiModalConversation, "call", call) + monkeypatch.setenv("DASHSCOPE_API_KEY", "test-key") + return frames, closed, call + + +def chunk(output=None, **kwargs): + """Build an SDK response, including source-only frames.""" + return MultiModalConversationResponse.from_api_response( + DashScopeAPIResponse( + status_code=200, + request_id="request-1", + output=output or {}, + **kwargs, + ), + ) + + +@pytest.mark.asyncio +async def test_default_job_starts_without_key_and_rejects_call(monkeypatch, tmp_path): + """Default config loads and the registered job starts without credentials.""" + monkeypatch.delenv("DASHSCOPE_API_KEY", raising=False) + call = AsyncMock(side_effect=AssertionError("must not contact provider")) + monkeypatch.setattr(AioMultiModalConversation, "call", call) + config = _load_config("default.yaml")["jobs"]["dashscope_search"] + app = Application( + workspace_dir=str(tmp_path), + service={"backend": "cli"}, + jobs={"dashscope_search": config}, + enable_logo=False, + log_to_file=False, + ) + await app.start() + try: + response = await app.run_job("dashscope_search", query="latest Qwen models") + assert not response.success + assert response.answer == "DASHSCOPE_API_KEY is required" + assert config["parameters"]["required"] == ["query"] + call.assert_not_awaited() + finally: + await app.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("query", [None, "", " "]) +async def test_empty_query_does_not_call_provider(provider, query): + """Reject missing or blank queries before contacting DashScope.""" + response = await DashScopeSearchStep()(query=query) + assert not response.success + assert response.answer == "query is required" + provider[2].assert_not_awaited() + + +@pytest.mark.asyncio +async def test_stream_preserves_sources_and_separates_generated_content(provider): + """Source-only, repeated, empty, and later source frames retain original fields and order.""" + frames, closed, call = provider + first = {"index": 1, "title": "One", "url": "https://example.com/1", "unknown": {"raw": True}} + second = {"index": 2, "title": "Two", "url": "https://example.com/2", "snippet": "provider snippet"} + frames.extend( + [ + chunk({"search_info": {"search_results": [first], "provider": "web"}}), + chunk( + { + "search_info": {"search_results": [dict(first, index=9), second]}, + "choices": [{"message": {"role": "assistant", "content": [{"text": "Generated "}]}}], + }, + ), + chunk( + { + "search_info": {"search_results": []}, + "choices": [{"message": {"role": "assistant", "content": "content"}}], + }, + usage={"input_tokens": 12, "output_tokens": 2, "plugins": {"search": {"count": 1}}}, + ), + chunk(), + ], + ) + response = await DashScopeSearchStep(api_key="configured-key")(query=0) + + assert response.success + assert response.answer == { + "search_info": {"search_results": [first, second], "provider": "web"}, + "content": "Generated content", + } + assert response.metadata == { + "model": "qwen3.8-flash", + "request_id": "request-1", + "usage": {"input_tokens": 12, "output_tokens": 2, "characters": 0, "plugins": {"search": {"count": 1}}}, + } + assert closed == [True] + session = call.await_args.kwargs["session"] + assert session.closed + call.assert_awaited_once_with( + api_key="configured-key", + session=session, + model="qwen3.8-flash", + messages=[{"role": "user", "content": [{"text": "0"}]}], + enable_search=True, + enable_thinking=False, + stream=True, + incremental_output=True, + search_options={"forced_search": True, "enable_source": True, "search_strategy": "turbo"}, + ) + + +@pytest.mark.asyncio +async def test_empty_results_remain_empty(provider): + """An ordinary response with no search hits does not invent sources.""" + provider[0].append(chunk({"search_info": {"search_results": []}})) + response = await DashScopeSearchStep()(query="no hits") + assert response.success + assert response.answer == {"search_info": {"search_results": []}, "content": ""} + + +@pytest.mark.asyncio +async def test_provider_error_after_sources_fails_job_and_closes_stream(provider): + """A failed stream cannot be published as a successful partial search.""" + frames, closed, _ = provider + frames.extend( + [ + chunk({"search_info": {"search_results": [{"url": "https://example.com"}]}}), + MultiModalConversationResponse( + status_code=429, + code="Throttling", + message="Try later", + request_id="failed-request", + ), + ], + ) + job = BaseJob() + job.step_specs = [(DashScopeSearchStep, {})] + response = await job(query="query") + assert not response.success + assert "Throttling: Try later" in response.answer + assert "failed-request" in response.answer + assert closed == [True] + + +@pytest.mark.asyncio +async def test_timeout_closes_stream(monkeypatch): + """Bound the entire stream, including waiting for another frame.""" + closed = [] + + async def stream(): + try: + yield chunk() + await asyncio.Event().wait() + finally: + closed.append(True) + + monkeypatch.setattr(AioMultiModalConversation, "call", AsyncMock(return_value=stream())) + with pytest.raises(TimeoutError): + await DashScopeSearchStep(api_key="test-key", timeout=0.01)(query="query") + assert closed == [True]