feat(llm): 统一用户模型解析层 + 召回侧改用用户 embedding 模型

将分散在 splitter / markdown_parser / /llm 路由三处的「查配置→解密→建 client→能力校验」
逻辑收敛到 src/core/llm/user_model_resolver.py，消除行为漂移并把 DB 访问收口到一处。

- 新增 aresolve_user_model / build_provider_from_config 统一解析入口；三处调用点接入，
  各自保留领域异常以不破坏既有失败码映射
- 召回 dense query 编码改用发起用户的 EMBEDDING 模型（与写入侧同源），缺配置硬失败，
  新增错误码 RECALL_EMBEDDING_CONFIG_MISSING，绕过宽松降级
- 清理 ModelFactory 客户端缓存死代码；CacheSyncService 去除对 ModelFactory 的联动
- /llm 路由边界归一 user_id / config_id 为 int（非法值 422），修类型契约漂移
- llm_user_config 增生成列 default_marker + 唯一键 uq_user_default_per_capability，
  强制每个 (user_id, provider_type, capability) 至多一条默认且启用配置（migration 0012）；
  默认配置查询改 order_by(priority).limit(1) 容错，避免脏数据下 MultipleResultsFound 误判
- 同步 docs/api/error_codes.md、docs/api/schemas/mysql.md、scripts/db/init.sql

Author: jixua

docs/api/error_codes.md

@@ -146,13 +146,18 @@ CODE: 中文业务原因；底层详情
 
 | 场景 | 事件 | code |
 | --- | --- | --- |
+| 发起用户无默认 EMBEDDING 配置（dense 路无法编码 query） | `error` | `RECALL_EMBEDDING_CONFIG_MISSING` |
 | 全部召回路失败 / 严格模式失败 | `error` | `RECALL_ALL_SOURCES_FAILED` |
 | 召回执行超过 `RECALL_STREAM_TIMEOUT_MS` | `error` | `RECALL_TIMEOUT` |
 | 未预期内部异常 | `error` | `RECALL_INTERNAL_ERROR` |
 
 宽松模式下单路失败但仍有成功路时**不是错误**：正常返回 `recall_done`，失败路计入
 `failed_sources`。客户端（Java）断连不作为业务错误，Python 停止发送事件并取消召回任务。
 
+例外：dense 召回 query 编码按发起用户的 EMBEDDING 配置解析（与写入侧同源）。用户无默认
+EMBEDDING 配置属**必备前置缺失**，走硬失败（`RECALL_EMBEDDING_CONFIG_MISSING`）而非宽松降级——
+即便其余路可用也不返回部分结果，避免"读侧系统模型 / 写侧用户模型"向量空间不一致的误召回。
+
 ## 6. Chunk Status Values
 
 | Status | 含义 |

docs/api/schemas/mysql.md

@@ -90,13 +90,15 @@ ORM：[`UserLLMConfigDB`](../../src/models/db_models.py)
 | `max_retries` | INT | 最大重试次数，默认 3 |
 | `stream_enabled` | BOOLEAN | 是否支持流式输出 |
 | `capability` | VARCHAR(32) | `CHAT` / `EMBEDDING` / `RERANK` / `OCR`，默认 `CHAT` |
+| `default_marker` | INT，生成列 | `default+active` 时为 `1`，否则 `NULL`，仅用于唯一约束（应用层不写入） |
 | `extra_config` | JSON | 扩展配置 |
 | `created_at` / `updated_at` | DATETIME | 创建 / 更新时间 |
 
 索引：
 - `uk_user_provider_model(user_id, provider_id, model_name)`
 - `idx_user_active_default(user_id, is_active, is_default)`
 - `idx_user_provider_cap(user_id, provider_type, capability)`
+- `uq_user_default_per_capability(user_id, provider_type, capability, default_marker)` — 唯一键。借助生成列 `default_marker`（默认+启用时为 1，否则 NULL）与 MySQL「唯一索引中 NULL 不计重复」语义，保证每个 `(user_id, provider_type, capability)` 至多一条默认且启用的配置；非默认/停用配置不受限（迁移 0012）
 
 ### `llm_usage_log` — LLM 调用用量日志
 

migrations/versions/0012_20260604_add_user_config_default_unique.py

@@ -0,0 +1,61 @@
+"""enforce one default LLM config per (user, provider_type, capability)
+
+ORM 注释长期承诺「is_default 在 (user_id, provider_type, capability) 范围内唯一」，
+但 schema 只有普通 idx_user_provider_cap 索引，从未强制。一旦同一能力下出现两条
+is_default=1，默认配置查询（scalar_one_or_none）会抛 MultipleResultsFound，被上层
+误判为「读取失败(可重试)」。
+
+参照 0011 的软删判别列思路，新增生成列 default_marker（仅 default+active 时为 1，
+否则 NULL），与三元组组成唯一键。MySQL 唯一索引里 NULL 不计重复，因此：
+- 每个 (user_id, provider_type, capability) 至多一条 default+active 配置；
+- 非默认 / 停用配置（marker=NULL）数量不受限。
+
+注意：应用到已有数据前需确认无重复默认，否则建唯一键会因 Duplicate entry 失败。
+
+Revision ID: 0012
+Revises: 0011
+Create Date: 2026-06-04
+"""
+
+from __future__ import annotations
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision: str = "0012"
+down_revision: Union[str, None] = "0011"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "llm_user_config",
+        sa.Column(
+            "default_marker",
+            sa.Integer(),
+            sa.Computed(
+                "(CASE WHEN is_default = 1 AND is_active = 1 THEN 1 ELSE NULL END)",
+                persisted=True,
+            ),
+            nullable=True,
+            comment="默认判别生成列：default+active 时为 1，否则 NULL，仅用于唯一约束",
+        ),
+    )
+    op.create_unique_constraint(
+        "uq_user_default_per_capability",
+        "llm_user_config",
+        ["user_id", "provider_type", "capability", "default_marker"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint(
+        "uq_user_default_per_capability",
+        "llm_user_config",
+        type_="unique",
+    )
+    op.drop_column("llm_user_config", "default_marker")

scripts/db/init.sql

@@ -68,11 +68,13 @@ CREATE TABLE IF NOT EXISTS llm_user_config (
     max_retries         INT             DEFAULT 3 COMMENT '最大重试次数',
     stream_enabled      BOOLEAN         DEFAULT TRUE COMMENT '是否支持流式输出',
     capability          VARCHAR(32)     NOT NULL DEFAULT 'CHAT' COMMENT '专用能力标识：CHAT/EMBEDDING/RERANK/OCR',
+    default_marker      INT GENERATED ALWAYS AS (CASE WHEN is_default = 1 AND is_active = 1 THEN 1 ELSE NULL END) STORED COMMENT '默认判别生成列：default+active 时为 1，否则 NULL，仅用于唯一约束',
     extra_config        JSON            COMMENT '扩展配置',
     created_at          DATETIME        NOT NULL DEFAULT CURRENT_TIMESTAMP,
     updated_at          DATETIME        NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
 
     UNIQUE KEY uk_user_provider_model (user_id, provider_id, model_name),
+    UNIQUE KEY uq_user_default_per_capability (user_id, provider_type, capability, default_marker),
     INDEX idx_user_active_default (user_id, is_active, is_default),
     INDEX idx_user_provider_cap (user_id, provider_type, capability)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 AUTO_INCREMENT=10000 COMMENT '用户级 LLM 配置表';

src/api/internal_auth.py

@@ -30,6 +30,8 @@
 CODE_ALL_SOURCES_FAILED = "RECALL_ALL_SOURCES_FAILED"
 CODE_TIMEOUT = "RECALL_TIMEOUT"
 CODE_INTERNAL_ERROR = "RECALL_INTERNAL_ERROR"
+# 发起用户无默认 EMBEDDING 配置：dense 召回无法编码 query，整请求硬失败。
+CODE_EMBEDDING_CONFIG_MISSING = "RECALL_EMBEDDING_CONFIG_MISSING"
 
 
 class RecallApiError(Exception):

src/api/recall_pipeline_provider.py

@@ -43,8 +43,14 @@ def _build_sparse_retriever() -> Retriever:
 
 
 def _build_dense_retriever() -> Retriever:
+    # dense 召回 query 编码按发起用户的 EMBEDDING 配置解析（与写入侧 index_chunks 同源）：
+    # 注入 aresolve_user_chunk_embedding_pipeline，facade.search_dense_chunks 据 user_id 解析。
+    from src.core.splitter.factory import aresolve_user_chunk_embedding_pipeline
+
     return DenseRetriever(
-        backend=compose_vector_storage_facade(),
+        backend=compose_vector_storage_facade(
+            query_embedding_resolver=aresolve_user_chunk_embedding_pipeline,
+        ),
         score_threshold=settings.DENSE_RETRIEVAL_SCORE_THRESHOLD,
     )
 

src/api/routes/llm.py

@@ -8,22 +8,58 @@
 from pydantic import BaseModel, Field
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from src.core.llm.response import (
-    APIResponse,
-    GenerateResult,
-    EmbeddingResult,
-    RerankResult,
-    StreamChunk,
-    UsageInfo,
-)
-from src.services.config_reader_service import ConfigReaderService
-from src.core.llm.factory import ModelFactory
+from src.core.llm.response import APIResponse
+from src.core.llm.base_provider import BaseProvider
+from src.core.llm.exceptions import UserModelConfigMissingError
+from src.core.llm.user_model_resolver import aresolve_user_model
 from src.database import get_db
 
 router = APIRouter(prefix="/api/v1/llm", tags=["llm"])
 
-# 依赖注入
-model_factory = ModelFactory()
+
+def _coerce_int(value: str, field: str) -> int:
+    """把请求边界传入的 ID 字符串归一成 int，非法值 → 422。
+
+    ``user_id``（来自 ``X-User-Id`` Header）与 ``config_id``（来自请求体）在路由层是
+    字符串，但下游 resolver / ConfigReaderService / ``BigInteger`` 主键都按 int 契约。
+    在此显式转换并校验，避免把弱类型一路下沉到 SQL 靠驱动隐式转换。
+    """
+    try:
+        return int(value)
+    except (TypeError, ValueError) as exc:
+        raise HTTPException(status_code=422, detail=f"invalid {field}") from exc
+
+
+async def _resolve_provider(
+    db: AsyncSession,
+    user_id: str,
+    capability: str,
+    *,
+    config_id: Optional[str] = None,
+    override_model: Optional[str] = None,
+) -> BaseProvider:
+    """按用户解析指定能力的 Provider，未命中（含系统兜底）→ 404。
+
+    统一走 :func:`aresolve_user_model`（``/llm`` 路由保留系统兜底）：config_id 指定优先，
+    否则取用户该能力默认配置，仍无则系统环境兜底；都没有抛 ``UserModelConfigMissingError``
+    在此翻成 404，保持原有对外行为。``user_id`` / ``config_id`` 在边界归一成 int。
+    """
+    uid = _coerce_int(user_id, "X-User-Id")
+    cid = _coerce_int(config_id, "config_id") if config_id is not None else None
+    try:
+        resolved = await aresolve_user_model(
+            user_id=uid,
+            capability=capability,
+            config_id=cid,
+            allow_system_fallback=True,
+            override_model=override_model,
+            db=db,
+        )
+    except UserModelConfigMissingError as exc:
+        raise HTTPException(
+            status_code=404, detail=f"{capability} configuration not found"
+        ) from exc
+    return resolved.provider
 
 
 # ============ 请求模型 ============
@@ -81,32 +117,9 @@ async def generate_text(
         APIResponse[GenerateResult]
     """
     try:
-        config_service = ConfigReaderService(db)
-
-        # 获取用户配置
-        if request.config_id:
-            config = await config_service.get_user_config_by_id(x_user_id, request.config_id)
-        else:
-            config = await config_service.get_user_default_config_by_capability(x_user_id, "CHAT")
-
-        if not config:
-            config = config_service.get_system_fallback_config_by_capability("CHAT")
-
-        if not config:
-            raise HTTPException(status_code=404, detail="LLM CHAT configuration not found")
-
-        # 获取 Provider
-        provider_type = config.get("provider_type", "openai")
-        if config.get("is_system_fallback"):
-            api_key = config.get("api_key", "")
-        else:
-            api_key = await config_service.decrypt_api_key(config.get("api_key", ""))
-
-        client = model_factory.create_client(
-            provider_type=provider_type,
-            api_key=api_key,
-            api_base_url=config.get("custom_api_base_url"),
-            model_name=request.model or config.get("model_name"),
+        client = await _resolve_provider(
+            db, x_user_id, "CHAT",
+            config_id=request.config_id, override_model=request.model,
         )
 
         # 调用生成
@@ -147,30 +160,9 @@ async def generate_text_stream(
     from fastapi.responses import StreamingResponse
 
     try:
-        config_service = ConfigReaderService(db)
-
-        if request.config_id:
-            config = await config_service.get_user_config_by_id(x_user_id, request.config_id)
-        else:
-            config = await config_service.get_user_default_config_by_capability(x_user_id, "CHAT")
-
-        if not config:
-            config = config_service.get_system_fallback_config_by_capability("CHAT")
-
-        if not config:
-            raise HTTPException(status_code=404, detail="LLM CHAT configuration not found")
-
-        provider_type = config.get("provider_type", "openai")
-        if config.get("is_system_fallback"):
-            api_key = config.get("api_key", "")
-        else:
-            api_key = await config_service.decrypt_api_key(config.get("api_key", ""))
-
-        client = model_factory.create_client(
-            provider_type=provider_type,
-            api_key=api_key,
-            api_base_url=config.get("custom_api_base_url"),
-            model_name=request.model or config.get("model_name"),
+        client = await _resolve_provider(
+            db, x_user_id, "CHAT",
+            config_id=request.config_id, override_model=request.model,
         )
 
         async def event_generator():
@@ -206,30 +198,9 @@ async def embed_text(
         APIResponse[EmbeddingResult]
     """
     try:
-        config_service = ConfigReaderService(db)
-
-        if request.config_id:
-            config = await config_service.get_user_config_by_id(x_user_id, request.config_id)
-        else:
-            config = await config_service.get_user_default_config_by_capability(x_user_id, "EMBEDDING")
-
-        if not config:
-            config = config_service.get_system_fallback_config_by_capability("EMBEDDING")
-
-        if not config:
-            raise HTTPException(status_code=404, detail="Embedding configuration not found")
-
-        provider_type = config.get("provider_type", "openai")
-        if config.get("is_system_fallback"):
-            api_key = config.get("api_key", "")
-        else:
-            api_key = await config_service.decrypt_api_key(config.get("api_key", ""))
-
-        client = model_factory.create_client(
-            provider_type=provider_type,
-            api_key=api_key,
-            api_base_url=config.get("custom_api_base_url"),
-            model_name=request.model or config.get("model_name"),
+        client = await _resolve_provider(
+            db, x_user_id, "EMBEDDING",
+            config_id=request.config_id, override_model=request.model,
         )
 
         result = await client.embed(texts=request.input, model=request.model)
@@ -258,30 +229,9 @@ async def rerank_documents(
         APIResponse[RerankResult]
     """
     try:
-        config_service = ConfigReaderService(db)
-
-        if request.config_id:
-            config = await config_service.get_user_config_by_id(x_user_id, request.config_id)
-        else:
-            config = await config_service.get_user_default_config_by_capability(x_user_id, "RERANK")
-
-        if not config:
-            config = config_service.get_system_fallback_config_by_capability("RERANK")
-
-        if not config:
-            raise HTTPException(status_code=404, detail="Rerank configuration not found")
-
-        provider_type = config.get("provider_type", "openai")
-        if config.get("is_system_fallback"):
-            api_key = config.get("api_key", "")
-        else:
-            api_key = await config_service.decrypt_api_key(config.get("api_key", ""))
-
-        client = model_factory.create_client(
-            provider_type=provider_type,
-            api_key=api_key,
-            api_base_url=config.get("custom_api_base_url"),
-            model_name=request.model or config.get("model_name"),
+        client = await _resolve_provider(
+            db, x_user_id, "RERANK",
+            config_id=request.config_id, override_model=request.model,
         )
 
         result = await client.rerank(
@@ -315,30 +265,8 @@ async def extract_text_from_image(
         APIResponse[dict]
     """
     try:
-        config_service = ConfigReaderService(db)
-
-        if request.config_id:
-            config = await config_service.get_user_config_by_id(x_user_id, request.config_id)
-        else:
-            config = await config_service.get_user_default_config_by_capability(x_user_id, "OCR")
-
-        if not config:
-            config = config_service.get_system_fallback_config_by_capability("OCR")
-
-        if not config:
-            raise HTTPException(status_code=404, detail="OCR configuration not found")
-
-        provider_type = config.get("provider_type", "openai")
-        if config.get("is_system_fallback"):
-            api_key = config.get("api_key", "")
-        else:
-            api_key = await config_service.decrypt_api_key(config.get("api_key", ""))
-
-        client = model_factory.create_client(
-            provider_type=provider_type,
-            api_key=api_key,
-            api_base_url=config.get("custom_api_base_url"),
-            model_name=request.model or config.get("model_name"),
+        client = await _resolve_provider(
+            db, x_user_id, "OCR", config_id=request.config_id,
         )
 
         result = await client.extract_text(