From 43ef6655954e286c99fe5abb49c4e695fbc6e76d Mon Sep 17 00:00:00 2001 From: jinliyl <6469360+jinliyl@users.noreply.github.com> Date: Wed, 14 Oct 2026 17:49:49 +0800 Subject: [PATCH] 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 --- .github/workflows/ci-python-tests.yml | 2 +- README.md | 3 + README_ZH.md | 3 + docs/en/configuration.md | 84 +++++ docs/en/framework.md | 2 + docs/zh/configuration.md | 78 +++++ docs/zh/framework.md | 2 + example.env | 7 + pyproject.toml | 4 + reme/components/__init__.py | 2 + reme/components/as_classifier/__init__.py | 6 + .../as_classifier/base_as_classifier.py | 25 ++ reme/components/as_classifier/jev.py | 121 +++++++ reme/config/classifier_demo.yaml | 39 +++ reme/config/default.yaml | 11 + reme/enumeration/component_enum.py | 2 + reme/steps/common/__init__.py | 2 + reme/steps/common/classifier_demo.py | 43 +++ tests/unit/test_as_classifier.py | 319 ++++++++++++++++++ tests/unit/test_classifier_config.py | 179 ++++++++++ tests/unit/test_classifier_demo.py | 142 ++++++++ 21 files changed, 1075 insertions(+), 1 deletion(-) create mode 100644 reme/components/as_classifier/__init__.py create mode 100644 reme/components/as_classifier/base_as_classifier.py create mode 100644 reme/components/as_classifier/jev.py create mode 100644 reme/config/classifier_demo.yaml create mode 100644 reme/steps/common/classifier_demo.py create mode 100644 tests/unit/test_as_classifier.py create mode 100644 tests/unit/test_classifier_config.py create mode 100644 tests/unit/test_classifier_demo.py diff --git a/.github/workflows/ci-python-tests.yml b/.github/workflows/ci-python-tests.yml index 5eeb5d29..1d3971cb 100644 --- a/.github/workflows/ci-python-tests.yml +++ b/.github/workflows/ci-python-tests.yml @@ -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 diff --git a/README.md b/README.md index 5f898e11..0343ab00 100644 --- a/README.md +++ b/README.md @@ -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: diff --git a/README_ZH.md b/README_ZH.md index 82126ffa..05b04fe0 100644 --- a/README_ZH.md +++ b/README_ZH.md @@ -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： diff --git a/docs/en/configuration.md b/docs/en/configuration.md index 438397f4..50e9091d 100644 --- a/docs/en/configuration.md +++ b/docs/en/configuration.md @@ -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: `. +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`. diff --git a/docs/en/framework.md b/docs/en/framework.md index ab2b286e..fef3d4b6 100644 --- a/docs/en/framework.md +++ b/docs/en/framework.md @@ -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. | diff --git a/docs/zh/configuration.md b/docs/zh/configuration.md index 956c3254..7076a7c0 100644 --- a/docs/zh/configuration.md +++ b/docs/zh/configuration.md @@ -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`： diff --git a/docs/zh/framework.md b/docs/zh/framework.md index 2457806a..e5f2e1fa 100644 --- a/docs/zh/framework.md +++ b/docs/zh/framework.md @@ -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 | diff --git a/example.env b/example.env index 72bfd5e4..9eed1eed 100644 --- a/example.env +++ b/example.env @@ -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 diff --git a/pyproject.toml b/pyproject.toml index 4c98feea..6adfc24f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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]", ] diff --git a/reme/components/__init__.py b/reme/components/__init__.py index f3b630c6..5cfa9ff9 100644 --- a/reme/components/__init__.py +++ b/reme/components/__init__.py @@ -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", diff --git a/reme/components/as_classifier/__init__.py b/reme/components/as_classifier/__init__.py new file mode 100644 index 00000000..e9b5f97a --- /dev/null +++ b/reme/components/as_classifier/__init__.py @@ -0,0 +1,6 @@ +"""Optional probabilistic classifier components.""" + +from .base_as_classifier import BaseAsClassifier +from .jev import JevAsClassifier + +__all__ = ["BaseAsClassifier", "JevAsClassifier"] diff --git a/reme/components/as_classifier/base_as_classifier.py b/reme/components/as_classifier/base_as_classifier.py new file mode 100644 index 00000000..fafcb8a9 --- /dev/null +++ b/reme/components/as_classifier/base_as_classifier.py @@ -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.""" diff --git a/reme/components/as_classifier/jev.py b/reme/components/as_classifier/jev.py new file mode 100644 index 00000000..fa482079 --- /dev/null +++ b/reme/components/as_classifier/jev.py @@ -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() diff --git a/reme/config/classifier_demo.yaml b/reme/config/classifier_demo.yaml new file mode 100644 index 00000000..42aef3bf --- /dev/null +++ b/reme/config/classifier_demo.yaml @@ -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 diff --git a/reme/config/default.yaml b/reme/config/default.yaml index e162d044..36184d1d 100644 --- a/reme/config/default.yaml +++ b/reme/config/default.yaml @@ -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} diff --git a/reme/enumeration/component_enum.py b/reme/enumeration/component_enum.py index 22095464..6b7b2743 100644 --- a/reme/enumeration/component_enum.py +++ b/reme/enumeration/component_enum.py @@ -10,6 +10,8 @@ class ComponentEnum(str, Enum): AS_LLM = "as_llm" + AS_CLASSIFIER = "as_classifier" + AS_EMBEDDING = "as_embedding" EMBEDDING_STORE = "embedding_store" diff --git a/reme/steps/common/__init__.py b/reme/steps/common/__init__.py index 50b5def1..d0feb3c7 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 .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", diff --git a/reme/steps/common/classifier_demo.py b/reme/steps/common/classifier_demo.py new file mode 100644 index 00000000..22cafe9f --- /dev/null +++ b/reme/steps/common/classifier_demo.py @@ -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 diff --git a/tests/unit/test_as_classifier.py b/tests/unit/test_as_classifier.py new file mode 100644 index 00000000..79066cc8 --- /dev/null +++ b/tests/unit/test_as_classifier.py @@ -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()) diff --git a/tests/unit/test_classifier_config.py b/tests/unit/test_classifier_config.py new file mode 100644 index 00000000..056cc5b3 --- /dev/null +++ b/tests/unit/test_classifier_config.py @@ -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"] diff --git a/tests/unit/test_classifier_demo.py b/tests/unit/test_classifier_demo.py new file mode 100644 index 00000000..ca681a20 --- /dev/null +++ b/tests/unit/test_classifier_demo.py @@ -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())