Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions src/powercontext/builtin/artifacts/memory/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@
EmbeddingVector,
InferenceTimeoutError,
InferenceUnavailableError,
embed_query,
)
from powercontext.builtin.tags import TagFilter
from powercontext.errors import RevisionConflictError
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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")
Expand Down
9 changes: 8 additions & 1 deletion src/powercontext/builtin/inference/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__ = [
Expand All @@ -45,8 +50,10 @@
"InferenceUnavailableError",
"InferenceUsage",
"InvalidInferenceOutputError",
"QueryEmbeddingModel",
"StructuredGenerator",
"TokenEstimator",
"TokenEstimatorProfile",
"character_token_estimator",
"embed_query",
]
8 changes: 6 additions & 2 deletions src/powercontext/builtin/inference/errors.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
198 changes: 198 additions & 0 deletions src/powercontext/builtin/inference/minimax.py
Original file line number Diff line number Diff line change
@@ -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"]
20 changes: 19 additions & 1 deletion src/powercontext/builtin/inference/protocols.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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)
29 changes: 23 additions & 6 deletions src/powercontext/builtin/inference/pydantic_ai.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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(
Expand All @@ -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)
Expand Down
7 changes: 6 additions & 1 deletion src/powercontext/builtin/inference/usage.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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"]
Loading
Loading