fix(models): 校验 Embedding 响应与输入逐条对应

backend/test/unit/models/test_embed_alignment.py

@@ -0,0 +1,106 @@
+"""Embedding 响应必须与请求输入逐条对应，非法响应整批失败。"""
+
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+
+from yuxi.modules.models.embed import OtherEmbedding
+
+
+@pytest.fixture
+def client():
+    """使用固定的测试供应商配置。"""
+    return OtherEmbedding(model="test-model", base_url="http://test.local/embeddings", api_key="test-key", batch_size=2)
+
+
+@pytest.fixture
+def response_data(monkeypatch):
+    """在 HTTP 边界为同步和异步客户端提供相同响应。"""
+    data = []
+    response = httpx.Response(200, request=httpx.Request("POST", "http://test.local/embeddings"))
+    monkeypatch.setattr(response, "json", lambda: {"data": data})
+    monkeypatch.setattr("yuxi.modules.models.embed.requests.post", lambda *args, **kwargs: response)
+    monkeypatch.setattr("yuxi.modules.models.embed.httpx.AsyncClient.post", AsyncMock(return_value=response))
+    return data
+
+
+@pytest.mark.parametrize("method", ["encode", "aencode"])
+@pytest.mark.parametrize(
+    ("items", "error"),
+    [
+        ([{"embedding": [0.1]}], "returned 1 vectors for 2 inputs"),
+        ([{"embedding": [0.1]}, {"embedding": [0.2]}, {"embedding": [0.3]}], "returned 3 vectors for 2 inputs"),
+        ([{"index": 0, "embedding": None}, {"index": 1, "embedding": [0.2]}], "invalid item"),
+        ([{"embedding": None}, {"embedding": [0.2]}], "invalid item"),
+        ([{"index": 0, "embedding": None}, {"index": 0, "embedding": [0.2]}], "invalid item"),
+        ([{"index": 0, "embedding": [0.1]}, {"index": 0, "embedding": [0.2]}], "duplicate index"),
+        ([{"index": 0, "embedding": [0.1]}, {"embedding": [0.2]}], "mixes indexed and unindexed"),
+        ([{"index": 0, "embedding": [0.1]}, {}], "invalid item"),
+    ],
+    ids=["short", "extra", "indexed-null", "unindexed-null", "null-and-duplicate", "duplicate", "mixed", "missing-vector"],
+)
+async def test_direct_encode_rejects_invalid_response(client, response_data, method, items, error):
+    """直接编码也必须拒绝数量错误、空向量和无效索引，避免语义切分丢句。"""
+    response_data.extend(items)
+
+    with pytest.raises(ValueError, match=error):
+        if method == "encode":
+            client.encode(["a", "b"])
+        else:
+            await client.aencode(["a", "b"])
+
+
+@pytest.mark.parametrize(
+    ("data", "error"),
+    [({}, "data must be a list"), (None, "data must be a list"), ([None], "invalid item")],
+)
+def test_extract_embeddings_rejects_malformed_data(data, error):
+    """响应容器和条目必须符合供应商协议形态。"""
+    with pytest.raises(ValueError, match=error):
+        OtherEmbedding._extract_embeddings({"data": data})
+
+
+@pytest.mark.parametrize("index", [-1, 2, True, 0.5, "1"])
+def test_extract_embeddings_rejects_invalid_index(index):
+    """索引必须是响应范围内的整数，布尔值不能充当索引。"""
+    with pytest.raises(ValueError, match="out-of-range index"):
+        OtherEmbedding._extract_embeddings({"data": [{"index": 0, "embedding": [0.1]}, {"index": index, "embedding": [0.2]}]}, expected=2)
+
+
+@pytest.mark.parametrize("method", ["encode", "aencode"])
+@pytest.mark.parametrize(
+    ("message", "items", "expected"),
+    [
+        (["a", "b"], [{"index": 1, "embedding": [0.2]}, {"index": 0, "embedding": [0.1]}], [[0.1], [0.2]]),
+        (["a", "b"], [{"embedding": [0.1]}, {"embedding": [0.2]}], [[0.1], [0.2]]),
+        (["a", "b"], [{"index": None, "embedding": [0.1]}, {"embedding": [0.2]}], [[0.1], [0.2]]),
+        ("hello", [{"index": 0, "embedding": [0.1]}], [[0.1]]),
+        ([], [], []),
+    ],
+    ids=["reordered", "unindexed", "null-index", "string", "empty"],
+)
+async def test_direct_encode_preserves_alignment(client, response_data, method, message, items, expected):
+    """同步、异步入口保持归位结果及无索引、单条和空输入兼容。"""
+    response_data.extend(items)
+
+    if method == "encode":
+        actual = client.encode(message)
+    else:
+        actual = await client.aencode(message)
+
+    assert actual == expected
+
+
+@pytest.mark.parametrize("method", ["batch_encode", "abatch_encode"])
+async def test_batch_encode_rejects_partial_later_batch(client, monkeypatch, method):
+    """首批成功后，后续批次缺项仍须整批抛错。"""
+    batches = iter([[[0.1], [0.2]], []])
+    monkeypatch.setattr(client, "encode", lambda message: next(batches))
+    monkeypatch.setattr(client, "aencode", AsyncMock(side_effect=lambda message: next(batches)))
+
+    with pytest.raises(ValueError, match="returned 0 vectors for 1 inputs"):
+        if method == "batch_encode":
+            client.batch_encode(["a", "b", "c"])
+        else:
+            await client.abatch_encode(["a", "b", "c"])

backend/yuxi/modules/models/embed.py

@@ -48,6 +48,7 @@ def batch_encode(self, messages: list[str], batch_size: int | None = None) -> li
             group_msg = messages[i : i + batch_size]
             logger.info(f"Encoding [{i}/{len(messages)}] messages (bsz={batch_size})")
             response = self.encode(group_msg)
+            self._require_one_vector_per_input(response, len(group_msg))
             data.extend(response)
 
         return data
@@ -60,6 +61,7 @@ async def abatch_encode(self, messages: list[str], batch_size: int | None = None
             group_msg = messages[i : i + batch_size]
             logger.info(f"Async encoding [{i}/{len(messages)}] messages (bsz={batch_size})")
             res = await self.aencode(group_msg)
+            self._require_one_vector_per_input(res, len(group_msg))
             data.extend(res)
 
         return data
@@ -79,6 +81,44 @@ async def test_connection(self) -> tuple[bool, str]:
             logger.error(error_msg)
             return False, error_msg
 
+    @staticmethod
+    def _extract_embeddings(result: dict, expected: int | None = None) -> list[list[float]]:
+        """只接受逐条对应的向量；提供者带 index 时按 index 归位，否则保持响应顺序。"""
+        if not isinstance(result, dict) or "data" not in result:
+            raise ValueError(f"Embedding failed: Invalid response format {result}")
+
+        items = result["data"]
+        if not isinstance(items, list):
+            raise ValueError("Embedding response data must be a list")
+        if expected is not None:
+            BaseEmbeddingModel._require_one_vector_per_input(items, expected)
+
+        for item in items:
+            if not isinstance(item, dict) or item.get("embedding") is None:
+                raise ValueError("Embedding response contains an invalid item")
+
+        if not any(item.get("index") is not None for item in items):
+            return [item["embedding"] for item in items]
+
+        indexed = {}
+        for item in items:
+            index = item.get("index")
+            if index is None:
+                raise ValueError("Embedding response mixes indexed and unindexed items")
+            if type(index) is not int or not 0 <= index < len(items):
+                raise ValueError("Embedding response contains an out-of-range index")
+            if index in indexed:
+                raise ValueError("Embedding response contains a duplicate index")
+            indexed[index] = item["embedding"]
+
+        return [indexed[index] for index in range(len(items))]
+
+    @staticmethod
+    def _require_one_vector_per_input(embeddings: list, expected: int) -> None:
+        """校验响应或批次条数，拒绝部分结果。"""
+        if len(embeddings) != expected:
+            raise ValueError(f"Embedding response returned {len(embeddings)} vectors for {expected} inputs")
+
 
 class OtherEmbedding(BaseEmbeddingModel):
     def __init__(self, **kwargs) -> None:
@@ -99,7 +139,7 @@ def encode(self, message: list[str] | str) -> list[list[float]]:
             try:
                 response = requests.post(self.base_url, json=payload, headers=self.headers, timeout=60)
                 response.raise_for_status()
-                return self._extract_embeddings(response.json())
+                return self._extract_embeddings(response.json(), 1 if isinstance(message, str) else len(message))
             except requests.RequestException as e:
                 retry = self._prepare_retry(
                     message,
@@ -125,7 +165,7 @@ async def aencode(self, message: list[str] | str) -> list[list[float]]:
                 try:
                     response = await client.post(self.base_url, json=payload, headers=self.headers, timeout=60)
                     response.raise_for_status()
-                    return self._extract_embeddings(response.json())
+                    return self._extract_embeddings(response.json(), 1 if isinstance(message, str) else len(message))
                 except httpx.HTTPStatusError as e:
                     retry = self._prepare_retry(
                         message,
@@ -195,12 +235,6 @@ def _prepare_retry(
         )
         return next_retry_index, delay
 
-    @staticmethod
-    def _extract_embeddings(result: dict) -> list[list[float]]:
-        if not isinstance(result, dict) or "data" not in result:
-            raise ValueError(f"Embedding failed: Invalid response format {result}")
-        return [item["embedding"] for item in result["data"]]
-
     @staticmethod
     def _log_long_inputs(message: list[str] | str, threshold: int = 4000) -> None:
         """调试辅助：记录超过字符阈值的 embedding 输入位置与长度，不输出内容以免泄露用户数据。"""

docs/develop-guides/decisions/implemented/2026-11-03-embedding-response-alignment.md

@@ -0,0 +1,35 @@
+# Embedding 响应必须与输入逐条对应
+
+状态：implemented
+类型：bug-fix
+Owner：backend/yuxi/modules/models/embed.py
+
+## 问题
+
+向量库按位置关联文本与向量，语义切分也按位置关联句子与聚类标签。供应商返回部分结果或乱序结果时，仅提取 `data[*].embedding` 会造成错配；将 `null` 当作索引空槽再过滤，还会减少直接 `encode` / `aencode` 返回的向量数量，让语义切分静默丢句，并掩盖重复索引。
+
+## 决策
+
+### 实现方案
+
+[响应解析](https://github.com/xerrors/Yuxi/blob/main/backend/yuxi/modules/models/embed.py)拥有逐条对应的校验。同步、异步 HTTP 入口传入请求条数（字符串计一条），解析函数拒绝条数不符、缺少向量或向量为 `null` 的响应。供应商提供索引时，使用索引到向量的映射独立检查整数类型、范围、重复及混合缺失，再按索引返回；不使用向量内容判断槽位是否占用，也不通过过滤修补结果。
+
+无索引响应保持供应商顺序；所有索引为空时保留无索引兼容行为。批处理入口在累加前复用数量检查，防止子类返回部分批次。任何不满足约束的批次都抛出 `ValueError`，调用方拿不到部分结果。
+
+这是模型响应边界内的完整修复，不新增状态、依赖、配置或待裁决方案，因此直接记录为 implemented。HTTP 重试策略、持久化顺序和事务边界保持不变。
+
+## 替代方案
+
+- 填零向量或截断结果：隐藏缺项并继续造成错配，不采用。
+- 在每个调用方校验：无法统一处理供应商索引，也遗漏直接调用路径，不采用。
+- 用 `None` 槽位同时判断重复与过滤结果：混淆供应商数据与索引占用状态，不采用。
+
+## 后果
+
+畸形响应明确失败；合法的乱序索引按输入顺序归位，无索引响应继续按响应顺序处理。数量正确但向量内容错误、维度或模型质量不在本契约范围内；不会修复已经写入的历史数据。
+
+## 验证
+
+[回归测试](https://github.com/xerrors/Yuxi/blob/main/backend/test/unit/models/test_embed_alignment.py)在 HTTP 边界替换供应商响应，覆盖同步和异步直接调用的缺项、多项、空向量、空向量加重复索引、混合索引，以及正常归位、无索引、字符串和空输入。批处理测试验证首批成功后，后续缺项仍整批失败；索引测试覆盖负数、越界及非整数。
+
+运行命令：`cd backend && uv run --no-sync --group test pytest test/unit/models/test_embed_alignment.py -q`。实际执行结果随 PR 记录；真实供应商调用与数据存储链路不由这些单元测试证明。