feat(classifier): add optional Jev component and service demo (#619)

* feat(classifier): add optional Jev component and CLI demo

* fix(classifier): expose Jev demo through inherited default service
This commit is contained in:
jinliyl 2026-10-14 17:49:49 +08:00 • committed by GitHub
parent c8a03dbd5a
commit 43ef665595
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 1075 additions and 1 deletions

View file

@ -38,7 +38,7 @@ jobs:
- name: Install dependencies
run: |
python -m pip install --upgrade pip setuptools wheel
pip install -e reme_studio -e ".[dev,core]"
pip install -e reme_studio -e ".[dev,core,classifier-jev]"
pip install --no-deps -e plugins/auto-fin
pip install -e plugins/daily_paper
pip install -e plugins/lme -e plugins/beam

View file

@ -122,6 +122,9 @@ cd ..
The static build requires Node.js 22.13 or newer and makes Studio available from the source tree.
Optional Jev classification uses `pip install "reme-ai[core,classifier-jev]"`. Configure a TypeSafe API key and run the
service demo as described in [Classifier configuration](https://reme.agentscope.io/en/configuration#classifier-configuration).
### Docker
With Docker and Compose 2.24.0+, build and start ReMe with the bundled Studio:

View file

@ -113,6 +113,9 @@ cd ..
静态构建要求 Node.js 22.13 或更高版本,并让源码安装可以直接使用 Studio。
可选 Jev 分类功能使用 `pip install "reme-ai[core,classifier-jev]"` 安装。配置 TypeSafe API key 后即可启动服务并调用 `reme jev`,
详见[Classifier 配置](https://reme.agentscope.io/zh/configuration#classifier-配置)。
### Docker
使用 Docker 和 Compose 2.24.0+,构建并启动包含 Studio 的 ReMe:

View file

@ -109,6 +109,90 @@ 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.
## Classifier Configuration
`as_classifier` is an independent probabilistic classifier component. Its `jev` backend calls the TypeSafe API through
AgentScope `2.0.10`. It evaluates named binary, choice, or score questions against shared text or a JSON object.
Install the optional dependencies:
```bash
pip install "reme-ai[core,classifier-jev]"
# For a source checkout:
pip install -e ".[core,classifier-jev]"
```
`core` excludes the Jev SDK; `full` includes this extra. Environment variables alone do not enable classification, and
the default classifier configuration remains commented out. Startup does not import optional classifier modules or
create a client. The first classification checks dependencies and credentials; shutdown waits for active requests before
closing the shared client.
| Environment variable | Default | Purpose |
|---|---|---|
| `CLASSIFIER_BACKEND` | `jev` | Classifier backend |
| `CLASSIFIER_MODEL_NAME` | `jev-latest` | Model name or alias |
| `CLASSIFIER_API_KEY` | Empty | TypeSafe API key, required when classifying |
| `CLASSIFIER_BASE_URL` | `null` | Optional API URL; empty values use the SDK default |
To enable a component:
```yaml
components:
as_classifier:
default:
backend: ${CLASSIFIER_BACKEND:-jev}
model: ${CLASSIFIER_MODEL_NAME:-jev-latest}
credential:
api_key: ${CLASSIFIER_API_KEY:-}
base_url: ${CLASSIFIER_BASE_URL:-null}
timeout: 30
max_retries: 2
retry_delay: 0.5
```
`timeout` is the per-request timeout in seconds; `max_retries` counts retries after the initial request; `retry_delay`
is the initial backoff in seconds. Configure these fields through YAML or CLI overrides. The SDK performs retries.
The `classifier_demo` configuration uses `extends: default` to inherit the HTTP/MCP service, components, and Jobs,
then adds a classifier and a service-exposed `jev` Job. The default background tasks also start. First start the service
in one terminal:
```bash
export CLASSIFIER_API_KEY=your-typesafe-api-key
reme start config=classifier_demo \
components.as_classifier.default.timeout=15
```
Then call it from another terminal; the client discovers the running service configuration:
```bash
reme jev state="My account was charged twice. Please help urgently."
```
Calls send the state and questions to TypeSafe. Omitting `questions` evaluates urgency, customer-support routing, and
priority together. You can replace the demo questions completely and pass a JSON object as state:
```bash
reme jev \
state='{"request":"My account was charged twice","channel":"email"}' \
questions='{"urgent":{"type":"binary","instructions":"Does this need urgent attention?"}}'
```
Question types are `binary`, `choice`, and `score`. Binary questions accept optional `criteria.true` / `criteria.false`
descriptions. Choice questions require a non-empty `criteria` object mapping category names to descriptions or `null`.
Score questions require a non-empty ordered list of descriptions, starting at score zero. Each question can include
`instructions`. Explicit empty questions fail validation instead of selecting the demo defaults.
`reme jev` displays the standard Response JSON, with the classifier result under `answer`, including `model`, `content`,
`usage`, response identity, and timing information. Answers preserve probabilities and
confidence and are keyed by question name; integer score-distribution keys become strings in JSON. The Step also writes
`context.data["classifier_result"]` for subsequent Steps in that request, without automatically writing memory files.
SDK errors follow BaseJob's `success=false` contract.
Use `classifier_demo_step` in your own Job and select a named component with the Step setting `as_classifier: <name>`.
See `reme/components/as_classifier/` for the component interface. AgentScope classifier schemas define question and
answer contracts.
## Embeddings
Vector retrieval is disabled by default. Credentials alone do not enable it: configure `as_embedding`, `embedding_store`, and connect the store to `file_store`.

View file

@ -73,6 +73,7 @@ reme/
file_chunker/ # Markdown / JSON / JSONL / generic text chunking
file_catalog/ # change checkpoints
as_llm/, as_embedding/ # model wrappers
as_classifier/ # optional Jev probabilistic classifier wrapper
agent_wrapper/ # AgentScope / Claude Code / Codex wrappers
steps/
base_step.py # BaseStep, Ref, dispatch_steps
@ -537,6 +538,7 @@ Current default components in `reme/config/default.yaml`:
| `service` | singleton | `http` | Default HTTP service. |
| `tokenizer` | `default` | `regex` | BM25 tokenizer. |
| `as_embedding` | `default` | Not configured by default; example uses `openai` | Provides the embedding model wrapper after uncommenting the example config. |
| `as_classifier` | `default` | Not configured by default; example uses `jev` | Optional probabilistic classifier; demonstrated by the classifier_demo config extending default. |
| `embedding_store` | `default` | Not configured by default; example uses `local` | Depends on `as_embedding: default` after uncommenting the example config. |
| `as_llm` | `default` | `${LLM_BACKEND:-openai}` | LLM model wrapper. |
| `agent_wrapper` | `default` | `agentscope` | AgentScope wrapper. |

View file

@ -113,6 +113,84 @@ components:
基础文件操作、BM25 检索、wikilink 遍历和 `proactive_read` 不需要 LLM。`auto_memory`、`auto_resource`、
`auto_dream` 和 proactive refresh 等演化流程需要可用 LLM。
## Classifier 配置
`as_classifier` 是独立的概率分类组件,当前提供 `jev` backend,通过 AgentScope `2.0.10` 调用 TypeSafe API。
它接受共享的文本或 JSON object 与命名问题,返回二分类概率、选项分类或评分结果。
安装可选依赖:
```bash
pip install "reme-ai[core,classifier-jev]"
# 从源码安装时:
pip install -e ".[core,classifier-jev]"
```
`core` 不包含 Jev SDK,`full` 包含该 extra。只设置环境变量不会自动启用组件;默认配置中的 classifier 示例保持注释。
组件启动不导入可选分类模块、不创建客户端,首次分类调用才检查依赖和凭证。客户端由组件复用,关闭时先等待正在执行的请求结束,再释放客户端。
| 环境变量 | 默认值 | 作用 |
|---|---|---|
| `CLASSIFIER_BACKEND` | `jev` | Classifier backend |
| `CLASSIFIER_MODEL_NAME` | `jev-latest` | 模型名或别名 |
| `CLASSIFIER_API_KEY` | 空 | TypeSafe API key,分类调用时必须提供 |
| `CLASSIFIER_BASE_URL` | `null` | 可选 API 地址;空值使用 SDK 默认地址 |
启用组件的配置示例:
```yaml
components:
as_classifier:
default:
backend: ${CLASSIFIER_BACKEND:-jev}
model: ${CLASSIFIER_MODEL_NAME:-jev-latest}
credential:
api_key: ${CLASSIFIER_API_KEY:-}
base_url: ${CLASSIFIER_BASE_URL:-null}
timeout: 30
max_retries: 2
retry_delay: 0.5
```
`timeout` 是单次请求的超时秒数;`max_retries` 是首次请求之外的重试次数;`retry_delay` 是初始退避秒数。
这三个字段通过 YAML 或 CLI 覆盖配置,重试由 SDK 执行。
内置 `classifier_demo` 配置通过 `extends: default` 继承默认 HTTP/MCP 服务、组件和 Job,并增加 classifier 与
可通过服务调用的 `jev` Job。默认后台任务也会启动。先在一个终端启动服务:
```bash
export CLASSIFIER_API_KEY=your-typesafe-api-key
reme start config=classifier_demo \
components.as_classifier.default.timeout=15
```
再在另一个终端调用,客户端会发现正在运行的服务配置:
```bash
reme jev state="我的账户被重复扣款,请尽快处理"
```
调用会把 `state` 和问题发送给 TypeSafe。省略 `questions` 时,demo 同时判断是否紧急、选择客服类别和评估优先级。
可以完整替换演示问题,也可以传入 JSON object:
```bash
reme jev \
state='{"request":"我的账户被重复扣款","channel":"email"}' \
questions='{"urgent":{"type":"binary","instructions":"是否需要紧急处理?"}}'
```
问题的 `type` 支持 `binary`、`choice`、`score`。`binary` 可提供 `criteria.true` / `criteria.false` 描述;
`choice` 需要非空 `criteria` object,key 是类别,value 是描述或 `null`;`score` 需要非空的有序描述列表,
分数从零开始。问题可提供 `instructions`。显式传入空的 `questions` 会报错,不会退回 demo 默认问题。
`reme jev` 显示标准 Response JSON,分类结果位于 `answer`,包含 `model`、`content`、
`usage`、响应标识和时间等信息。`content` 按问题名组织,保留概率与置信度;评分分布的整数 key 在 JSON 中变为字符串。
Step 同时写入请求内的 `context.data["classifier_result"]`,可供后续 Step 消费,不自动写入记忆文件。
SDK 错误沿用 BaseJob 的 `success=false` 契约。
在自己的 Job 中使用 `classifier_demo_step`,通过 Step 配置 `as_classifier: <组件名>` 选择命名组件。
组件接口见 `reme/components/as_classifier/`;问题与答案契约由 AgentScope classifier schema 定义。
## Embedding 配置
向量检索默认关闭。只设置 `EMBEDDING_API_KEY` 不会自动启用它;还需要同时启用 `as_embedding`、`embedding_store`,并把它连接到 `file_store`:

View file

@ -68,6 +68,7 @@ reme/
file_chunker/ # Markdown / JSON / JSONL / 通用文本分块
file_catalog/ # 变更 checkpoint
as_llm/, as_embedding/ # 模型封装
as_classifier/ # 可选的 Jev 概率分类封装
agent_wrapper/ # AgentScope / Claude Code / Codex wrapper
steps/
base_step.py # BaseStep、Ref、dispatch_steps
@ -520,6 +521,7 @@ flowchart LR
| `as_embedding` | `default` | 默认未配置;示例为 `openai` | 取消配置注释后提供 embedding 模型封装 |
| `embedding_store` | `default` | 默认未配置;示例为 `local` | 取消配置注释后依赖 `as_embedding: default` |
| `as_llm` | `default` | `${LLM_BACKEND:-openai}` | LLM 模型封装 |
| `as_classifier` | `default` | 默认未配置;示例为 `jev` | 可选概率分类;通过继承 default 的 classifier_demo 配置演示 |
| `agent_wrapper` | `default` | `agentscope` | AgentScope wrapper |
| `agent_wrapper` | `claude_code` | `claude_code` | Claude Code wrapper |
| `agent_wrapper` | `codex/codex_oauth` | `codex` | 分别使用 API key 与 OAuth 的 Codex wrapper |

View file

@ -3,6 +3,13 @@ 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 Jev classification: install reme-ai[core,classifier-jev].
# Setting these variables alone does not enable a classifier component.
# CLASSIFIER_API_KEY=your-typesafe-api-key
# CLASSIFIER_BACKEND=jev
# CLASSIFIER_MODEL_NAME=jev-latest
# CLASSIFIER_BASE_URL=null
# Optional data-source mirrors. Unset variables use the official services.
# HF_MIRROR_URL=https://hf-mirror.com
# ARXIV_MIRROR_URL=https://export.arxiv.org

View file

@ -44,6 +44,9 @@ dependencies = [
as = [
"agentscope[model-ollama]==2.0.10",
]
classifier-jev = [
"agentscope[classifier-jev]==2.0.10",
]
web = [
"reme_studio",
]
@ -74,6 +77,7 @@ dev = [
]
full = [
"reme-ai[core]",
"reme-ai[classifier-jev]",
"reme-ai[dev]",
"reme-ai[image-heif]",
]

View file

@ -2,6 +2,7 @@
from . import agent_wrapper
from . import as_llm
from . import as_classifier
from . import client
from . import as_embedding
from . import embedding_store
@ -33,6 +34,7 @@ __all__ = [
# base components
"agent_wrapper",
"as_llm",
"as_classifier",
"client",
"as_embedding",
"embedding_store",

View file

@ -0,0 +1,6 @@
"""Optional probabilistic classifier components."""
from .base_as_classifier import BaseAsClassifier
from .jev import JevAsClassifier
__all__ = ["BaseAsClassifier", "JevAsClassifier"]

View file

@ -0,0 +1,25 @@
"""Lifecycle-managed wrappers for AgentScope classifier models."""
from abc import abstractmethod
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any
from ..base_component import BaseComponent
from ...enumeration import ComponentEnum
if TYPE_CHECKING:
from agentscope.classifier import ClassifierResponse
class BaseAsClassifier(BaseComponent):
"""Classify shared text or JSON state with named probabilistic questions."""
component_type = ComponentEnum.AS_CLASSIFIER
@abstractmethod
async def __call__(
self,
state: str | dict,
questions: Mapping[str, Any],
) -> "ClassifierResponse":
"""Return typed answers; concrete components own provider resources."""

View file

@ -0,0 +1,121 @@
"""Optional TypeSafe Jev classifier, initialized only when called."""
import asyncio
from collections.abc import Mapping
from importlib import import_module
import json
from typing import TYPE_CHECKING, Any
from pydantic import BaseModel, ConfigDict, Field, JsonValue, TypeAdapter
from .base_as_classifier import BaseAsClassifier
from ..component_registry import R
if TYPE_CHECKING:
from agentscope.classifier import ClassifierResponse, JevClassifierModel
class _JevOptions(BaseModel):
"""Validate transport options without importing optional provider modules."""
model_config = ConfigDict(extra="forbid")
model: str = Field(default="jev-latest", min_length=1)
timeout: float = Field(default=30.0, gt=0, allow_inf_nan=False)
max_retries: int = Field(default=2, ge=0)
retry_delay: float = Field(default=0.5, ge=0, allow_inf_nan=False)
@R.register("jev")
class JevAsClassifier(BaseAsClassifier):
"""Wrap AgentScope's Jev API without requiring its extra during startup.
``questions`` accepts AgentScope question objects or JSON dictionaries with
a ``type`` discriminator (``binary``, ``choice``, or ``score``). Only this
component's configuration supplies credentials and transport options.
"""
def __init__(
self,
credential: dict | None = None,
model: str = "jev-latest",
timeout: float = 30.0,
max_retries: int = 2,
retry_delay: float = 0.5,
**kwargs,
) -> None:
super().__init__(**kwargs)
self.credential = dict(credential or {})
self.options = _JevOptions(model=model, timeout=timeout, max_retries=max_retries, retry_delay=retry_delay)
self._model: "JevClassifierModel | None" = None
self._active_calls = 0
self._idle = asyncio.Event()
self._idle.set()
@staticmethod
def _load_classifier():
"""Delay the optional AgentScope classifier import until classification."""
try:
module = import_module("agentscope.classifier")
# Check exports as well as module presence for older installations.
return module.ClassifierQuestion, module.JevClassifierModel
except (ImportError, AttributeError) as exc:
raise ImportError(
"Jev classification requires compatible AgentScope classifier support. "
"Install it with `pip install 'reme-ai[classifier-jev]'`.",
) from exc
async def __call__(self, state: str | dict, questions: Mapping[str, Any]) -> "ClassifierResponse":
"""Validate the request before constructing or calling the provider."""
if not isinstance(state, (str, dict)) or (isinstance(state, str) and not state.strip()):
raise ValueError("Classifier state must be non-empty text or a JSON object.")
if isinstance(state, dict):
state = TypeAdapter(dict[str, JsonValue]).validate_python(state, strict=True)
json.dumps(state, allow_nan=False)
if not isinstance(questions, Mapping) or not questions:
raise ValueError("Classifier questions must be a non-empty mapping.")
if any(not isinstance(name, str) or not name.strip() for name in questions):
raise ValueError("Classifier question names must be non-empty strings.")
question_type, model_cls = self._load_classifier()
adapter = TypeAdapter(question_type)
parsed = {name: adapter.validate_python(question) for name, question in questions.items()}
await self.start()
async with self._lock:
if not self.is_started:
raise RuntimeError("Classifier component closed before the request could start.")
if self._model is None:
credential = dict(self.credential)
api_key = credential.get("api_key")
if not isinstance(api_key, str) or not api_key.strip():
raise ValueError("Jev classification requires CLASSIFIER_API_KEY or credential.api_key.")
if not credential.get("base_url"):
credential["base_url"] = None
try:
credential_cls = import_module("agentscope.credential").TypeSafeCredential
self._model = model_cls(credential=credential_cls(**credential), **self.options.model_dump())
except ImportError as exc:
raise ImportError(
"Jev classification requires a compatible typesafe-sdk. "
"Install it with `pip install 'reme-ai[classifier-jev]'`.",
) from exc
model = self._model
self._active_calls += 1
self._idle.clear()
try:
return await model(state=state, questions=parsed)
finally:
# No await here: finishing requests must unblock close(), which owns
# the lifecycle lock while draining active requests.
self._active_calls -= 1
if self._active_calls == 0:
self._idle.set()
async def _close(self) -> None:
"""Release the owned TypeSafe client and allow a later fresh start."""
try:
await self._idle.wait()
finally:
model, self._model = self._model, None
if model is not None:
await model.client.aclose()

View file

@ -0,0 +1,39 @@
# Optional Jev demo. Install reme-ai[core,classifier-jev] and set CLASSIFIER_API_KEY.
# Start the service: reme start config=classifier_demo
# Call it from another terminal: reme jev state="I was charged twice."
extends: default
jobs:
jev:
backend: base
enable_serve: true
description: "Classify customer requests with Jev; returns typed answers, probabilities, and usage."
parameters:
type: object
properties:
state:
type: [string, object]
minLength: 1
description: "Text or JSON object to classify"
questions:
type: object
minProperties: 1
additionalProperties:
type: object
description: "Named binary, choice, or score questions; omission uses three customer-support demo questions"
required: [state]
steps:
- backend: classifier_demo_step
as_classifier: default
components:
as_classifier:
default:
backend: ${CLASSIFIER_BACKEND:-jev}
model: ${CLASSIFIER_MODEL_NAME:-jev-latest}
credential:
api_key: ${CLASSIFIER_API_KEY:-}
base_url: ${CLASSIFIER_BASE_URL:-null}
timeout: 30
max_retries: 2
retry_delay: 0.5

View file

@ -845,6 +845,17 @@ components:
# backend: local
# as_embedding: default
# as_classifier:
# default:
# backend: ${CLASSIFIER_BACKEND:-jev}
# model: ${CLASSIFIER_MODEL_NAME:-jev-latest}
# credential:
# api_key: ${CLASSIFIER_API_KEY:-}
# base_url: ${CLASSIFIER_BASE_URL:-null}
# timeout: 30
# max_retries: 2
# retry_delay: 0.5
as_llm:
default:
backend: ${LLM_BACKEND:-openai}

View file

@ -10,6 +10,8 @@ class ComponentEnum(str, Enum):
AS_LLM = "as_llm"
AS_CLASSIFIER = "as_classifier"
AS_EMBEDDING = "as_embedding"
EMBEDDING_STORE = "embedding_store"

View file

@ -3,6 +3,7 @@
from .add import AddStep
from .app_config import AppConfigStep
from .chat import ChatStep
from .classifier_demo import ClassifierDemoStep
from .demo import DemoEchoStep1, DemoEchoStep2
from .health_check import HealthCheckStep
from .help import HelpStep
@ -18,6 +19,7 @@ __all__ = [
"AddStep",
"AppConfigStep",
"ChatStep",
"ClassifierDemoStep",
"DemoEchoStep1",
"DemoEchoStep2",
"HealthCheckStep",

View file

@ -0,0 +1,43 @@
"""Demo for typed probabilistic classification of text or JSON state."""
import json
from dataclasses import asdict
from ..base_step import BaseStep, Ref
from ...components import R
from ...components.as_classifier import BaseAsClassifier
from ...enumeration import ComponentEnum
@R.register("classifier_demo_step")
class ClassifierDemoStep(BaseStep):
"""Return complete classifier output and share it with subsequent Steps."""
as_classifier = Ref(BaseAsClassifier, ComponentEnum.AS_CLASSIFIER)
@staticmethod
def default_questions() -> dict:
"""Create independent demo questions for each invocation."""
return {
"urgent": {"type": "binary", "instructions": "Does this customer request need urgent attention?"},
"route": {
"type": "choice",
"instructions": "Select the team that should handle this customer request.",
"criteria": {"billing": "Payments and charges", "support": "Technical help", "other": "Other requests"},
},
"priority": {
"type": "score",
"instructions": "Rate the priority of this customer request.",
"criteria": ["Low", "Medium", "High"],
},
}
async def execute(self):
assert self.context is not None
# Explicit null/empty questions are invalid; only omission enables the demo defaults.
questions = self.context.get("questions", self.default_questions())
result = await self.as_classifier(state=self.context.get("state"), questions=questions)
answer = json.loads(json.dumps(asdict(result), ensure_ascii=False, allow_nan=False))
self.context.data["classifier_result"] = answer
self.context.response.answer = answer
return self.context.response

View file

@ -0,0 +1,319 @@
"""Jev transport is optional, lazy, validated, and lifecycle managed."""
# pylint: disable=protected-access,redefined-outer-name
import asyncio
import sys
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock
import pytest
from pydantic import ValidationError
from reme.components.as_classifier import JevAsClassifier
from reme.components.as_classifier import jev as jev_module
@pytest.fixture
def classifier_api():
"""Use actual AgentScope schemas when the optional adapter is installed."""
return pytest.importorskip("agentscope.classifier")
@pytest.fixture
def provider(classifier_api, monkeypatch):
"""Keep the question schema real while mocking the model boundary."""
model = AsyncMock(return_value=SimpleNamespace(model="jev-test"))
model.client = SimpleNamespace(aclose=AsyncMock())
factory = Mock(return_value=model)
monkeypatch.setattr(classifier_api, "JevClassifierModel", factory)
return factory, model
def test_start_and_close_never_load_optional_dependencies(monkeypatch):
"""An enabled but unused component does not need credentials or an SDK."""
loader = Mock(side_effect=AssertionError("Unexpected provider import"))
monkeypatch.setattr(jev_module, "import_module", loader)
async def run():
component = JevAsClassifier()
await component.start()
await component.close()
await component.close()
assert component._model is None
loader.assert_not_called()
asyncio.run(run())
@pytest.mark.parametrize("failure", [ModuleNotFoundError("No classifier"), AttributeError("No Jev export")])
def test_old_agentscope_has_actionable_error(monkeypatch, failure):
"""Older installations fail on use, not at application startup."""
monkeypatch.setattr(jev_module, "import_module", Mock(side_effect=failure))
component = JevAsClassifier()
with pytest.raises(ImportError, match=r"reme-ai\[classifier-jev\]"):
asyncio.run(component(state="hello", questions={"q": {"type": "binary"}}))
@pytest.mark.parametrize(
"options",
[{"timeout": 0}, {"timeout": float("inf")}, {"max_retries": -1}, {"retry_delay": -1}, {"model": ""}],
)
def test_transport_options_are_validated_locally(options):
"""Invalid numeric options cannot silently reach the SDK."""
with pytest.raises(ValidationError):
JevAsClassifier(**options)
@pytest.mark.parametrize(
"state,questions",
[
(None, {"q": {"type": "binary"}}),
(" ", {"q": {"type": "binary"}}),
(["hello"], {"q": {"type": "binary"}}),
({"bad": object()}, {"q": {"type": "binary"}}),
({"bad": float("nan")}, {"q": {"type": "binary"}}),
("hello", {}),
("hello", None),
("hello", {" ": {"type": "binary"}}),
],
)
def test_invalid_request_shape_never_loads_provider(monkeypatch, state, questions):
"""Reject malformed state and mappings before optional imports."""
loader = Mock(side_effect=AssertionError("Unexpected provider import"))
monkeypatch.setattr(jev_module, "import_module", loader)
with pytest.raises(ValueError):
asyncio.run(JevAsClassifier()(state=state, questions=questions))
loader.assert_not_called()
def test_valid_questions_reuse_model_and_restart_closes_clients(provider, classifier_api):
"""Concurrent calls construct one model; close and restart own each client."""
factory, model = provider
questions = {
"binary": {"type": "binary", "criteria": {"true": "Urgent", "false": "Routine"}},
"choice": {"type": "choice", "criteria": {"billing": None, "support": "Technical help"}},
"score": {"type": "score", "criteria": ["low", "high"]},
}
async def run():
component = JevAsClassifier(
credential={"api_key": "test-key", "base_url": ""},
model="jev-test",
timeout=12,
max_retries=4,
retry_delay=0.2,
)
await component.start()
factory.assert_not_called()
responses = await asyncio.gather(*(component(state={"text": "hello"}, questions=questions) for _ in range(3)))
assert all(response.model == "jev-test" for response in responses)
factory.assert_called_once()
options = factory.call_args.kwargs
assert options["credential"].api_key.get_secret_value() == "test-key"
assert options["credential"].base_url is None
assert {key: value for key, value in options.items() if key != "credential"} == {
"model": "jev-test",
"timeout": 12.0,
"max_retries": 4,
"retry_delay": 0.2,
}
parsed = model.await_args.kwargs["questions"]
assert isinstance(parsed["binary"], classifier_api.BinaryQuestion)
assert isinstance(parsed["choice"], classifier_api.ChoiceQuestion)
assert isinstance(parsed["score"], classifier_api.ScoreQuestion)
await component.close()
await component.close()
model.client.aclose.assert_awaited_once()
assert component._model is None
await component(state="hello", questions={"q": classifier_api.BinaryQuestion()})
assert factory.call_count == 2
await component.close()
assert model.client.aclose.await_count == 2
asyncio.run(run())
@pytest.mark.parametrize(
"question",
[
{"type": "unknown"},
{"type": "choice", "criteria": {}},
{"type": "score", "criteria": []},
{"type": "binary", "unexpected": True},
],
)
def test_invalid_question_never_constructs_model(provider, question):
"""Use the provider-independent schema before touching credentials."""
factory, model = provider
with pytest.raises(ValidationError):
asyncio.run(JevAsClassifier()(state="hello", questions={"q": question}))
factory.assert_not_called()
model.assert_not_awaited()
def test_missing_credentials_can_be_corrected_after_failure(provider):
"""Failure leaves no partially initialized model and permits a later call."""
factory, _ = provider
async def run():
component = JevAsClassifier()
try:
with pytest.raises(ValueError, match="CLASSIFIER_API_KEY"):
await component(state="hello", questions={"q": {"type": "binary"}})
factory.assert_not_called()
assert component._model is None
component.credential["api_key"] = "test-key"
await component(state="hello", questions={"q": {"type": "binary"}})
finally:
await component.close()
asyncio.run(run())
def test_close_waits_for_all_active_requests(provider, monkeypatch):
"""Shutdown cannot close a shared client while another call is using it."""
_, model = provider
async def run():
started = asyncio.Event()
closing = asyncio.Event()
releases = [asyncio.Event(), asyncio.Event()]
count = 0
async def reply(**_kwargs):
nonlocal count
index = count
count += 1
if count == 2:
started.set()
await releases[index].wait()
return SimpleNamespace(model="jev-test")
model.side_effect = reply
component = JevAsClassifier(credential={"api_key": "test-key"})
original_close = component._close
async def mark_close():
closing.set()
await original_close()
monkeypatch.setattr(component, "_close", mark_close)
calls = [asyncio.create_task(component(state="hello", questions={"q": {"type": "binary"}})) for _ in range(2)]
await asyncio.wait_for(started.wait(), timeout=2)
close = asyncio.create_task(component.close())
await asyncio.wait_for(closing.wait(), timeout=2)
assert not close.done()
model.client.aclose.assert_not_awaited()
releases[0].set()
await asyncio.wait_for(calls[0], timeout=2)
assert not close.done()
model.client.aclose.assert_not_awaited()
releases[1].set()
await asyncio.wait_for(calls[1], timeout=2)
await asyncio.wait_for(close, timeout=2)
model.client.aclose.assert_awaited_once()
assert component._model is None
asyncio.run(run())
@pytest.mark.parametrize("failure", [TimeoutError("timed out"), asyncio.CancelledError()])
def test_failed_or_cancelled_call_does_not_block_close(provider, failure):
"""Both normal failures and task cancellation release the active-call slot."""
_, model = provider
model.side_effect = failure
async def run():
component = JevAsClassifier(credential={"api_key": "test-key"})
with pytest.raises(type(failure)):
await component(state="hello", questions={"q": {"type": "binary"}})
await asyncio.wait_for(component.close(), timeout=2)
model.client.aclose.assert_awaited_once()
asyncio.run(run())
def test_client_close_error_clears_cached_model(provider):
"""A failed close cannot leave a stale client selected on the next start."""
factory, model = provider
async def run():
component = JevAsClassifier(credential={"api_key": "test-key"})
await component(state="hello", questions={"q": {"type": "binary"}})
model.client.aclose.side_effect = RuntimeError("close failed")
with pytest.raises(RuntimeError, match="close failed"):
await component.close()
assert component._model is None
assert not component.is_started
model.client.aclose.side_effect = None
await component(state="hello", questions={"q": {"type": "binary"}})
assert factory.call_count == 2
await component.close()
asyncio.run(run())
def test_missing_sdk_has_actionable_error(classifier_api, monkeypatch):
"""The actual AgentScope constructor reports the optional dependency failure."""
assert classifier_api.JevClassifierModel is not None
monkeypatch.setitem(sys.modules, "typesafe_sdk", None)
async def run():
component = JevAsClassifier(credential={"api_key": "test-key"})
try:
with pytest.raises(ImportError, match=r"reme-ai\[classifier-jev\]"):
await component(state="hello", questions={"q": {"type": "binary"}})
assert component._model is None
finally:
await component.close()
asyncio.run(run())
def test_sdk_response_and_retry_policy_round_trip(classifier_api, monkeypatch):
"""Exercise the real adapter with official SDK types and a mocked HTTP boundary."""
sdk = pytest.importorskip("typesafe_sdk")
client = SimpleNamespace(
system_one=AsyncMock(
return_value=SimpleNamespace(
model="jev-test",
answers={
"binary": sdk.NoulAnswer(type="noul", noul=0.8),
"choice": sdk.ChoiceAnswer(type="choice", choice="a", confidence=1, probabilities={"a": 1}),
"score": sdk.ScoreAnswer(
type="score",
score=0.7,
confidence=0.7,
legend={0: "low", 1: "high"},
probabilities={0: 0.3, 1: 0.7},
),
},
usage=sdk.Usage(input_tokens=12, output_tokens=3),
),
),
aclose=AsyncMock(),
)
factory = Mock(return_value=client)
monkeypatch.setattr(sdk, "AsyncTypeSafeClient", factory)
async def run():
async with JevAsClassifier(credential={"api_key": "test-key"}, max_retries=4, retry_delay=0.2) as component:
response = await component(
state="hello",
questions={
"binary": {"type": "binary"},
"choice": {"type": "choice", "criteria": {"a": None}},
"score": {"type": "score", "criteria": ["low", "high"]},
},
)
assert isinstance(response, classifier_api.ClassifierResponse)
assert response.content["binary"].probability == 0.8
assert response.content["choice"].choice == "a"
assert response.content["score"].score == 0.7
assert response.usage.input_tokens == 12
assert factory.call_args.kwargs["retry"] == sdk.RetryPolicy(max_retries=4, backoff_initial=0.2)
assert client.system_one.await_args.kwargs["questions"]["binary"].type == "noul"
client.aclose.assert_awaited_once()
asyncio.run(run())

View file

@ -0,0 +1,179 @@
"""Classifier registration, optional config, and secret redaction."""
# pylint: disable=redefined-outer-name,unused-argument
import asyncio
from dataclasses import dataclass, field
from importlib import import_module
import json
from pathlib import Path
import subprocess
import sys
import tomllib
import pytest
import httpx
from reme.application import Application
from reme.components import R
from reme.components.as_classifier import JevAsClassifier
from reme.components.as_classifier import jev as jev_module
from reme.components.client import HttpClient
from reme.config import parse_kwargs, resolve_app_config
from reme.enumeration import ComponentEnum
from reme.steps.common.app_config import AppConfigStep
from reme.steps.common.classifier_demo import ClassifierDemoStep
@pytest.fixture
def classifier_env(monkeypatch):
"""Keep all examples independent of the user's local provider settings."""
for name in ("CLASSIFIER_BACKEND", "CLASSIFIER_MODEL_NAME", "CLASSIFIER_API_KEY", "CLASSIFIER_BASE_URL"):
monkeypatch.delenv(name, raising=False)
def test_registration_and_default_opt_in(classifier_env):
"""Imports register providers, but default jobs never enable classification."""
assert R.get(ComponentEnum.AS_CLASSIFIER, "jev") is JevAsClassifier
assert R.get(ComponentEnum.STEP, "classifier_demo_step") is ClassifierDemoStep
default = resolve_app_config(log_config=False)
assert "as_classifier" not in default["components"]
assert "jev" not in default["jobs"]
def test_demo_config_env_and_cli_override(classifier_env, monkeypatch):
"""Provider env expands and CLI transport settings deep-merge over the file."""
monkeypatch.setenv("CLASSIFIER_API_KEY", "test-key")
monkeypatch.setenv("CLASSIFIER_MODEL_NAME", "jev-test")
config = resolve_app_config(
log_config=False,
**parse_kwargs(
"config=classifier_demo",
"components.as_classifier.default.timeout=12",
),
)
component = config["components"]["as_classifier"]["default"]
assert component == {
"backend": "jev",
"model": "jev-test",
"credential": {"api_key": "test-key", "base_url": None},
"timeout": 12,
"max_retries": 2,
"retry_delay": 0.5,
}
default = resolve_app_config(log_config=False)
assert config["service"] == default["service"]
assert config["service"]["backend"] == "http"
assert config["jobs"]["jev"]["enable_serve"] is True
assert "classifier_demo" not in config["jobs"]
assert {name: job for name, job in config["jobs"].items() if name != "jev"} == default["jobs"]
assert {name: group for name, group in config["components"].items() if name != "as_classifier"} == default[
"components"
]
def test_application_starts_without_optional_imports_and_redacts_key(classifier_env, monkeypatch, tmp_path):
"""The inherited config wires components without creating a classifier client."""
monkeypatch.setenv("CLASSIFIER_API_KEY", "test-key")
monkeypatch.setattr(jev_module, "import_module", lambda _: pytest.fail("Unexpected optional import"))
config = resolve_app_config(config="classifier_demo", log_config=False)
async def run():
app = Application(
**config,
workspace_dir=str(tmp_path),
enable_logo=False,
log_to_console=False,
log_to_file=False,
)
async with app:
response = await AppConfigStep(app_context=app.context)()
assert response.answer["components"]["as_classifier"]["default"]["credential"]["api_key"] == "***"
asyncio.run(run())
def test_jev_client_calls_inherited_http_service(classifier_env, monkeypatch, tmp_path, capsys):
"""The two-stage workflow exposes /jev and sends client payloads to its Step."""
@dataclass
class Result:
"""Mock only the classifier boundary; exercise the actual service and client."""
model: str = "jev-test"
content: dict = field(default_factory=lambda: {"urgent": {"type": "binary_answer", "probability": 0.8}})
calls = []
async def classify(_self, state, questions):
calls.append((state, questions))
return Result()
monkeypatch.setattr(JevAsClassifier, "__call__", classify)
config = resolve_app_config(config="classifier_demo", log_config=False)
reme_module = import_module("reme.reme")
monkeypatch.setattr(reme_module, "running_app_config", lambda: config)
async def run():
app = Application(
**config,
workspace_dir=str(tmp_path),
enable_logo=False,
log_to_console=False,
log_to_file=False,
)
service = app.context.service
service.build_service(app)
service.add_jobs(app)
assert "/jev" in {route.path for route in service.service.routes}
assert service.mcp_server is not None
assert "jev" in {tool.name for tool in await service.mcp_server.list_tools()}
async def start_client(client):
client.client = httpx.AsyncClient(
transport=httpx.ASGITransport(app=service.service),
base_url=client.base_url,
)
monkeypatch.setattr(HttpClient, "_start", start_client)
async with app:
await reme_module.call_server("jev", state="I was charged twice.")
displayed = json.loads(capsys.readouterr().out)
assert displayed["success"] is True
assert displayed["answer"]["content"]["urgent"]["probability"] == 0.8
assert calls[0][0] == "I was charged twice."
assert {q["type"] for q in calls[0][1].values()} == {"binary", "choice", "score"}
custom = {"only": {"type": "binary", "instructions": "Is this urgent?"}}
await reme_module.call_server("jev", state={"request": "hello"}, questions=custom)
assert calls[1] == ({"request": "hello"}, custom)
assert json.loads(capsys.readouterr().out)["success"] is True
asyncio.run(run())
def test_import_with_optional_modules_blocked():
"""A fresh interpreter catches accidental optional imports anywhere in bootstrap."""
script = """
import sys
from importlib.abc import MetaPathFinder
class BlockOptional(MetaPathFinder):
def find_spec(self, fullname, path=None, target=None):
if fullname == 'agentscope.classifier' or fullname.startswith('typesafe_sdk'):
raise ModuleNotFoundError(fullname)
sys.meta_path.insert(0, BlockOptional())
import reme
from reme.components.as_classifier import JevAsClassifier
from reme.steps.common.classifier_demo import ClassifierDemoStep
"""
result = subprocess.run([sys.executable, "-c", script], capture_output=True, text=True, check=False)
assert result.returncode == 0, result.stderr
def test_extra_reuses_agentscope_pin():
"""Installing core and classifier together must not request different AgentScope versions."""
config = tomllib.loads((Path(__file__).resolve().parents[2] / "pyproject.toml").read_text())
extras = config["project"]["optional-dependencies"]
assert extras["classifier-jev"] == ["agentscope[classifier-jev]==2.0.10"]
assert extras["as"] == ["agentscope[model-ollama]==2.0.10"]
assert "reme-ai[classifier-jev]" in extras["full"]
assert "reme-ai[classifier-jev]" not in extras["core"]

View file

@ -0,0 +1,142 @@
"""Classifier demo wiring and JSON output without optional SDK dependencies."""
import asyncio
from dataclasses import dataclass, field
from unittest.mock import AsyncMock
import pytest
from reme.components.application_context import ApplicationContext
from reme.components.as_classifier import BaseAsClassifier
from reme.components.job import BaseJob
from reme.components.runtime_context import RuntimeContext
from reme.enumeration import ComponentEnum
from reme.steps.base_step import BaseStep
from reme.steps.common.classifier_demo import ClassifierDemoStep
@dataclass
class _Result:
"""Small classifier-response stand-in for Step boundary tests."""
model: str = "jev-test"
content: dict = field(
default_factory=lambda: {
"urgent": {"type": "binary_answer", "probability": 0.8},
"priority": {"type": "score_answer", "score": 0.7, "legend": {0: "low", 1: "high"}},
},
)
usage: dict = field(default_factory=lambda: {"input_tokens": 12, "output_tokens": 2})
class _Classifier(BaseAsClassifier):
"""Recording component accepting the same inputs as a provider."""
def __init__(self):
super().__init__()
self.call = AsyncMock(return_value=_Result())
async def __call__(self, state, questions):
return await self.call(state=state, questions=questions)
def test_demo_serializes_complete_result_and_preserves_context():
"""The answer is JSON-ready and available to later Steps in the same request."""
async def run():
classifier = _Classifier()
step = ClassifierDemoStep(as_classifier=classifier)
context = RuntimeContext(state="I was charged twice.", keep="original")
context.response.metadata["earlier"] = True
response = await step(context)
assert response.success
assert response.answer["model"] == "jev-test"
assert response.answer["content"]["urgent"]["probability"] == 0.8
assert response.answer["content"]["priority"]["legend"] == {"0": "low", "1": "high"}
assert response.answer["usage"]["input_tokens"] == 12
assert context.data["classifier_result"] is response.answer
assert context.data["keep"] == "original"
assert response.metadata == {"earlier": True}
assert {q["type"] for q in classifier.call.await_args.kwargs["questions"].values()} == {
"binary",
"choice",
"score",
}
asyncio.run(run())
@pytest.mark.parametrize("questions", [{"custom": {"type": "binary"}}, {}, None])
def test_explicit_questions_replace_defaults(questions):
"""Even invalid explicit values reach component validation rather than demo fallback."""
async def run():
classifier = _Classifier()
await ClassifierDemoStep(as_classifier=classifier)(state={"request": "hello"}, questions=questions)
assert classifier.call.await_args.kwargs["questions"] == questions
asyncio.run(run())
def test_ref_uses_named_component_and_clears_between_calls(tmp_path):
"""Reusing a standalone Step respects each request's component selection."""
async def run():
context = ApplicationContext(workspace_dir=str(tmp_path))
first, second = _Classifier(), _Classifier()
context.components[ComponentEnum.AS_CLASSIFIER] = {"first": first, "second": second}
step = ClassifierDemoStep(app_context=context)
await step(state="hello", as_classifier="first")
await step(state="hello", as_classifier="second")
first.call.assert_awaited_once()
second.call.assert_awaited_once()
asyncio.run(run())
def test_job_shares_result_and_rebuilds_steps(tmp_path):
"""Sequential Steps see classifier data, and no result leaks across invocations."""
class ConsumeResultStep(BaseStep):
"""Read the prior Step's output using the ordinary request context."""
async def execute(self):
self.context.response.metadata["consumer_model"] = self.context.data["classifier_result"]["model"]
async def run():
context = ApplicationContext(workspace_dir=str(tmp_path))
classifier = _Classifier()
context.components[ComponentEnum.AS_CLASSIFIER] = {"default": classifier}
context.registry.add("consume_classifier_result", ConsumeResultStep, owner="test")
job = BaseJob(
app_context=context,
steps=[
{"backend": "classifier_demo_step"},
{"backend": "consume_classifier_result"},
],
)
async with job:
first = await job(state="first")
second = await job(state="second")
assert first.success and second.success
assert first.metadata["consumer_model"] == "jev-test"
assert first.answer is not second.answer
assert [call.kwargs["state"] for call in classifier.call.await_args_list] == ["first", "second"]
asyncio.run(run())
def test_job_converts_provider_error_to_failed_response(tmp_path):
"""SDK errors follow the existing BaseJob error contract."""
async def run():
context = ApplicationContext(workspace_dir=str(tmp_path))
classifier = _Classifier()
classifier.call.side_effect = TimeoutError("classifier timed out")
context.components[ComponentEnum.AS_CLASSIFIER] = {"default": classifier}
async with BaseJob(app_context=context, steps=[{"backend": "classifier_demo_step"}]) as job:
response = await job(state="hello")
assert not response.success
assert "classifier timed out" in response.answer
asyncio.run(run())