feat: add DashScope web search job (#710)

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

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。

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

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",

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."

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",

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

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]