diff --git a/src/powercontext/builtin/artifacts/memory/service.py b/src/powercontext/builtin/artifacts/memory/service.py index c5706f43b8..4b55e2da19 100644 --- a/src/powercontext/builtin/artifacts/memory/service.py +++ b/src/powercontext/builtin/artifacts/memory/service.py @@ -86,6 +86,7 @@ EmbeddingVector, InferenceTimeoutError, InferenceUnavailableError, + embed_query, ) from powercontext.builtin.tags import TagFilter from powercontext.errors import RevisionConflictError @@ -539,7 +540,7 @@ async def _resolve_query_vector( if reuse is not None and reuse.embedding_profile == profile: return selected_mode, reuse.query_vector, reuse, 0 try: - query_vector = (await self._embed_texts((query,), profile))[0] + query_vector = (await self._embed_texts((query,), profile, query=True))[0] except (InferenceUnavailableError, InferenceTimeoutError) as error: if requested_mode == "auto" and capabilities.fts: return "fts", None, None, 1 @@ -987,11 +988,13 @@ async def _embed_texts( self, texts: tuple[str, ...], profile: EmbeddingProfile, + *, + query: bool = False, ) -> tuple[EmbeddingVector, ...]: embedding_model = self._embedding_model if embedding_model is None or embedding_model.profile != profile: raise CapabilityNotSupportedError("embedding-profile") - result = await embedding_model.embed(texts) + result = await embed_query(embedding_model, texts) if query else await embedding_model.embed(texts) vectors = result.vectors if len(vectors) != len(texts): raise InvalidEmbeddingError("count") diff --git a/src/powercontext/builtin/inference/__init__.py b/src/powercontext/builtin/inference/__init__.py index d66c29763b..980e16b6f3 100644 --- a/src/powercontext/builtin/inference/__init__.py +++ b/src/powercontext/builtin/inference/__init__.py @@ -31,7 +31,12 @@ GenerationResult, InferenceUsage, ) -from powercontext.builtin.inference.protocols import EmbeddingModel, StructuredGenerator +from powercontext.builtin.inference.protocols import ( + EmbeddingModel, + QueryEmbeddingModel, + StructuredGenerator, + embed_query, +) from powercontext.builtin.inference.tokens import TokenEstimator, TokenEstimatorProfile, character_token_estimator __all__ = [ @@ -45,8 +50,10 @@ "InferenceUnavailableError", "InferenceUsage", "InvalidInferenceOutputError", + "QueryEmbeddingModel", "StructuredGenerator", "TokenEstimator", "TokenEstimatorProfile", "character_token_estimator", + "embed_query", ] diff --git a/src/powercontext/builtin/inference/errors.py b/src/powercontext/builtin/inference/errors.py index abdeba68e2..91dea21661 100644 --- a/src/powercontext/builtin/inference/errors.py +++ b/src/powercontext/builtin/inference/errors.py @@ -30,9 +30,13 @@ class InferenceConfigurationError(InferenceError, RuntimeError): class InferenceUnavailableError(InferenceError, RuntimeError): """Raised when a transient provider failure prevents inference.""" - def __init__(self, operation: str) -> None: + def __init__(self, operation: str, detail: str | None = None) -> None: self.operation = operation - super().__init__(f"inference is temporarily unavailable for {operation}") + self.detail = detail + message = f"inference is temporarily unavailable for {operation}" + if detail is not None: + message = f"{message}: {detail}" + super().__init__(message) class InferenceTimeoutError(InferenceError, TimeoutError): diff --git a/src/powercontext/builtin/inference/minimax.py b/src/powercontext/builtin/inference/minimax.py new file mode 100644 index 0000000000..9ecc8bf8d0 --- /dev/null +++ b/src/powercontext/builtin/inference/minimax.py @@ -0,0 +1,198 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""MiniMax embedding backend adapter. + +MiniMax publishes embeddings under the OpenAI path prefix but does not implement +the OpenAI embeddings contract. Its ``/v1/embeddings`` endpoint requires the +native request shape ``{"model", "texts": [...], "type"}`` instead of OpenAI's +``{"input": [...]}``, and returns ``{"vectors": [...], "base_resp": {"status_code"}}`` +instead of ``{"data": [{"embedding": [...]}]}``. It also answers HTTP 200 with a +non-zero ``base_resp.status_code`` on error rather than a 4xx/5xx. This adapter +speaks the MiniMax-native shape directly so PowerContext can use MiniMax as an +embedding provider without a custom Pydantic AI provider shim. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Mapping, Sequence +from urllib.parse import urlparse + +import httpx + +from powercontext.builtin.artifacts.memory.canonical import canonical_embedding +from powercontext.builtin.artifacts.memory.models import EmbeddingProfile +from powercontext.builtin.inference.errors import ( + InferenceConfigurationError, + InferenceTimeoutError, + InferenceUnavailableError, + InvalidInferenceOutputError, +) +from powercontext.builtin.inference.models import EmbeddingResult, InferenceUsage + +_MINIMAX_DOCUMENT_TYPE = "db" +_MINIMAX_QUERY_TYPE = "query" + + +class MiniMaxEmbeddingModel: + """Embed an ordered text batch with MiniMax ``embo`` via ``/v1/embeddings``.""" + + def __init__( + self, + *, + base_url: str, + model: str, + headers: Mapping[str, str] | None = None, + profile: EmbeddingProfile, + batch_size: int = 10, + timeout_seconds: float = 30.0, + http_client: httpx.AsyncClient | None = None, + ) -> None: + if not model or not model.strip(): + raise InferenceConfigurationError("embedding-model-empty") + if profile.dimension < 1: + raise InferenceConfigurationError("embedding-dimension-positive") + if batch_size < 1: + raise InferenceConfigurationError("embedding-batch-size-positive") + if not profile.profile_id.strip() or not profile.model.strip() or not profile.normalization.strip(): + raise InferenceConfigurationError("embedding-profile-identifiers") + self.profile = profile + self._model = model + self._batch_size = batch_size + self._timeout = timeout_seconds + self._endpoint = f"{base_url.rstrip('/')}/embeddings" + request_headers: dict[str, str] = {"Content-Type": "application/json"} + if headers: + request_headers.update(headers) + self._headers = request_headers + self._client = http_client or httpx.AsyncClient(timeout=timeout_seconds) + + async def aclose(self) -> None: + """Close the underlying HTTP client (registered with the runtime stack).""" + + await self._client.aclose() + + async def embed(self, texts: tuple[str, ...], /) -> EmbeddingResult: + """Embed documents, validating order, count, dimension, and finite values.""" + + return await self._embed(texts, embedding_type=_MINIMAX_DOCUMENT_TYPE) + + async def embed_query(self, texts: tuple[str, ...], /) -> EmbeddingResult: + """Embed retrieval queries, validating order, count, dimension, and finite values.""" + + return await self._embed(texts, embedding_type=_MINIMAX_QUERY_TYPE) + + async def _embed(self, texts: tuple[str, ...], *, embedding_type: str) -> EmbeddingResult: + """Embed one MiniMax document or query batch.""" + + if not texts: + return EmbeddingResult(vectors=()) + + try: + result = await asyncio.wait_for( + self._embed_batches(texts, embedding_type=embedding_type), timeout=self._timeout + ) + except asyncio.CancelledError: + raise + except (InvalidInferenceOutputError, InferenceConfigurationError): + raise + except TimeoutError as error: + raise InferenceTimeoutError("embed", self._timeout) from error + except httpx.TimeoutException as error: + raise InferenceTimeoutError("embed", self._timeout) from error + except httpx.HTTPError as error: + raise InferenceUnavailableError("embed") from error + return result + + async def _embed_batches(self, texts: tuple[str, ...], *, embedding_type: str) -> EmbeddingResult: + vectors: list[tuple[float, ...]] = [] + requests = 0 + input_tokens = 0 + for start in range(0, len(texts), self._batch_size): + batch = texts[start : start + self._batch_size] + rows, tokens = await self._embed_one(batch, embedding_type=embedding_type) + vectors.extend(self._validated_vectors(batch, rows)) + requests += 1 + input_tokens += tokens + return EmbeddingResult( + vectors=tuple(vectors), + usage=InferenceUsage(requests=requests, input_tokens=input_tokens, output_tokens=None), + ) + + async def _embed_one(self, batch: Sequence[str], *, embedding_type: str) -> tuple[list[list[float]], int]: + payload = {"model": self._model, "texts": list(batch), "type": embedding_type} + response = await self._client.post(self._endpoint, json=payload, headers=self._headers) + # MiniMax returns HTTP 200 with a non-zero base_resp.status_code on error; + # only a real transport/HTTP failure reaches raise_for_status first. + response.raise_for_status() + try: + data = response.json() + except ValueError as error: + raise InvalidInferenceOutputError("embed", "provider response was not valid JSON") from error + if not isinstance(data, Mapping): + raise InvalidInferenceOutputError("embed", "provider response was not a JSON object") + base_resp = data.get("base_resp") + if isinstance(base_resp, Mapping): + status_code = base_resp.get("status_code") + if status_code is not None and status_code != 0: + status_msg = base_resp.get("status_msg") + detail = f"provider returned MiniMax status_code {status_code}" + if isinstance(status_msg, str) and status_msg.strip(): + detail = f"{detail}: {status_msg}" + raise InferenceUnavailableError("embed", detail) + vectors = data.get("vectors") + if not isinstance(vectors, list) or len(vectors) != len(batch): + raise InvalidInferenceOutputError("embed", "provider returned no vectors or the wrong vector count") + total_tokens = data.get("total_tokens") + tokens = int(total_tokens) if isinstance(total_tokens, int) else 0 + return vectors, tokens + + def _validated_vectors( + self, texts: Sequence[str], rows: Sequence[Sequence[float]] + ) -> tuple[tuple[float, ...], ...]: + out: list[tuple[float, ...]] = [] + for row in rows: + try: + out.append( + canonical_embedding( + tuple(row), + dimension=self.profile.dimension, + normalization=self.profile.normalization, + ) + ) + except (TypeError, ValueError) as error: + raise InvalidInferenceOutputError("embed", str(error)) from error + return tuple(out) + + +def _embedding_model_name(model: str | None) -> str: + """Strip a provider prefix such as ``openai:embo-01`` or ``minimax:embo-01``.""" + + if model is None: + return "" + return model.partition(":")[2] or model + + +def is_minimax_embedding(base_url: str | None, model: str | None) -> bool: + """Detect a MiniMax embedding endpoint by host or explicit model prefix.""" + + host = (urlparse(base_url or "").hostname or "").lower() + if any(host == domain or host.endswith(f".{domain}") for domain in ("minimaxi.com", "minimax.io")): + return True + provider_prefix, separator, _model_name = (model or "").partition(":") + return bool(separator) and provider_prefix.lower() == "minimax" + + +__all__ = ["MiniMaxEmbeddingModel", "is_minimax_embedding"] diff --git a/src/powercontext/builtin/inference/protocols.py b/src/powercontext/builtin/inference/protocols.py index 9c4ca682f4..4e5618aaa0 100644 --- a/src/powercontext/builtin/inference/protocols.py +++ b/src/powercontext/builtin/inference/protocols.py @@ -16,7 +16,7 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Protocol, TypeVar +from typing import TYPE_CHECKING, Protocol, TypeVar, runtime_checkable from powercontext.builtin.inference.models import EmbeddingResult, GenerationResult @@ -45,3 +45,21 @@ async def embed(self, texts: tuple[str, ...], /) -> EmbeddingResult: """Return exactly one ordered vector for each input text.""" ... + + +@runtime_checkable +class QueryEmbeddingModel(Protocol): + """Embed retrieval queries when a provider distinguishes queries from documents.""" + + async def embed_query(self, texts: tuple[str, ...], /) -> EmbeddingResult: + """Return exactly one ordered query vector for each input text.""" + + ... + + +async def embed_query(model: EmbeddingModel, texts: tuple[str, ...], /) -> EmbeddingResult: + """Embed retrieval queries, falling back to document embeddings for symmetric models.""" + + if isinstance(model, QueryEmbeddingModel): + return await model.embed_query(texts) + return await model.embed(texts) diff --git a/src/powercontext/builtin/inference/pydantic_ai.py b/src/powercontext/builtin/inference/pydantic_ai.py index d1653ccdb4..758a780d59 100644 --- a/src/powercontext/builtin/inference/pydantic_ai.py +++ b/src/powercontext/builtin/inference/pydantic_ai.py @@ -20,7 +20,7 @@ from collections.abc import Sequence from contextlib import nullcontext from copy import copy -from typing import Generic, Self, TypeVar, cast +from typing import Generic, Literal, Self, TypeVar, cast from pydantic import BaseModel, Field from typing_extensions import override @@ -245,11 +245,23 @@ def _without_instrumentation(self) -> Self: async def embed(self, texts: tuple[str, ...], /) -> EmbeddingResult: """Embed documents and validate order, count, dimension, and finite values.""" + return await self._embed(texts, input_type="document") + + async def embed_query(self, texts: tuple[str, ...], /) -> EmbeddingResult: + """Embed retrieval queries and validate order, count, dimension, and finite values.""" + + return await self._embed(texts, input_type="query") + + async def _embed(self, texts: tuple[str, ...], *, input_type: Literal["document", "query"]) -> EmbeddingResult: + """Embed one document or query batch through Pydantic AI.""" + if not texts: return EmbeddingResult(vectors=()) try: - result = await asyncio.wait_for(self._embed_batches(texts), timeout=self._limits.timeout_seconds) + result = await asyncio.wait_for( + self._embed_batches(texts, input_type=input_type), timeout=self._limits.timeout_seconds + ) except asyncio.CancelledError: raise except InvalidInferenceOutputError: @@ -266,14 +278,18 @@ async def embed(self, texts: tuple[str, ...], /) -> EmbeddingResult: return result - async def _embed_batches(self, texts: tuple[str, ...]) -> EmbeddingResult: + async def _embed_batches( + self, texts: tuple[str, ...], *, input_type: Literal["document", "query"] + ) -> EmbeddingResult: vectors: list[tuple[float, ...]] = [] requests = 0 input_tokens = 0 for start in range(0, len(texts), self._batch_size): batch = texts[start : start + self._batch_size] - result = await self._embedder.embed_documents(batch) - vectors.extend(self._validated_vectors(batch, result.inputs, result.input_type, result.embeddings)) + result = await self._embedder.embed(batch, input_type=input_type) + vectors.extend( + self._validated_vectors(batch, result.inputs, result.input_type, result.embeddings, input_type) + ) requests += 1 input_tokens += result.usage.input_tokens return EmbeddingResult( @@ -287,8 +303,9 @@ def _validated_vectors( inputs: Sequence[str], input_type: str, values: Sequence[Sequence[float]], + expected_input_type: Literal["document", "query"], ) -> tuple[tuple[float, ...], ...]: - if input_type != "document": + if input_type != expected_input_type: raise InvalidInferenceOutputError("embed", "provider returned the wrong input type") returned_inputs = tuple(inputs) rows = tuple(values) diff --git a/src/powercontext/builtin/inference/usage.py b/src/powercontext/builtin/inference/usage.py index 1d02963abf..d32e4ff9d0 100644 --- a/src/powercontext/builtin/inference/usage.py +++ b/src/powercontext/builtin/inference/usage.py @@ -23,7 +23,7 @@ from typing import Generic, TypeVar from powercontext.builtin.inference.models import EmbeddingResult, GenerationResult, InferenceUsage -from powercontext.builtin.inference.protocols import EmbeddingModel, StructuredGenerator +from powercontext.builtin.inference.protocols import EmbeddingModel, StructuredGenerator, embed_query from powercontext.builtin.statistics import ModelUsageOperation, ModelUsagePurpose InputT = TypeVar("InputT") @@ -101,5 +101,10 @@ async def embed(self, texts: tuple[str, ...], /) -> EmbeddingResult: await _report(ModelUsageOperation.EMBEDDING, result.usage) return result + async def embed_query(self, texts: tuple[str, ...], /) -> EmbeddingResult: + result = await embed_query(self._delegate, texts) + await _report(ModelUsageOperation.EMBEDDING, result.usage) + return result + __all__ = ["UsageReportingEmbeddingModel", "UsageReportingStructuredGenerator", "bind_usage_reporter"] diff --git a/src/powercontext/builtin/runtime/application.py b/src/powercontext/builtin/runtime/application.py index e431902c56..a1c552ac0b 100644 --- a/src/powercontext/builtin/runtime/application.py +++ b/src/powercontext/builtin/runtime/application.py @@ -122,8 +122,10 @@ from powercontext.builtin.evidence.resolver import AuthorizationContext, ScopedEvidenceAuthorizer from powercontext.builtin.inference import ( EmbeddingModel, + InferenceTimeoutError, InferenceUnavailableError, InvalidInferenceOutputError, + embed_query, ) from powercontext.builtin.inference.models import InferenceUsage from powercontext.builtin.inference.usage import bind_usage_reporter @@ -2626,10 +2628,10 @@ async def _search_with_embedding( return result.model_copy(update={"query_embedding": query_embedding, "embedding_calls": 0}), False try: async with asyncio.timeout(embedding_timeout_seconds): - embedded = await embedding.embed((request.query,)) + embedded = await embed_query(embedding, (request.query,)) if len(embedded.vectors) != 1: raise InvalidInferenceOutputError("embed", "provider returned the wrong vector count") - except (InferenceUnavailableError, TimeoutError) as error: + except (InferenceUnavailableError, InferenceTimeoutError, TimeoutError) as error: used_fallback = True log_safely( logger, @@ -2639,7 +2641,11 @@ async def _search_with_embedding( "event": "topic_memory.search.embedding_fallback", "outcome": "fallback", "mode": "fts", - "error_code": ("inference_timeout" if isinstance(error, TimeoutError) else "inference_unavailable"), + "error_code": ( + "inference_timeout" + if isinstance(error, (InferenceTimeoutError, TimeoutError)) + else "inference_unavailable" + ), "unit": "topic-memory", }, ) diff --git a/src/powercontext/builtin/runtime/composition.py b/src/powercontext/builtin/runtime/composition.py index 22e2408ae9..71c78b34a2 100644 --- a/src/powercontext/builtin/runtime/composition.py +++ b/src/powercontext/builtin/runtime/composition.py @@ -183,7 +183,7 @@ def __init__(self, issue: str) -> None: "Topic Memory processing requires child-reconstructible inference resources" ), "topic-memory-generation-budget": "Topic Memory generation budget is not executable", - "topic-memory-provider-budget": "Topic Memory workers require OpenAI/Anthropic SDK providers with transport retries disabled and bounded stateless model settings", + "topic-memory-provider-budget": "Topic Memory workers require supported providers with transport retries disabled and bounded stateless model settings", "topic-memory-generation": "Topic Memory processing requires a configured generation model", "topic-memory-database": "Topic Memory workers require file-backed SQLite; an in-memory database cannot be shared with spawned workers", "artifact-processing-worker-authorization-provider": ( @@ -1420,6 +1420,14 @@ async def _embedding_models( if settings.embedding_model is None: return None, None + from powercontext.builtin.inference.minimax import is_minimax_embedding + + if is_minimax_embedding( + str(settings.embedding_base_url) if settings.embedding_base_url is not None else None, + settings.embedding_model, + ): + return _minimax_embedding_models(settings, resources, instrumentation) + from pydantic_ai import Embedder from pydantic_ai.embeddings import EmbeddingSettings, infer_embedding_model @@ -1469,6 +1477,49 @@ def adapter(instrument: InstrumentationSettings | bool | None) -> EmbeddingModel return adapter(instrumentation), adapter(False) +def _minimax_embedding_models( + settings: InferenceConfig, + resources: AsyncExitStack, + instrumentation: InstrumentationSettings | None, +) -> tuple[EmbeddingModel | None, EmbeddingModel | None]: + """Build operational and readiness MiniMax embedding adapters. + + MiniMax is detected by host or model prefix and requires an explicit base URL + plus a header secret, both already supplied by the inference configuration. + """ + + from powercontext.builtin.artifacts.memory import EmbeddingProfile + from powercontext.builtin.inference.minimax import MiniMaxEmbeddingModel, _embedding_model_name + + if settings.embedding_model is None or settings.embedding_base_url is None: + raise BuiltinConfigurationError("inference-endpoint-provider") + profile = EmbeddingProfile( + profile_id=_required(settings.embedding_profile_id), + model=settings.embedding_model, + dimension=_required(settings.embedding_dimension), + distance="l2", + normalization=settings.embedding_normalization, + ) + headers = _resolve_headers(settings.embedding_headers) if settings.embedding_headers else None + model_name = _embedding_model_name(settings.embedding_model) + + def adapter(instrument: InstrumentationSettings | bool | None) -> EmbeddingModel: + model = MiniMaxEmbeddingModel( + base_url=str(settings.embedding_base_url), + model=model_name, + headers=headers, + profile=profile, + batch_size=settings.embedding_batch_size, + timeout_seconds=settings.embedding_timeout_seconds, + ) + resources.push_async_callback(model.aclose) + return model + + # MiniMax has no OpenTelemetry instrumentation hook; readiness uses the same + # instance kind as the operational adapter. + return adapter(instrumentation), adapter(False) + + def _embedding_readiness_probe(model: EmbeddingModel) -> ReadinessProbe: async def probe_embedding() -> None: await model.embed(("PowerContext readiness probe",)) diff --git a/src/powercontext/builtin/runtime/topic_memory_processing.py b/src/powercontext/builtin/runtime/topic_memory_processing.py index 819e443132..f9c4fd6c5f 100644 --- a/src/powercontext/builtin/runtime/topic_memory_processing.py +++ b/src/powercontext/builtin/runtime/topic_memory_processing.py @@ -88,6 +88,7 @@ StructuredGenerator, TokenEstimator, character_token_estimator, + embed_query, ) from powercontext.builtin.inference.usage import bind_usage_reporter from powercontext.builtin.persistence.cursors import SourceCursorRepository @@ -573,6 +574,15 @@ async def _embed(self, texts: tuple[str, ...]): ) return await self._embedding_model.embed(texts) + async def _embed_query(self, texts: tuple[str, ...]): + if self._embedding_model is None: + raise TopicMemoryGenerationError("embedding_unavailable") + # At most one provider request per text (the adapter may batch them). + await self._reserve( + requests=max(1, len(texts)), tokens=max(1, sum(self._stages.estimator.estimate(text) for text in texts)) + ) + return await embed_query(self._embedding_model, texts) + async def _read_window(self, assignment: TopicMemoryWindowAssignment) -> tuple[StoredSource, ...]: if assignment.source_through - assignment.source_after > MAX_TOPIC_MEMORY_WINDOW_SOURCES: raise TopicMemoryGenerationError("source_complexity_limit") @@ -837,7 +847,7 @@ async def _search( profile = None if self._embedding_model is not None: with self._usage(ModelUsagePurpose.TOPIC_MEMORY_RECALL, embedding=True): - embedded = await self._embed((query if semantic_query is None else semantic_query,)) + embedded = await self._embed_query((query if semantic_query is None else semantic_query,)) query_vector = embedded.vectors[0] profile = self._embedding_model.profile async with self._database.transaction() as connection: @@ -1496,12 +1506,18 @@ def validate_topic_memory_provider_settings(inference: InferenceConfig) -> None: "deepseek", "openrouter", } - for name, settings, allowed in ( - (inference.generation_model, inference.generation_model_settings, generation), - (inference.embedding_model, inference.embedding_model_settings, {"dimensions", "truncate"}), + embedding_providers = providers | {"minimax"} + for name, settings, allowed, provider_names in ( + (inference.generation_model, inference.generation_model_settings, generation, providers), + ( + inference.embedding_model, + inference.embedding_model_settings, + {"dimensions", "truncate"}, + embedding_providers, + ), ): # The built-in test model has no external I/O; retain hermetic workers. - if name is not None and name != "test" and name.split(":", 1)[0] not in providers: + if name is not None and name != "test" and name.split(":", 1)[0] not in provider_names: raise BuiltinConfigurationError("topic-memory-provider-budget") if set(settings) - allowed or any( value is not None diff --git a/src/powercontext/service/controller.py b/src/powercontext/service/controller.py index 0d6ceaefc8..3ae426ee6b 100644 --- a/src/powercontext/service/controller.py +++ b/src/powercontext/service/controller.py @@ -54,7 +54,7 @@ from powercontext.transport import is_loopback_host _PERSISTED_ENVIRONMENT_PREFIX = "POWERCONTEXT_SERVER_" -_START_TIMEOUT_SECONDS = 30.0 +_START_TIMEOUT_SECONDS = 60.0 class ServiceController: diff --git a/tests/builtin/artifacts/memory/test_service.py b/tests/builtin/artifacts/memory/test_service.py index 6f37a11754..c46aeeb469 100644 --- a/tests/builtin/artifacts/memory/test_service.py +++ b/tests/builtin/artifacts/memory/test_service.py @@ -15,9 +15,13 @@ from __future__ import annotations import asyncio +from typing import cast from powercontext.builtin.artifacts.memory import ( + EmbeddingProfile, Memory, + MemoryBackend, + MemoryCapabilities, MemoryCommit, MemoryContent, MemoryEntryInput, @@ -26,10 +30,11 @@ MemoryManifestEntry, MemoryProjection, MemoryRerankDecision, + MemorySearchChannels, MemoryService, ) from powercontext.builtin.artifacts.memory.canonical import entry_content_hash, memory_content_hash -from powercontext.builtin.inference import InferenceUsage +from powercontext.builtin.inference import EmbeddingResult, InferenceUsage from powercontext.builtin.persistence.memory import RelationalMemoryBackend from powercontext.builtin.persistence.sqlite import SQLiteConfig from powercontext.builtin.runtime import BuiltinConfig, open_builtin_contexts @@ -52,6 +57,67 @@ async def rerank(self, query, candidates, limit, /) -> MemoryRerankDecision: ) +QUERY_EMBEDDING_PROFILE = EmbeddingProfile(profile_id="query-v1", model="test:query", dimension=3) + + +class _QueryRecordingEmbedding: + def __init__(self) -> None: + self.document_texts: list[str] = [] + self.query_texts: list[str] = [] + self.profile = QUERY_EMBEDDING_PROFILE + + async def embed(self, texts: tuple[str, ...], /) -> EmbeddingResult: + self.document_texts.extend(texts) + return EmbeddingResult(vectors=tuple((0.0, 1.0, 0.0) for _ in texts)) + + async def embed_query(self, texts: tuple[str, ...], /) -> EmbeddingResult: + self.query_texts.extend(texts) + return EmbeddingResult(vectors=tuple((1.0, 0.0, 0.0) for _ in texts)) + + +class _QuerySearchBackend: + def __init__(self, memory: Memory) -> None: + self._memory = memory + self.query_vector: tuple[float, ...] | None = None + + async def capabilities(self) -> MemoryCapabilities: + return MemoryCapabilities(fts=True, vector=True, hybrid=True, embedding_profile=QUERY_EMBEDDING_PROFILE) + + async def get(self, _ref): + return self._memory + + async def latest(self, _artifact_id): + return self._memory + + async def vector_complete(self, _memories, _profile) -> bool: + return True + + async def search(self, request, /) -> MemorySearchChannels: + self.query_vector = request.query_vector + return MemorySearchChannels() + + +def test_memory_vector_search_uses_query_embedding_path() -> None: + memory = Memory( + artifact_id="memory", + revision=1, + content=MemoryContent(manifest=MemoryManifest(entries=())), + ) + + async def scenario() -> None: + embedding = _QueryRecordingEmbedding() + backend = _QuerySearchBackend(memory) + service = MemoryService(backend=cast(MemoryBackend, backend), embedding_model=embedding) + + await service.search("project", memories=(memory,), mode="vector") + + assert embedding.document_texts == [] + assert embedding.query_texts == ["project"] + assert backend.query_vector == (1.0, 0.0, 0.0) + + asyncio.run(scenario()) + + def test_memory_search_applies_injected_reranker_after_coarse_fusion() -> None: async def scenario() -> None: reranker = _SelectingReranker() diff --git a/tests/builtin/inference/test_minimax_embedding.py b/tests/builtin/inference/test_minimax_embedding.py new file mode 100644 index 0000000000..901e623d4a --- /dev/null +++ b/tests/builtin/inference/test_minimax_embedding.py @@ -0,0 +1,295 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests for the MiniMax embedding adapter (native /v1/embeddings contract).""" + +from __future__ import annotations + +import asyncio +import json +from typing import Any + +import httpx +import pytest + +from powercontext.builtin.artifacts.memory.models import EmbeddingProfile +from powercontext.builtin.inference.errors import ( + InferenceConfigurationError, + InferenceTimeoutError, + InferenceUnavailableError, + InvalidInferenceOutputError, +) +from powercontext.builtin.inference.minimax import ( + MiniMaxEmbeddingModel, + _embedding_model_name, + is_minimax_embedding, +) + +TEST_PROFILE = EmbeddingProfile( + profile_id="mm-v1", + model="openai:embo-01", + dimension=3, + distance="l2", + normalization="none", +) + + +def _model( + profile: EmbeddingProfile = TEST_PROFILE, + batch_size: int = 10, + headers=None, + http_client=None, + timeout_seconds: float = 30.0, +) -> MiniMaxEmbeddingModel: + return MiniMaxEmbeddingModel( + base_url="https://api.minimaxi.com/v1", + model="embo-01", + headers=headers, + profile=profile, + batch_size=batch_size, + timeout_seconds=timeout_seconds, + http_client=http_client, + ) + + +def _ok_response(vectors: list[list[float]], status_code: int = 0) -> httpx.Response: + return httpx.Response( + 200, + json={ + "vectors": vectors, + "total_tokens": len(vectors), + "base_resp": {"status_code": status_code, "status_msg": "success"}, + }, + ) + + +def test_embed_returns_validated_vectors_and_sends_native_shape() -> None: + captured: list[dict[str, Any]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured.append(json.loads(request.content)) + return _ok_response([[0.1, 0.2, 0.3], [0.4, 0.5, 0.6]]) + + async def scenario() -> None: + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + model = _model(http_client=client) + result = await model.embed(("alpha", "beta")) + assert result.vectors == ((0.1, 0.2, 0.3), (0.4, 0.5, 0.6)) + assert result.usage.requests == 1 + assert result.usage.input_tokens == 2 + + asyncio.run(scenario()) + assert captured[0] == {"model": "embo-01", "texts": ["alpha", "beta"], "type": "db"} + + +def test_embed_query_sends_query_type() -> None: + captured: list[dict[str, Any]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured.append(json.loads(request.content)) + return _ok_response([[0.1, 0.2, 0.3]]) + + async def scenario() -> None: + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + model = _model(http_client=client) + result = await model.embed_query(("alpha",)) + assert result.vectors == ((0.1, 0.2, 0.3),) + + asyncio.run(scenario()) + assert captured[0] == {"model": "embo-01", "texts": ["alpha"], "type": "query"} + + +def test_embed_batches_requests_without_split_size_mismatch() -> None: + calls: list[dict[str, Any]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content) + calls.append(body) + return _ok_response([[0.1, 0.2, 0.3] for _ in body["texts"]]) + + async def scenario() -> None: + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + model = _model(batch_size=2, http_client=client) + result = await model.embed(("a", "b", "c", "d", "e")) + assert result.usage.requests == 3 + + asyncio.run(scenario()) + assert [len(call["texts"]) for call in calls] == [2, 2, 1] + + +def test_embed_raises_unavailable_on_business_error() -> None: + def handler(request: httpx.Request) -> httpx.Response: + # MiniMax answers HTTP 200 with a non-zero base_resp.status_code. + return httpx.Response( + 200, json={"vectors": None, "base_resp": {"status_code": 2013, "status_msg": "missing texts"}} + ) + + async def scenario() -> None: + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + model = _model(http_client=client) + with pytest.raises(InferenceUnavailableError) as error: + await model.embed(("alpha",)) + assert "MiniMax status_code 2013" in str(error.value) + assert "missing texts" in str(error.value) + + asyncio.run(scenario()) + + +def test_embed_raises_unavailable_on_http_error() -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(401, json={"base_resp": {"status_code": 1001}}) + + async def scenario() -> None: + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + model = _model(http_client=client) + with pytest.raises(InferenceUnavailableError): + await model.embed(("alpha",)) + + asyncio.run(scenario()) + + +def test_embed_maps_http_timeout_to_inference_timeout() -> None: + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ReadTimeout("provider did not respond", request=request) # noqa: TRY003 + + async def scenario() -> None: + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + model = _model(http_client=client) + with pytest.raises(InferenceTimeoutError): + await model.embed(("alpha",)) + + asyncio.run(scenario()) + + +def test_default_http_client_uses_configured_timeout() -> None: + async def scenario() -> None: + model = _model(timeout_seconds=7.0) + try: + assert model._client.timeout.read == 7.0 + finally: + await model.aclose() + + asyncio.run(scenario()) + + +def test_embed_raises_invalid_output_on_wrong_vector_count() -> None: + def handler(request: httpx.Request) -> httpx.Response: + return _ok_response([[0.1, 0.2, 0.3]]) + + async def scenario() -> None: + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + model = _model(http_client=client) + with pytest.raises(InvalidInferenceOutputError): + await model.embed(("alpha", "beta")) + + asyncio.run(scenario()) + + +def test_embed_raises_invalid_output_on_malformed_json() -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, content=b"not-json") + + async def scenario() -> None: + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + model = _model(http_client=client) + with pytest.raises(InvalidInferenceOutputError): + await model.embed(("alpha",)) + + asyncio.run(scenario()) + + +def test_embed_does_not_classify_unexpected_adapter_errors_as_unavailable() -> None: + class AdapterBugError(RuntimeError): + pass + + class BrokenClient: + async def post(self, *_args, **_kwargs): + raise AdapterBugError + + async def scenario() -> None: + model = _model(http_client=BrokenClient()) + with pytest.raises(AdapterBugError): + await model.embed(("alpha",)) + + asyncio.run(scenario()) + + +def test_embed_raises_invalid_output_on_dimension_mismatch() -> None: + def handler(request: httpx.Request) -> httpx.Response: + return _ok_response([[0.1, 0.2, 0.3, 0.4]]) + + async def scenario() -> None: + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + model = _model(http_client=client) + with pytest.raises(InvalidInferenceOutputError): + await model.embed(("alpha",)) + + asyncio.run(scenario()) + + +def test_embed_empty_returns_no_vectors() -> None: + captured: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured.append(request) + return _ok_response([]) + + async def scenario() -> None: + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + model = _model(http_client=client) + result = await model.embed(()) + assert result.vectors == () + + asyncio.run(scenario()) + assert captured == [] + + +def test_embed_sends_configured_headers() -> None: + captured: list[dict[str, Any]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured.append(dict(request.headers)) + return _ok_response([[0.1, 0.2, 0.3]]) + + async def scenario() -> None: + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + model = _model(headers={"Authorization": "Bearer secret-key"}, http_client=client) + await model.embed(("alpha",)) + + asyncio.run(scenario()) + assert captured[0]["authorization"] == "Bearer secret-key" + + +def test_init_rejects_non_positive_dimension() -> None: + profile = EmbeddingProfile(profile_id="p", model="m", dimension=0, distance="l2", normalization="none") + with pytest.raises(InferenceConfigurationError): + MiniMaxEmbeddingModel(base_url="https://api.minimaxi.com/v1", model="embo-01", profile=profile) + + +def test_is_minimax_embedding_detects_host_and_model() -> None: + assert is_minimax_embedding("https://api.minimaxi.com/v1", "openai:embo-01") is True + assert is_minimax_embedding("https://api.minimax.io/v1", None) is True + assert is_minimax_embedding(None, "minimax:embo-01") is True + assert is_minimax_embedding("https://proxy.example/v1?upstream=api.minimaxi.com", "openai:embo-01") is False + assert is_minimax_embedding("https://api.openai.com/v1/minimax.io", "openai:embo-01") is False + assert is_minimax_embedding("https://api.openai.com/v1", "openai:my-minimax-proxy") is False + assert is_minimax_embedding("https://api.openai.com/v1", "openai:text-embedding-3-small") is False + assert is_minimax_embedding(None, "openai:embo-01") is False + + +def test_embedding_model_name_strips_provider_prefix() -> None: + assert _embedding_model_name("openai:embo-01") == "embo-01" + assert _embedding_model_name("minimax:embo-01") == "embo-01" + assert _embedding_model_name("embo-01") == "embo-01" + assert _embedding_model_name(None) == "" diff --git a/tests/builtin/inference/test_pydantic_ai.py b/tests/builtin/inference/test_pydantic_ai.py index aa64f77774..ad8dd76784 100644 --- a/tests/builtin/inference/test_pydantic_ai.py +++ b/tests/builtin/inference/test_pydantic_ai.py @@ -132,6 +132,7 @@ class RecordingEmbeddingModel(PydanticAIEmbeddingModelBase): def __init__(self) -> None: super().__init__() self.calls: list[tuple[str, ...]] = [] + self.input_types: list[EmbedInputType] = [] @property def model_name(self) -> str: @@ -150,6 +151,7 @@ async def embed( ) -> PydanticAIEmbeddingResult: prepared, _ = self.prepare_embed(inputs, settings) self.calls.append(tuple(prepared)) + self.input_types.append(input_type) return PydanticAIEmbeddingResult( embeddings=((1.0, 2.0, 3.0),) * len(prepared), inputs=prepared, @@ -465,6 +467,23 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_embedding_adapter_forwards_query_input_type() -> None: + async def scenario() -> None: + provider = RecordingEmbeddingModel() + model = PydanticAIEmbeddingModel( + embedder=Embedder(provider), + profile=TEST_PROFILE, + ) + + result = await model.embed_query(("find this",)) + + assert provider.calls == [("find this",)] + assert provider.input_types == ["query"] + assert result.vectors == ((1.0, 2.0, 3.0),) + + asyncio.run(scenario()) + + def test_embedding_adapter_enforces_unit_normalization_profile() -> None: async def scenario() -> None: model = PydanticAIEmbeddingModel( diff --git a/tests/builtin/runtime/test_composition_embedding.py b/tests/builtin/runtime/test_composition_embedding.py index ccaedbe154..863239e7fe 100644 --- a/tests/builtin/runtime/test_composition_embedding.py +++ b/tests/builtin/runtime/test_composition_embedding.py @@ -17,7 +17,7 @@ import asyncio from contextlib import AsyncExitStack from pathlib import Path -from typing import ClassVar +from typing import Any, ClassVar import pytest from pydantic_ai import Embedder @@ -186,3 +186,62 @@ async def scenario() -> None: ] asyncio.run(scenario()) + + +def test_embedding_models_use_minimax_adapter_for_minimax_host(monkeypatch) -> None: + import json + + import httpx + from pydantic import AnyHttpUrl, SecretStr + + from powercontext.builtin.inference.minimax import MiniMaxEmbeddingModel + + captured: list[dict[str, Any]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured.append(json.loads(request.content)) + return httpx.Response( + 200, + json={ + "vectors": [[0.1] * 1536], + "total_tokens": 1, + "base_resp": {"status_code": 0, "status_msg": "success"}, + }, + ) + + original_async_client = httpx.AsyncClient + + def make_client(*args, **kwargs) -> httpx.AsyncClient: + return original_async_client(transport=httpx.MockTransport(handler)) + + # MiniMaxEmbeddingModel builds its own httpx.AsyncClient; route it to the mock. + monkeypatch.setattr(httpx, "AsyncClient", make_client) + + async def scenario() -> None: + config = InferenceConfig( + embedding_model="openai:embo-01", + embedding_base_url=AnyHttpUrl("https://api.minimaxi.com/v1"), + embedding_headers={"Authorization": SecretStr("Bearer test")}, + embedding_profile_id="mm-1536-v1", + embedding_dimension=1536, + ) + async with AsyncExitStack() as resources: + operational, _readiness = await _embedding_models(config, resources, None) + assert operational is not None + assert isinstance(operational, MiniMaxEmbeddingModel) + # The embedding adapter owns an httpx client registered with the runtime + # stack, so it must be exercised before the stack closes it. + result = await operational.embed(("hello",)) + assert len(result.vectors) == 1 + assert len(result.vectors[0]) == 1536 + query_result = await operational.embed_query(("hello",)) + assert len(query_result.vectors) == 1 + assert len(query_result.vectors[0]) == 1536 + + asyncio.run(scenario()) + assert captured[0]["texts"] == ["hello"] + assert captured[0]["type"] == "db" + assert captured[0]["model"] == "embo-01" + assert captured[1]["texts"] == ["hello"] + assert captured[1]["type"] == "query" + assert captured[1]["model"] == "embo-01" diff --git a/tests/builtin/runtime/test_processing_composition.py b/tests/builtin/runtime/test_processing_composition.py index d005936d9e..3ec0e547bb 100644 --- a/tests/builtin/runtime/test_processing_composition.py +++ b/tests/builtin/runtime/test_processing_composition.py @@ -50,7 +50,7 @@ ) from powercontext.builtin.runtime.family_processing import FamilyWorkerSpec, run_family_worker from powercontext.builtin.runtime.processing_contracts import ArtifactProcessingWorkAssignment -from powercontext.builtin.runtime.processing_registry import canonical_processing_manifest +from powercontext.builtin.runtime.processing_registry import canonical_processing_manifest, processing_capabilities from powercontext.builtin.runtime.relational import RelationalContexts from powercontext.builtin.runtime.topic_memory_processing import TopicMemoryWorkerSpec, run_topic_memory_worker @@ -209,6 +209,21 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_processing_capabilities_accept_minimax_embedding_but_not_generation() -> None: + embedding = BuiltinConfig( + inference=InferenceConfig( + generation_model="test", + embedding_model="minimax:embo-01", + embedding_profile_id="mm-v1", + embedding_dimension=1536, + ) + ) + assert "topic-memory" in processing_capabilities(embedding) + + generation = BuiltinConfig(inference=InferenceConfig(generation_model="minimax:chat-model")) + assert "topic-memory" not in processing_capabilities(generation) + + @pytest.mark.parametrize("family", ["experience", "skill"]) def test_dream_injection_requires_a_reconstructible_worker(tmp_path: Path, family: str) -> None: async def scenario() -> None: diff --git a/tests/builtin/runtime/test_topic_memory_processing.py b/tests/builtin/runtime/test_topic_memory_processing.py index 7ce5fe2c86..ca9fd04a92 100644 --- a/tests/builtin/runtime/test_topic_memory_processing.py +++ b/tests/builtin/runtime/test_topic_memory_processing.py @@ -1540,6 +1540,7 @@ def test_probe_query_is_bounded_before_embedding_and_repository_io() -> None: async def scenario() -> None: manager, profile, sources, _ = await _repositories() observed_queries: list[str] = [] + embedded_documents: list[str] = [] embedded_queries: list[str] = [] class FakeTopics: @@ -1551,6 +1552,10 @@ class RecordingEmbedding: profile = None async def embed(self, texts: tuple[str, ...], /) -> EmbeddingResult: + embedded_documents.extend(texts) + return EmbeddingResult(vectors=tuple((1.0, 0.0) for _ in texts)) + + async def embed_query(self, texts: tuple[str, ...], /) -> EmbeddingResult: embedded_queries.extend(texts) return EmbeddingResult(vectors=tuple((1.0, 0.0) for _ in texts)) @@ -1572,6 +1577,7 @@ async def embed(self, texts: tuple[str, ...], /) -> EmbeddingResult: ) assert observed_queries != embedded_queries + assert embedded_documents == [] assert len(observed_queries[0]) <= MAX_TOPIC_MEMORY_QUERY_LENGTH assert len(embedded_queries[0]) <= MAX_TOPIC_MEMORY_QUERY_LENGTH assert embedded_queries[0].startswith("部署环境") diff --git a/tests/test_service.py b/tests/test_service.py index a6e8a97c7c..b9e510272a 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -372,6 +372,30 @@ def test_service_controller_installs_and_starts_one_native_registration(tmp_path assert adapter.events == ["write", "reload", "enable", "start:True"] +def test_service_controller_allows_slow_native_startup(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + adapter = FakeAdapter(tmp_path) + clock = 0.0 + + def monotonic() -> float: + return clock + + def sleep(delay: float) -> None: + nonlocal clock + clock += delay + + def probe(endpoint: str) -> ProbeResult: + if clock >= 45.0: + return ProbeResult(ProbeState.LIVE, f"{endpoint} status=ok") + return ProbeResult(ProbeState.UNREACHABLE, f"cannot reach {endpoint}") + + monkeypatch.setattr("powercontext.service.controller.time.monotonic", monotonic) + + status = ServiceController(adapter, probe=probe, sleep=sleep).install() + + assert status.ok + assert clock >= 45.0 + + @pytest.mark.skipif(sys.platform != "win32", reason="login auto-start opt-out is Windows-specific") def test_service_controller_can_install_without_login_autostart(tmp_path: Path) -> None: adapter = FakeAdapter(tmp_path)