feat(search): support multimodal hybrid queries and AML payloads

benchmark/aml/README.md

@@ -62,6 +62,58 @@ default; use `AML_RETRY_ATTEMPTS` and `AML_RETRY_DELAY_SECONDS` to adjust that
 behavior. `Token`, `Bearer`, and `X-Api-Key` authentication are accepted when
 `AML_API_KEY` is set.
 
+`/add` accepts either a content string (Textual/Coding) or an ordered array of
+`text` and `image_url` parts in `messages[].content` (Multimodal). Strings are
+forwarded as native `content`; arrays are forwarded as native Session `parts`,
+preserving part order, message roles, and timestamps. Images must be inline
+Base64 JPEG, PNG, or WebP Data URIs, with at most 10 MiB decoded per image and
+30 MiB decoded across one Add request. For example:
+
+```json
+{
+  "request_id": "example:add:1",
+  "user_id": "example:user",
+  "session_id": "example:session",
+  "messages": [{
+    "role": "user",
+    "content": [
+      {"type": "text", "text": "Remember this receipt."},
+      {"type": "image_url", "image_url": {"url": "data:image/png;base64,<image-bytes>"}}
+    ]
+  }]
+}
+```
+
+The example's image payload is a placeholder; supply actual Base64 image bytes.
+Image understanding and memory extraction use OpenViking's existing Session
+commit pipeline and require a configured vision-capable VLM. The adapter does
+not perform separate image captioning.
+
+`/search` accepts a query string or an array of the same `text` and `image_url`
+parts, including image-only and multiple-image queries. Text parts are joined
+with newlines and image URLs retain their relative order. The adapter forwards
+them to native `find` using `query` and `image_url`, without an extra VLM call.
+String queries keep their original request and response behavior. For example:
+
+```json
+{
+  "user_id": "example:user",
+  "query": [
+    {"type": "text", "text": "Which workshop was this receipt for?"},
+    {"type": "image_url", "image_url": {"url": "data:image/png;base64,<image-bytes>"}}
+  ],
+  "top_k": 5
+}
+```
+
+The same 10 MiB per-image limit applies, with a 30 MiB decoded total per Search
+request. Image queries require a multimodal dense embedding model compatible
+with the stored vectors. In hybrid mode, OV embeds the complete image/text
+input through dense and only the text through sparse; image-only queries skip
+sparse. Volcengine text-only queries retain their combined dense/sparse request.
+Keep the configured dense model and dimension consistent with the existing
+index. Search responses remain text memory evidence rather than raw images.
+
 The native `find` call uses quick retrieval without reranking, even when the
 OpenViking server has a reranker configured. The adapter returns each hit's
 `content` when present, otherwise its `abstract`, without an additional

benchmark/aml/server.py

@@ -5,14 +5,16 @@
 
 import argparse
 import asyncio
+import base64
+import binascii
 import hmac
 import math
 import os
 import time
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
 from datetime import datetime, timezone
-from typing import Any, AsyncIterator, Awaitable, Callable, Literal, TypeVar
+from typing import Annotated, Any, AsyncIterator, Awaitable, Callable, Literal, TypeVar
 
 import httpx
 import uvicorn
@@ -24,6 +26,9 @@
 DEFAULT_PORT = 8088
 DEFAULT_OPENVIKING_URL = "http://127.0.0.1:1933"
 MEMORY_ROOT_URI = "viking://~/memories"
+MAX_IMAGE_BYTES = 10 * 1024 * 1024
+MAX_ADD_IMAGE_BYTES = 30 * 1024 * 1024
+MAX_QUERY_IMAGE_BYTES = 30 * 1024 * 1024
 RETRYABLE_OPENVIKING_CODES = {
     "ABORTED",
     "DEADLINE_EXCEEDED",
@@ -103,14 +108,72 @@ def _not_blank(value: str) -> str:
     return value
 
 
+class AMLTextPart(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    type: Literal["text"]
+    text: str = Field(min_length=1)
+
+    _text_not_blank = field_validator("text")(_not_blank)
+
+
+class AMLImageURL(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    url: str
+    detail: Literal["auto", "low", "high"] | None = None
+
+    @field_validator("url")
+    @classmethod
+    def validate_inline_image(cls, value: str) -> str:
+        header, _, encoded = value.partition(",")
+        if (
+            header
+            not in {
+                "data:image/jpeg;base64",
+                "data:image/png;base64",
+                "data:image/webp;base64",
+            }
+            or not encoded
+        ):
+            raise ValueError("image_url must be an inline Base64 JPEG, PNG, or WebP Data URI")
+        if len(encoded) > 4 * ((MAX_IMAGE_BYTES + 2) // 3):
+            raise ValueError("image exceeds the 10 MiB decoded size limit")
+        try:
+            size = len(base64.b64decode(encoded, validate=True))
+        except (binascii.Error, ValueError) as exc:
+            raise ValueError("image_url contains invalid Base64 data") from exc
+        if size > MAX_IMAGE_BYTES:
+            raise ValueError("image exceeds the 10 MiB decoded size limit")
+        return value
+
+    @property
+    def decoded_size(self) -> int:
+        # Base64 was validated above; count bytes without decoding images again.
+        return len(self.url.partition(",")[2].rstrip("=")) * 3 // 4
+
+
+class AMLImagePart(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    type: Literal["image_url"]
+    image_url: AMLImageURL
+
+
+AMLContentPart = Annotated[AMLTextPart | AMLImagePart, Field(discriminator="type")]
+
+
 class AMLMessage(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
     role: Literal["user", "assistant"]
-    content: str = Field(min_length=1)
+    content: str | list[AMLContentPart] = Field(min_length=1)
     timestamp: int | None = Field(default=None, ge=0, strict=True)
 
-    _content_not_blank = field_validator("content")(_not_blank)
+    @field_validator("content")
+    @classmethod
+    def content_not_blank(cls, value: str | list[AMLContentPart]) -> str | list[AMLContentPart]:
+        return _not_blank(value) if isinstance(value, str) else value
 
     @field_validator("timestamp")
     @classmethod
@@ -137,18 +200,41 @@ class AMLAddRequest(BaseModel):
     _user_id_not_blank = field_validator("user_id")(_not_blank)
     _session_id_not_blank = field_validator("session_id")(_not_blank)
 
+    @field_validator("messages")
+    @classmethod
+    def validate_image_budget(cls, messages: list[AMLMessage]) -> list[AMLMessage]:
+        size = sum(
+            part.image_url.decoded_size
+            for message in messages
+            if isinstance(message.content, list)
+            for part in message.content
+            if isinstance(part, AMLImagePart)
+        )
+        if size > MAX_ADD_IMAGE_BYTES:
+            raise ValueError("images exceed the 30 MiB decoded size limit per Add request")
+        return messages
+
 
 class AMLSearchRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
-    query: str = Field(min_length=1)
+    query: str | list[AMLContentPart] = Field(min_length=1)
     options: list[str] | None = None
     user_id: str = Field(min_length=1)
     top_k: int = Field(ge=1, strict=True)
 
-    _query_not_blank = field_validator("query")(_not_blank)
     _user_id_not_blank = field_validator("user_id")(_not_blank)
 
+    @field_validator("query")
+    @classmethod
+    def validate_query(cls, value: str | list[AMLContentPart]) -> str | list[AMLContentPart]:
+        if isinstance(value, str):
+            return _not_blank(value)
+        size = sum(part.image_url.decoded_size for part in value if isinstance(part, AMLImagePart))
+        if size > MAX_QUERY_IMAGE_BYTES:
+            raise ValueError("images exceed the 30 MiB decoded size limit per Search request")
+        return value
+
 
 class OpenVikingBackendError(RuntimeError):
     pass
@@ -246,14 +332,17 @@ async def add_and_commit(
                     raise OpenVikingBackendError("OpenViking commit timed out")
                 await asyncio.sleep(min(self.settings.task_poll_seconds, remaining))
 
-    async def find(self, *, user_id: str, query: str, limit: int) -> list[dict[str, Any]]:
+    async def find(
+        self, *, user_id: str, query: str, limit: int, image: str | list[str] | None = None
+    ) -> list[dict[str, Any]]:
         async with self._client(user_id) as client:
             result = await self._retry(
                 "find",
                 lambda: client.find(
                     query=query,
                     target_uri=MEMORY_ROOT_URI,
                     limit=limit,
+                    **({"image": image} if image else {}),
                     options={
                         "context_type": ["memory"],
                         "level": 2,
@@ -298,7 +387,11 @@ def authorize(self, request: Request) -> None:
     async def add(self, request: AMLAddRequest) -> dict[str, Any]:
         messages: list[dict[str, Any]] = []
         for message in request.messages:
-            item: dict[str, Any] = {"role": message.role, "content": message.content}
+            item: dict[str, Any] = {"role": message.role}
+            if isinstance(message.content, str):
+                item["content"] = message.content
+            else:
+                item["parts"] = [part.model_dump(exclude_none=True) for part in message.content]
             if message.timestamp is not None:
                 item["created_at"] = _format_timestamp(message.timestamp)
             messages.append(item)
@@ -321,6 +414,10 @@ async def add(self, request: AMLAddRequest) -> dict[str, Any]:
 
     async def search(self, request: AMLSearchRequest) -> dict[str, list[dict[str, Any]]]:
         query = request.query
+        images = []
+        if isinstance(query, list):
+            images = [part.image_url.url for part in query if isinstance(part, AMLImagePart)]
+            query = "\n".join(part.text for part in query if isinstance(part, AMLTextPart))
         if request.options:
             options = "\n".join(
                 f"{index}. {option}" for index, option in enumerate(request.options, start=1)
@@ -331,6 +428,7 @@ async def search(self, request: AMLSearchRequest) -> dict[str, list[dict[str, An
                 user_id=request.user_id,
                 query=query,
                 limit=request.top_k,
+                **({"image": images[0] if len(images) == 1 else images} if images else {}),
             )
         except OpenVikingBackendError as exc:
             raise HTTPException(

docs/en/api/06-retrieval.md

@@ -54,7 +54,7 @@ The `find()` method runs one global vector similarity search in QUICK mode for s
 | Parameter | Type | Required | Default | Description |
 |-----------|------|----------|---------|-------------|
 | query | str | No | "" | Search query string. May be omitted for an image query or a filter-only lookup |
-| image_url | str | No | None | Image query as a `data:image/...;base64,...`, `http(s)://`, or `viking://` URI. Requires a multimodal embedding model |
+| image_url | str or list[str] | No | None | One image URI or an ordered list of image URIs, each as a `data:image/...;base64,...`, `http(s)://`, or `viking://` URI. Requires a multimodal embedding model |
 | target_uri | str \| List[str] | No | "" | Limit search to specific URI prefix |
 | events_time_decay_protection | str \| null | No | null | Omit or pass `null` to disable decay. Pass `"0"` to decay immediately, or a duration such as `"7d"` to preserve the original score during that period and decay afterward. Supports non-negative integer `Xm`/`Xh`/`Xd` |
 | context_type | str \| List[str] | No | None | Limit results to one or more `ContextType` values: `memory`, `resource`, or `skill` |
@@ -394,7 +394,7 @@ Semantic retrieval with optional session-aware query planning.
 | Parameter | Type | Required | Default | Description |
 |-----------|------|----------|---------|-------------|
 | query | str | No | "" | Search query string. Required unless `image_url` is provided |
-| image_url | str | No | None | Image query as a `data:image/...;base64,...`, `http(s)://`, or `viking://` URI. Requires a multimodal embedding model |
+| image_url | str or list[str] | No | None | One image URI or an ordered list of image URIs, each as a `data:image/...;base64,...`, `http(s)://`, or `viking://` URI. Requires a multimodal embedding model |
 | search_type | `"semantic"` \| `"keywords"` | No | `"semantic"` | Retrieval type. `semantic` uses embeddings and vector recall; `keywords` uses BM25 keyword recall |
 | mode | `"list"` \| `"context"` | No | `"list"` | Return ranked hits or assembled context. CLI `--mode` maps to this field; context-specific parameters are described below |
 | target_uri | str \| List[str] | No | "" | Limit search to specific URI prefix |

docs/en/guides/01-configuration.md

@@ -388,6 +388,8 @@ When the embedding provider experiences consecutive transient failures (e.g. `42
 
 With `input: "multimodal"`, OpenViking can embed text, images (PNG, JPG, etc.), and mixed content. Image-to-image search requires this mode; text-only embedding models continue to index image summaries but cannot accept image queries.
 
+Hybrid image support depends on the dense channel. The complete mixed input goes to dense, while sparse receives only text; image-only inputs skip sparse. Volcengine hybrid keeps one combined request for text-only input and uses separate multimodal dense and text-only sparse requests for mixed input. Keep the dense model and dimension compatible with the vectors already stored in the index.
+
 **Supported providers:**
 - `openai`: OpenAI Embedding API
 - `azure`: Azure OpenAI Embedding API

docs/zh/api/06-retrieval.md

@@ -54,7 +54,7 @@ OpenViking 提供多种检索方法，包括简单的向量相似度搜索、关
 | 参数 | 类型 | 必填 | 默认值 | 说明 |
 |------|------|------|--------|------|
 | query | str | 否 | "" | 查询文本；图片查询或仅按过滤条件查询时可省略 |
-| image_url | str | 否 | None | 图片查询，支持 `data:image/...;base64,...`、`http(s)://` 或 `viking://` URI；需要 multimodal embedding 模型 |
+| image_url | str 或 list[str] | 否 | None | 单个图片 URI 或有序图片 URI 数组，每项支持 `data:image/...;base64,...`、`http(s)://` 或 `viking://` URI；需要 multimodal embedding 模型 |
 | target_uri | str \| List[str] | 否 | "" | 限制搜索范围到指定的 URI 前缀 |
 | events_time_decay_protection | str \| null | 否 | null | 不传或传 `null` 关闭衰减；传 `"0"` 立即衰减；传 `"7d"` 等时长则在保护期内保持原分，之后衰减。支持非负整数 `Xm`/`Xh`/`Xd` |
 | context_type | str \| List[str] | 否 | None | 限定一个或多个 `ContextType` 取值：`memory`、`resource` 或 `skill` |
@@ -394,7 +394,7 @@ ov find "红色海报风格" --image ./poster.png --uri "viking://resources/imag
 | 参数 | 类型 | 必填 | 默认值 | 说明 |
 |------|------|------|--------|------|
 | query | str | 否 | "" | 搜索查询字符串；未提供 `image_url` 时必填 |
-| image_url | str | 否 | None | 图片查询，支持 `data:image/...;base64,...`、`http(s)://` 或 `viking://` URI；需要 multimodal embedding 模型 |
+| image_url | str 或 list[str] | 否 | None | 单个图片 URI 或有序图片 URI 数组，每项支持 `data:image/...;base64,...`、`http(s)://` 或 `viking://` URI；需要 multimodal embedding 模型 |
 | search_type | `"semantic"` \| `"keywords"` | 否 | `"semantic"` | 检索类型。`semantic` 使用 Embedding 和向量召回；`keywords` 使用 BM25 关键词召回 |
 | mode | `"list"` \| `"context"` | 否 | `"list"` | 返回排序命中列表或组装好的上下文。CLI `--mode` 会映射到这个字段；context 专用参数见下文 |
 | target_uri | str \| List[str] | 否 | "" | 限制搜索范围到指定的 URI 前缀 |

docs/zh/guides/01-configuration.md

@@ -385,6 +385,8 @@ OpenAI 已于 2026 年 8 月 31 日[停止在 ChatGPT 登录的 Codex 中提供
 
 使用 `input: "multimodal"` 时，OpenViking 可以嵌入文本、图片（PNG、JPG 等）和混合内容。以图搜图需要该模式；纯文本 embedding 模型仍会索引图片 summary，但不能接收图片查询。
 
+Hybrid 是否支持图片取决于 dense 通道。图文完整输入进入 dense，sparse 只接收文字；纯图输入跳过 sparse。Volcengine hybrid 对纯文本保留一次合并请求，对图文分别请求多模态 dense 和文本 sparse。dense 模型和维度应与索引中已有的向量保持兼容。
+
 **支持的 provider:**
 - `openai`: OpenAI Embedding API
 - `azure`: Azure OpenAI Embedding API

openviking/models/embedder/base.py

@@ -656,28 +656,38 @@ def __init__(self, dense_embedder: DenseEmbedderBase, sparse_embedder: SparseEmb
 
     @property
     def supports_multimodal(self) -> bool:
-        """Supports multimodal input only if both sub-embedders do."""
-        return self.dense_embedder.supports_multimodal and self.sparse_embedder.supports_multimodal
+        """Images are handled by the dense channel; sparse consumes only text."""
+        return self.dense_embedder.supports_multimodal
+
+    def _sparse_input(self, content: "EmbeddingInput") -> Optional[str]:
+        text = extract_text_from_content(content)
+        if isinstance(content, list) and not text.strip():
+            return None
+        return self.sparse_embedder.prepare_embedding_input(text)
 
     def embed(self, content: "EmbeddingInput", is_query: bool = False) -> EmbedResult:
         """Combine results from both embedders"""
         dense_input = self.dense_embedder.prepare_embedding_input(content)
-        sparse_input = self.sparse_embedder.prepare_embedding_input(content)
+        sparse_input = self._sparse_input(content)
         dense_res = self.dense_embedder.embed(dense_input, is_query=is_query)
-        sparse_res = self.sparse_embedder.embed(sparse_input, is_query=is_query)
+        sparse_res = (
+            self.sparse_embedder.embed(sparse_input, is_query=is_query)
+            if sparse_input is not None
+            else None
+        )
 
         return EmbedResult(
-            dense_vector=dense_res.dense_vector, sparse_vector=sparse_res.sparse_vector
+            dense_vector=dense_res.dense_vector,
+            sparse_vector=sparse_res.sparse_vector if sparse_res is not None else None,
         )
 
     async def embed_async(self, content: "EmbeddingInput", is_query: bool = False) -> EmbedResult:
         dense_input = self.dense_embedder.prepare_embedding_input(content)
-        sparse_input = self.sparse_embedder.prepare_embedding_input(content)
-        calls = asyncio.gather(
-            self.dense_embedder.embed_async(dense_input, is_query=is_query),
-            self.sparse_embedder.embed_async(sparse_input, is_query=is_query),
-            return_exceptions=True,
-        )
+        sparse_input = self._sparse_input(content)
+        providers = [self.dense_embedder.embed_async(dense_input, is_query=is_query)]
+        if sparse_input is not None:
+            providers.append(self.sparse_embedder.embed_async(sparse_input, is_query=is_query))
+        calls = asyncio.gather(*providers, return_exceptions=True)
         # Keep the parent borrow alive until every child has stopped. Shielding
         # also lets cancellation wait for providers with asynchronous cleanup.
         from openviking.service.task_tracker_concurrency import run_to_completion
@@ -686,9 +696,11 @@ async def embed_async(self, content: "EmbeddingInput", is_query: bool = False) -
         for result in results:
             if isinstance(result, BaseException):
                 raise result
-        dense_res, sparse_res = results
+        dense_res = results[0]
+        sparse_res = results[1] if sparse_input is not None else None
         return EmbedResult(
-            dense_vector=dense_res.dense_vector, sparse_vector=sparse_res.sparse_vector
+            dense_vector=dense_res.dense_vector,
+            sparse_vector=sparse_res.sparse_vector if sparse_res is not None else None,
         )
 
     def get_dimension(self) -> int: