feat: add production embedding pgvector retrieval

project-docs/V2-ai-platform-embedding-pgvector-2026-06-22.md

@@ -0,0 +1,67 @@
+# V2 AI Platform Embedding + pgvector 闭环记录
+
+日期：2026-06-22
+
+## 本轮目标
+
+把知识库检索从 `local-hash-v1` 的可验证 fallback，升级为可接入真实 embedding provider 的生产级向量检索路径，并保留本地 fallback，保证测试、离线开发和临时 provider 故障时仍可运行。
+
+## 已落地内容
+
+1. 新增 embedding provider 抽象。
+   - `local-hash`：默认 fallback，固定 64 维，便于测试和离线验证。
+   - `openai-compatible`：按 `/embeddings` 接口调用真实模型，支持 OpenAI、DeepSeek 兼容网关、通义兼容网关或私有兼容服务。
+
+2. 新增热配置项。
+   - `AI_PLATFORM_EMBEDDING_PROVIDER`
+   - `AI_PLATFORM_EMBEDDING_MODEL`
+   - `AI_PLATFORM_EMBEDDING_API_KEY`
+   - `AI_PLATFORM_EMBEDDING_BASE_URL`
+   - `AI_PLATFORM_EMBEDDING_REQUEST_TIMEOUT_SECONDS`
+   - `AI_PLATFORM_EMBEDDING_FALLBACK_ENABLED`
+   - `AI_PLATFORM_VECTOR_SEARCH_BACKEND`
+
+3. 新增数据库迁移。
+   - `20260622_0004_add_embedding_provider_and_pgvector.py`
+   - `knowledge_chunks.embedding_provider`
+   - PostgreSQL 环境自动 `CREATE EXTENSION IF NOT EXISTS vector`
+   - PostgreSQL 环境新增 `knowledge_chunks.embedding vector`
+
+4. Repository 检索路径升级。
+   - 新增知识 chunk 时写入 provider/model/dimension 元数据。
+   - PostgreSQL + pgvector 可用时优先走 `embedding <=> query_vector`。
+   - 非 PostgreSQL、未启用 pgvector、pgvector 查询失败时回退 JSON embedding 排序。
+   - 关键词、向量、混合分数继续返回，法律问答审计仍记录 chunk ids、来源、分数。
+
+5. 契约同步。
+   - OpenAPI `KnowledgeChunk` 新增 `embedding_provider`。
+   - Python SDK `KnowledgeChunk` 新增 `embedding_provider`。
+
+## 生产配置建议
+
+```powershell
+$env:AI_PLATFORM_REPOSITORY_BACKEND="sqlalchemy"
+$env:AI_PLATFORM_DATABASE_URL="postgresql+psycopg://ai:ai@localhost:5432/yuqei_ai"
+$env:AI_PLATFORM_EMBEDDING_PROVIDER="openai-compatible"
+$env:AI_PLATFORM_EMBEDDING_MODEL="text-embedding-3-small"
+$env:AI_PLATFORM_EMBEDDING_API_KEY="真实 embedding key"
+$env:AI_PLATFORM_EMBEDDING_BASE_URL="https://api.openai.com/v1"
+$env:AI_PLATFORM_EMBEDDING_FALLBACK_ENABLED="true"
+$env:AI_PLATFORM_VECTOR_SEARCH_BACKEND="pgvector"
+python -m alembic upgrade head
+```
+
+## 运维注意
+
+- pgvector 是数据库扩展，需要 PostgreSQL 实例支持 `CREATE EXTENSION vector`。
+- 迁移只新增列，不会自动把历史 JSON embedding 转成真实向量；启用真实 embedding 后需要重导入或重建索引。
+- `local-hash-v1` 继续作为 fallback，不作为最终生产语义向量模型。
+- `AI_PLATFORM_VECTOR_SEARCH_BACKEND=json` 可强制关闭 pgvector，用于排障。
+- `AI_PLATFORM_EMBEDDING_FALLBACK_ENABLED=false` 会在 embedding 参数缺失或 provider 失败时返回空向量，更适合严格生产验收；默认 `true` 更适合平滑上线。
+
+## 验收口径
+
+- provider 成功时：chunk 记录真实 `embedding_provider=openai-compatible`、真实模型名和真实维度。
+- provider 不可用且允许 fallback 时：chunk 仍可写入，检索仍可用，但 provider 显示 `local-hash`。
+- pgvector 可用时：优先使用数据库向量距离排序。
+- pgvector 不可用时：回退 JSON embedding，不影响法律问答主链路。

yuqei-ai-platform/packages/ai-openapi/openapi.yaml

@@ -847,6 +847,7 @@ components:
         - content
         - chunk_index
         - locator
+        - embedding_provider
         - created_at
       properties:
         id:
@@ -870,6 +871,9 @@ components:
           type: integer
         locator:
           type: string
+        embedding_provider:
+          type: string
+          default: local-hash
         embedding_model:
           type: string
           default: local-hash-v1

yuqei-ai-platform/services/ai-platform-api/README.md

@@ -34,6 +34,28 @@ $env:AI_PLATFORM_PROVIDER_BASE_URL="https://api.deepseek.com/v1"
 $env:AI_PLATFORM_PROVIDER_FALLBACK_ENABLED="true"
 ```
 
+Embedding defaults to deterministic `local-hash-v1` so tests and offline
+development remain reproducible. For production-grade RAG, enable an
+OpenAI-compatible embedding provider and PostgreSQL pgvector retrieval:
+
+```powershell
+$env:AI_PLATFORM_REPOSITORY_BACKEND="sqlalchemy"
+$env:AI_PLATFORM_DATABASE_URL="postgresql+psycopg://ai:ai@localhost:5432/yuqei_ai"
+$env:AI_PLATFORM_EMBEDDING_PROVIDER="openai-compatible"
+$env:AI_PLATFORM_EMBEDDING_MODEL="text-embedding-3-small"
+$env:AI_PLATFORM_EMBEDDING_API_KEY="your-embedding-api-key"
+$env:AI_PLATFORM_EMBEDDING_BASE_URL="https://api.openai.com/v1"
+$env:AI_PLATFORM_EMBEDDING_FALLBACK_ENABLED="true"
+$env:AI_PLATFORM_VECTOR_SEARCH_BACKEND="pgvector"
+python -m alembic upgrade head
+```
+
+The pgvector migration creates the `vector` extension and adds an `embedding`
+column for new knowledge chunks. The JSON embedding payload remains as a
+portable fallback, so SQLite tests and non-pgvector environments continue to
+work. Re-import or re-index existing knowledge documents after enabling a real
+embedding model so old chunks receive production vectors.
+
 Health endpoints:
 
 - `GET /health`

yuqei-ai-platform/services/ai-platform-api/migrations/versions/20260622_0004_add_embedding_provider_and_pgvector.py

@@ -0,0 +1,40 @@
+"""add embedding provider and pgvector column
+
+Revision ID: 20260622_0004
+Revises: 20260622_0003
+Create Date: 2026-06-22
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "20260622_0004"
+down_revision: Union[str, None] = "20260622_0003"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    op.add_column(
+        "knowledge_chunks",
+        sa.Column("embedding_provider", sa.String(length=64), nullable=False, server_default="local-hash"),
+    )
+    op.create_index("ix_knowledge_chunks_embedding_provider", "knowledge_chunks", ["embedding_provider"])
+
+    if bind.dialect.name == "postgresql":
+        op.execute("CREATE EXTENSION IF NOT EXISTS vector")
+        op.execute("ALTER TABLE knowledge_chunks ADD COLUMN IF NOT EXISTS embedding vector")
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+
+    if bind.dialect.name == "postgresql":
+        op.execute("ALTER TABLE knowledge_chunks DROP COLUMN IF EXISTS embedding")
+
+    op.drop_index("ix_knowledge_chunks_embedding_provider", table_name="knowledge_chunks")
+    op.drop_column("knowledge_chunks", "embedding_provider")

yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/config.py

@@ -20,6 +20,14 @@ class AiPlatformSettings(BaseSettings):
     provider_base_url: str | None = None
     provider_request_timeout_seconds: float = 30.0
     provider_fallback_enabled: bool = True
+    embedding_provider: Literal["local-hash", "openai-compatible"] = "local-hash"
+    embedding_model: str = "local-hash-v1"
+    embedding_dimension: int = 64
+    embedding_api_key: str | None = None
+    embedding_base_url: str | None = None
+    embedding_request_timeout_seconds: float = 30.0
+    embedding_fallback_enabled: bool = True
+    vector_search_backend: Literal["auto", "json", "pgvector"] = "auto"
 
     model_config = SettingsConfigDict(
         env_file=".env",

yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/embeddings.py

@@ -0,0 +1,223 @@
+from __future__ import annotations
+
+import hashlib
+import math
+import re
+from dataclasses import dataclass
+from typing import Any, Protocol
+
+import httpx
+
+from yuqei_ai_platform_api.config import AiPlatformSettings
+
+
+LOCAL_HASH_EMBEDDING_MODEL = "local-hash-v1"
+LOCAL_HASH_EMBEDDING_DIMENSION = 64
+
+GENERIC_RETRIEVAL_TERMS = {
+    "合同",
+    "这个",
+    "需要",
+    "注意",
+    "什么",
+    "事项",
+    "问题",
+    "风险",
+    "如何",
+    "怎么",
+    "是否",
+    "可以",
+    "应该",
+    "处理",
+    "相关",
+}
+
+
+@dataclass(frozen=True)
+class EmbeddingVector:
+    vector: list[float]
+    model: str
+    dimension: int
+    provider: str
+
+
+class EmbeddingProvider(Protocol):
+    provider_name: str
+    model: str
+
+    def embed_texts(self, texts: list[str]) -> list[EmbeddingVector]: ...
+
+
+class LocalHashEmbeddingProvider:
+    provider_name = "local-hash"
+
+    def __init__(
+        self,
+        *,
+        model: str = LOCAL_HASH_EMBEDDING_MODEL,
+        dimension: int = LOCAL_HASH_EMBEDDING_DIMENSION,
+    ) -> None:
+        self.model = model
+        self.dimension = dimension
+
+    def embed_texts(self, texts: list[str]) -> list[EmbeddingVector]:
+        return [
+            EmbeddingVector(
+                vector=build_local_hash_embedding(text, dimension=self.dimension),
+                model=self.model,
+                dimension=self.dimension,
+                provider=self.provider_name,
+            )
+            for text in texts
+        ]
+
+
+class OpenAICompatibleEmbeddingProvider:
+    provider_name = "openai-compatible"
+
+    def __init__(
+        self,
+        settings: AiPlatformSettings,
+        *,
+        transport: httpx.BaseTransport | None = None,
+        fallback_provider: EmbeddingProvider | None = None,
+    ) -> None:
+        self._settings = settings
+        self._transport = transport
+        self._fallback_provider = fallback_provider or LocalHashEmbeddingProvider()
+        self.model = settings.embedding_model
+
+    def embed_texts(self, texts: list[str]) -> list[EmbeddingVector]:
+        if not texts:
+            return []
+        api_key = self._settings.embedding_api_key or self._settings.provider_api_key
+        base_url = (self._settings.embedding_base_url or self._settings.provider_base_url or "").rstrip("/")
+        if not api_key or not base_url:
+            return self._fallback(texts)
+        try:
+            with httpx.Client(
+                timeout=self._settings.embedding_request_timeout_seconds,
+                transport=self._transport,
+            ) as client:
+                response = client.post(
+                    f"{base_url}/embeddings",
+                    headers={
+                        "Authorization": f"Bearer {api_key}",
+                        "Content-Type": "application/json",
+                    },
+                    json={
+                        "model": self._settings.embedding_model,
+                        "input": texts,
+                    },
+                )
+                response.raise_for_status()
+        except httpx.HTTPError:
+            return self._fallback(texts)
+
+        try:
+            payload = response.json()
+        except ValueError:
+            return self._fallback(texts)
+        if not isinstance(payload, dict):
+            return self._fallback(texts)
+
+        vectors = _extract_openai_embedding_vectors(
+            payload,
+            model=self._settings.embedding_model,
+            provider=self.provider_name,
+        )
+        if len(vectors) != len(texts):
+            return self._fallback(texts)
+        return vectors
+
+    def _fallback(self, texts: list[str]) -> list[EmbeddingVector]:
+        if not self._settings.embedding_fallback_enabled:
+            return [
+                EmbeddingVector(
+                    vector=[],
+                    model=self._settings.embedding_model,
+                    dimension=self._settings.embedding_dimension,
+                    provider=self.provider_name,
+                )
+                for _ in texts
+            ]
+        return self._fallback_provider.embed_texts(texts)
+
+
+def build_embedding_provider(
+    settings: AiPlatformSettings,
+    *,
+    transport: httpx.BaseTransport | None = None,
+) -> EmbeddingProvider:
+    fallback = LocalHashEmbeddingProvider(
+        model=LOCAL_HASH_EMBEDDING_MODEL,
+        dimension=LOCAL_HASH_EMBEDDING_DIMENSION,
+    )
+    if settings.embedding_provider == "openai-compatible":
+        return OpenAICompatibleEmbeddingProvider(settings, transport=transport, fallback_provider=fallback)
+    return LocalHashEmbeddingProvider(
+        model=settings.embedding_model or LOCAL_HASH_EMBEDDING_MODEL,
+        dimension=settings.embedding_dimension or LOCAL_HASH_EMBEDDING_DIMENSION,
+    )
+
+
+def build_local_hash_embedding(text: str, *, dimension: int = LOCAL_HASH_EMBEDDING_DIMENSION) -> list[float]:
+    terms = tokenize_embedding_text(text)
+    if not terms:
+        return []
+    vector = [0.0 for _ in range(dimension)]
+    for term in terms:
+        digest = hashlib.sha256(term.encode("utf-8")).digest()
+        bucket = int.from_bytes(digest[:4], "big") % dimension
+        sign = 1.0 if digest[4] % 2 == 0 else -1.0
+        vector[bucket] += sign
+    norm = math.sqrt(sum(value * value for value in vector))
+    if norm <= 0:
+        return []
+    return [round(value / norm, 6) for value in vector]
+
+
+def tokenize_embedding_text(text: str) -> list[str]:
+    normalized = text.strip().lower()
+    if not normalized:
+        return []
+    latin_terms = re.findall(r"[a-z0-9]+", normalized)
+    cjk_chars = re.findall(r"[\u4e00-\u9fff]", normalized)
+    cjk_bigrams = [f"{cjk_chars[index]}{cjk_chars[index + 1]}" for index in range(len(cjk_chars) - 1)]
+    terms = [*latin_terms, *cjk_bigrams]
+    if not terms:
+        terms = [term for term in normalized.replace("\n", " ").split(" ") if term]
+    return [term for term in dict.fromkeys(terms or [normalized]) if term not in GENERIC_RETRIEVAL_TERMS]
+
+
+def _extract_openai_embedding_vectors(
+    payload: dict[str, Any],
+    *,
+    model: str,
+    provider: str,
+) -> list[EmbeddingVector]:
+    rows = payload.get("data")
+    if not isinstance(rows, list):
+        return []
+
+    ordered_rows = sorted(
+        [row for row in rows if isinstance(row, dict)],
+        key=lambda row: row.get("index", 0),
+    )
+    vectors: list[EmbeddingVector] = []
+    for row in ordered_rows:
+        embedding = row.get("embedding")
+        if not isinstance(embedding, list):
+            return []
+        values = [float(value) for value in embedding if isinstance(value, int | float)]
+        if len(values) != len(embedding):
+            return []
+        vectors.append(
+            EmbeddingVector(
+                vector=values,
+                model=str(payload.get("model") or model),
+                dimension=len(values),
+                provider=provider,
+            )
+        )
+    return vectors

yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/models.py

@@ -90,6 +90,7 @@ class KnowledgeChunkModel(Base):
     content: Mapped[str] = mapped_column(Text)
     chunk_index: Mapped[int] = mapped_column(Integer, default=0)
     locator: Mapped[str] = mapped_column(String(128), default="")
+    embedding_provider: Mapped[str] = mapped_column(String(64), default="local-hash", nullable=False)
     embedding_model: Mapped[str] = mapped_column(String(64), default="local-hash-v1")
     embedding_dimension: Mapped[int] = mapped_column(Integer, default=64)
     embedding_json: Mapped[str | None] = mapped_column(Text, nullable=True)

yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/repository.py

@@ -1,9 +1,6 @@
 from __future__ import annotations
 
-import hashlib
 import json
-import math
-import re
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from threading import Lock
@@ -11,11 +8,18 @@ from typing import Protocol
 from uuid import uuid4
 
 from pydantic import BaseModel, Field
-from sqlalchemy import create_engine, select
+from sqlalchemy import create_engine, select, text
 from sqlalchemy.engine import Engine
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import Session, sessionmaker
 from sqlalchemy.pool import StaticPool
 
+from yuqei_ai_platform_api.embeddings import (
+    EmbeddingProvider,
+    LOCAL_HASH_EMBEDDING_DIMENSION,
+    LOCAL_HASH_EMBEDDING_MODEL,
+    LocalHashEmbeddingProvider,
+    tokenize_embedding_text,
+)
 from yuqei_ai_platform_api.legal_qa import LegalCitation
 from yuqei_ai_platform_api.models import (
     AiProviderConfigModel,
@@ -28,27 +32,10 @@ from yuqei_ai_platform_api.models import (
 )
 
 
-EMBEDDING_MODEL = "local-hash-v1"
-EMBEDDING_DIMENSION = 64
 RETRIEVAL_STRATEGIES = {"keyword", "vector", "hybrid"}
 MIN_VECTOR_RECALL_SCORE = 0.35
-GENERIC_RETRIEVAL_TERMS = {
-    "合同",
-    "这个",
-    "需要",
-    "注意",
-    "什么",
-    "事项",
-    "问题",
-    "风险",
-    "如何",
-    "怎么",
-    "是否",
-    "可以",
-    "应该",
-    "处理",
-    "相关",
-}
+EMBEDDING_MODEL = LOCAL_HASH_EMBEDDING_MODEL
+EMBEDDING_DIMENSION = LOCAL_HASH_EMBEDDING_DIMENSION
 
 
 class ProviderConfigCreate(BaseModel):
@@ -120,6 +107,7 @@ class KnowledgeChunk(BaseModel):
     content: str
     chunk_index: int
     locator: str
+    embedding_provider: str = "local-hash"
     embedding_model: str = EMBEDDING_MODEL
     embedding_dimension: int = EMBEDDING_DIMENSION
     embedding_vector: list[float] = Field(default_factory=list, exclude=True)
@@ -227,6 +215,7 @@ class InMemoryAiPlatformRepository:
     knowledge_documents: dict[str, KnowledgeDocument] = field(default_factory=dict)
     knowledge_chunks: dict[str, KnowledgeChunk] = field(default_factory=dict)
     ai_run_audits: list[AiRunAudit] = field(default_factory=list)
+    embedding_provider: EmbeddingProvider = field(default_factory=LocalHashEmbeddingProvider)
     _lock: Lock = field(default_factory=Lock)
 
     def upsert_provider_config(self, payload: ProviderConfigCreate) -> ProviderConfig:
@@ -308,7 +297,7 @@ class InMemoryAiPlatformRepository:
                 **payload.model_dump(),
             )
             self.knowledge_documents[document.id] = document
-            chunks = split_knowledge_document(document)
+            chunks = split_knowledge_document(document, embedding_provider=self.embedding_provider)
             for chunk in chunks:
                 self.knowledge_chunks[chunk.id] = chunk
             if document.import_batch_id and document.import_batch_id in self.knowledge_import_batches:
@@ -387,6 +376,7 @@ class InMemoryAiPlatformRepository:
             knowledge_base_ids=knowledge_base_ids,
             limit=limit,
             strategy=strategy,
+            embedding_provider=self.embedding_provider,
         )
 
     def add_ai_run_audit(self, audit: AiRunAudit) -> AiRunAudit:
@@ -415,9 +405,13 @@ class SqlAlchemyAiPlatformRepository:
         *,
         engine: Engine | None = None,
         auto_create: bool = True,
+        embedding_provider: EmbeddingProvider | None = None,
+        vector_search_backend: str = "auto",
     ) -> None:
         self._engine = engine or build_sqlalchemy_engine(database_url)
         self._session_factory = sessionmaker(bind=self._engine, expire_on_commit=False)
+        self._embedding_provider = embedding_provider or LocalHashEmbeddingProvider()
+        self._vector_search_backend = vector_search_backend
         if auto_create:
             Base.metadata.create_all(self._engine)
 
@@ -516,18 +510,26 @@ class SqlAlchemyAiPlatformRepository:
         )
         with self._session_factory.begin() as session:
             session.add(document)
-            chunks = [
-                _knowledge_chunk_model_from_dto(chunk)
-                for chunk in split_knowledge_document(_knowledge_document_from_model(document))
-            ]
+            chunk_dtos = split_knowledge_document(
+                _knowledge_document_from_model(document),
+                embedding_provider=self._embedding_provider,
+            )
+            chunks = [_knowledge_chunk_model_from_dto(chunk) for chunk in chunk_dtos]
             session.add_all(chunks)
+            session.flush()
+            if self._engine.dialect.name == "postgresql" and self._has_pgvector_embedding_column(session):
+                for chunk in chunk_dtos:
+                    if chunk.embedding_vector:
+                        session.execute(
+                            text("UPDATE knowledge_chunks SET embedding = CAST(:embedding AS vector) WHERE id = :id"),
+                            {"embedding": _to_pgvector_literal(chunk.embedding_vector), "id": chunk.id},
+                        )
             if document.import_batch_id:
                 batch = session.get(KnowledgeImportBatchModel, document.import_batch_id)
                 if batch:
                     batch.documents_count += 1
                     batch.chunks_count += len(chunks)
                     batch.updated_at = datetime.now(UTC)
-            session.flush()
             return _knowledge_document_from_model(document)
 
     def create_knowledge_import_batch(self, payload: KnowledgeImportBatchCreate) -> KnowledgeImportBatch:
@@ -599,6 +601,15 @@ class SqlAlchemyAiPlatformRepository:
         limit: int = 5,
         strategy: str = "hybrid",
     ) -> list[KnowledgeSearchResult]:
+        if self._should_use_pgvector(strategy):
+            results = self._search_knowledge_chunks_with_pgvector(
+                keyword,
+                knowledge_base_ids=knowledge_base_ids,
+                limit=limit,
+                strategy=strategy,
+            )
+            if results:
+                return results
         with self._session_factory() as session:
             statement = select(KnowledgeChunkModel).order_by(KnowledgeChunkModel.created_at.desc())
             if knowledge_base_ids:
@@ -610,8 +621,105 @@ class SqlAlchemyAiPlatformRepository:
             knowledge_base_ids=knowledge_base_ids,
             limit=limit,
             strategy=strategy,
+            embedding_provider=self._embedding_provider,
         )
 
+    def _should_use_pgvector(self, strategy: str) -> bool:
+        if strategy not in {"vector", "hybrid"}:
+            return False
+        if self._vector_search_backend == "json":
+            return False
+        if self._vector_search_backend == "pgvector":
+            return self._engine.dialect.name == "postgresql"
+        return self._engine.dialect.name == "postgresql"
+
+    def _has_pgvector_embedding_column(self, session: Session) -> bool:
+        try:
+            return bool(
+                session.execute(
+                    text(
+                        "SELECT EXISTS ("
+                        "SELECT 1 FROM information_schema.columns "
+                        "WHERE table_name = 'knowledge_chunks' AND column_name = 'embedding'"
+                        ")"
+                    )
+                ).scalar()
+            )
+        except Exception:
+            return False
+
+    def _search_knowledge_chunks_with_pgvector(
+        self,
+        keyword: str,
+        *,
+        knowledge_base_ids: list[str] | None = None,
+        limit: int = 5,
+        strategy: str = "hybrid",
+    ) -> list[KnowledgeSearchResult]:
+        terms = _tokenize(keyword)
+        if keyword.strip() and not terms:
+            return []
+        query_vector = _embed_query(keyword, self._embedding_provider)
+        if not query_vector:
+            return []
+
+        statement = (
+            "SELECT id, document_id, import_batch_id, knowledge_base_id, title, source_type, reference, "
+            "content, chunk_index, locator, embedding_provider, embedding_model, embedding_dimension, "
+            "embedding_json, created_at, "
+            "GREATEST(1 - (embedding <=> CAST(:query_vector AS vector)), 0) AS vector_score "
+            "FROM knowledge_chunks "
+            "WHERE embedding IS NOT NULL AND embedding_dimension = :query_dimension"
+        )
+        params: dict[str, object] = {
+            "query_vector": _to_pgvector_literal(query_vector),
+            "query_dimension": len(query_vector),
+            "limit": limit,
+        }
+        if knowledge_base_ids:
+            placeholders: list[str] = []
+            for index, knowledge_base_id in enumerate(knowledge_base_ids):
+                param_name = f"knowledge_base_id_{index}"
+                placeholders.append(f":{param_name}")
+                params[param_name] = knowledge_base_id
+            statement += f" AND knowledge_base_id IN ({', '.join(placeholders)})"
+        statement += " ORDER BY embedding <=> CAST(:query_vector AS vector) LIMIT :limit"
+
+        try:
+            with self._session_factory() as session:
+                rows = session.execute(text(statement), params).mappings().all()
+        except Exception:
+            return []
+
+        results: list[KnowledgeSearchResult] = []
+        retrieval_strategy = strategy if strategy in RETRIEVAL_STRATEGIES else "hybrid"
+        for row in rows:
+            chunk = _knowledge_chunk_from_mapping(row)
+            haystack = f"{chunk.title}\n{chunk.reference}\n{chunk.content}".lower()
+            matched_terms = [term for term in terms if term in haystack]
+            keyword_score = _score_chunk(chunk, matched_terms, terms_count=len(terms))
+            vector_score = round(float(row.get("vector_score") or 0.0), 4)
+            if terms and not _passes_retrieval_threshold(
+                keyword_score=keyword_score,
+                vector_score=vector_score,
+                strategy=retrieval_strategy,
+            ):
+                continue
+            hybrid_score = round(keyword_score * 0.55 + vector_score * 0.45, 4)
+            score = vector_score if retrieval_strategy == "vector" else hybrid_score
+            results.append(
+                KnowledgeSearchResult(
+                    chunk=chunk,
+                    score=score,
+                    keyword_score=keyword_score,
+                    vector_score=vector_score,
+                    hybrid_score=hybrid_score,
+                    retrieval_strategy=retrieval_strategy,
+                    matched_terms=matched_terms,
+                )
+            )
+        return sorted(results, key=lambda item: (item.score, item.chunk.created_at), reverse=True)[:limit]
+
     def add_ai_run_audit(self, audit: AiRunAudit) -> AiRunAudit:
         row = AiRunAuditModel(
             id=audit.id,
@@ -776,6 +884,7 @@ def _knowledge_chunk_from_model(row: KnowledgeChunkModel) -> KnowledgeChunk:
         content=row.content,
         chunk_index=row.chunk_index,
         locator=row.locator,
+        embedding_provider=row.embedding_provider,
         embedding_model=row.embedding_model,
         embedding_dimension=row.embedding_dimension,
         embedding_vector=_decode_embedding(row.embedding_json),
@@ -795,6 +904,7 @@ def _knowledge_chunk_model_from_dto(chunk: KnowledgeChunk) -> KnowledgeChunkMode
         content=chunk.content,
         chunk_index=chunk.chunk_index,
         locator=chunk.locator,
+        embedding_provider=chunk.embedding_provider,
         embedding_model=chunk.embedding_model,
         embedding_dimension=chunk.embedding_dimension,
         embedding_json=_encode_embedding(chunk.embedding_vector),
@@ -885,7 +995,12 @@ def _default_knowledge_documents() -> list[KnowledgeDocumentCreate]:
     ]
 
 
-def split_knowledge_document(document: KnowledgeDocument, *, max_chars: int = 500) -> list[KnowledgeChunk]:
+def split_knowledge_document(
+    document: KnowledgeDocument,
+    *,
+    max_chars: int = 500,
+    embedding_provider: EmbeddingProvider | None = None,
+) -> list[KnowledgeChunk]:
     raw_segments = [segment.strip() for segment in document.content.splitlines() if segment.strip()]
     segments: list[str] = []
     current = ""
@@ -900,8 +1015,11 @@ def split_knowledge_document(document: KnowledgeDocument, *, max_chars: int = 50
         segments.append(current)
 
     chunks: list[KnowledgeChunk] = []
+    provider = embedding_provider or LocalHashEmbeddingProvider()
+    embedding_texts = [f"{document.title}\n{document.reference}\n{content}" for content in segments]
+    embeddings = provider.embed_texts(embedding_texts)
     for index, content in enumerate(segments, start=1):
-        embedding_text = f"{document.title}\n{document.reference}\n{content}"
+        embedding = embeddings[index - 1] if index - 1 < len(embeddings) else None
         chunks.append(
             KnowledgeChunk(
                 id=f"chunk-{uuid4().hex[:12]}",
@@ -914,9 +1032,10 @@ def split_knowledge_document(document: KnowledgeDocument, *, max_chars: int = 50
                 content=content,
                 chunk_index=index,
                 locator=f"chunk-{index}",
-                embedding_model=EMBEDDING_MODEL,
-                embedding_dimension=EMBEDDING_DIMENSION,
-                embedding_vector=_build_embedding(embedding_text),
+                embedding_provider=embedding.provider if embedding else provider.provider_name,
+                embedding_model=embedding.model if embedding else provider.model,
+                embedding_dimension=embedding.dimension if embedding else 0,
+                embedding_vector=embedding.vector if embedding else [],
                 created_at=datetime.now(UTC),
             )
         )
@@ -930,13 +1049,15 @@ def _rank_knowledge_chunks(
     knowledge_base_ids: list[str] | None = None,
     limit: int = 5,
     strategy: str = "hybrid",
+    embedding_provider: EmbeddingProvider | None = None,
 ) -> list[KnowledgeSearchResult]:
     allowed_base_ids = set(knowledge_base_ids or [])
     retrieval_strategy = strategy if strategy in RETRIEVAL_STRATEGIES else "hybrid"
     terms = _tokenize(keyword)
     if keyword.strip() and not terms:
         return []
-    query_vector = _build_embedding(keyword)
+    provider = embedding_provider or LocalHashEmbeddingProvider()
+    query_vector = _embed_query(keyword, provider)
     results: list[KnowledgeSearchResult] = []
     for chunk in chunks:
         if allowed_base_ids and chunk.knowledge_base_id not in allowed_base_ids:
@@ -975,16 +1096,7 @@ def _rank_knowledge_chunks(
 
 
 def _tokenize(keyword: str) -> list[str]:
-    normalized = keyword.strip().lower()
-    if not normalized:
-        return []
-    latin_terms = re.findall(r"[a-z0-9]+", normalized)
-    cjk_chars = re.findall(r"[\u4e00-\u9fff]", normalized)
-    cjk_bigrams = [f"{cjk_chars[index]}{cjk_chars[index + 1]}" for index in range(len(cjk_chars) - 1)]
-    terms = [*latin_terms, *cjk_bigrams]
-    if not terms:
-        terms = [term for term in normalized.replace("\n", " ").split(" ") if term]
-    return [term for term in dict.fromkeys(terms or [normalized]) if term not in GENERIC_RETRIEVAL_TERMS]
+    return tokenize_embedding_text(keyword)
 
 
 def _score_chunk(chunk: KnowledgeChunk, matched_terms: list[str], *, terms_count: int) -> float:
@@ -1012,26 +1124,15 @@ def _passes_retrieval_threshold(
     return keyword_score > 0 or vector_score >= MIN_VECTOR_RECALL_SCORE
 
 
-def _build_embedding(text: str, *, dimension: int = EMBEDDING_DIMENSION) -> list[float]:
-    terms = _tokenize(text)
-    if not terms:
-        return []
-    vector = [0.0 for _ in range(dimension)]
-    for term in terms:
-        digest = hashlib.sha256(term.encode("utf-8")).digest()
-        bucket = int.from_bytes(digest[:4], "big") % dimension
-        sign = 1.0 if digest[4] % 2 == 0 else -1.0
-        vector[bucket] += sign
-    norm = math.sqrt(sum(value * value for value in vector))
-    if norm <= 0:
-        return []
-    return [round(value / norm, 6) for value in vector]
+def _embed_query(text: str, embedding_provider: EmbeddingProvider) -> list[float]:
+    embeddings = embedding_provider.embed_texts([text])
+    return embeddings[0].vector if embeddings else []
 
 
 def _chunk_embedding(chunk: KnowledgeChunk) -> list[float]:
     if chunk.embedding_vector:
         return chunk.embedding_vector
-    return _build_embedding(f"{chunk.title}\n{chunk.reference}\n{chunk.content}")
+    return _embed_query(f"{chunk.title}\n{chunk.reference}\n{chunk.content}", LocalHashEmbeddingProvider())
 
 
 def _vector_score(query_vector: list[float], chunk_vector: list[float]) -> float:
@@ -1064,6 +1165,30 @@ def _decode_embedding(payload: str | None) -> list[float]:
     return values
 
 
+def _knowledge_chunk_from_mapping(row: object) -> KnowledgeChunk:
+    return KnowledgeChunk(
+        id=str(row["id"]),  # type: ignore[index]
+        document_id=str(row["document_id"]),  # type: ignore[index]
+        import_batch_id=row["import_batch_id"],  # type: ignore[index]
+        knowledge_base_id=str(row["knowledge_base_id"]),  # type: ignore[index]
+        title=str(row["title"]),  # type: ignore[index]
+        source_type=str(row["source_type"]),  # type: ignore[index]
+        reference=str(row["reference"]),  # type: ignore[index]
+        content=str(row["content"]),  # type: ignore[index]
+        chunk_index=int(row["chunk_index"]),  # type: ignore[index]
+        locator=str(row["locator"]),  # type: ignore[index]
+        embedding_provider=str(row["embedding_provider"] or "local-hash"),  # type: ignore[index]
+        embedding_model=str(row["embedding_model"]),  # type: ignore[index]
+        embedding_dimension=int(row["embedding_dimension"] or 0),  # type: ignore[index]
+        embedding_vector=_decode_embedding(row["embedding_json"]),  # type: ignore[index]
+        created_at=_ensure_utc(row["created_at"]),  # type: ignore[index]
+    )
+
+
+def _to_pgvector_literal(vector: list[float]) -> str:
+    return "[" + ",".join(str(round(value, 8)) for value in vector) + "]"
+
+
 def _encode_json_list(values: list[object]) -> str:
     return json.dumps(values, ensure_ascii=False, separators=(",", ":"))
 

yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/state.py

@@ -2,6 +2,7 @@ from functools import lru_cache
 from typing import Literal
 
 from yuqei_ai_platform_api.config import AiPlatformSettings, get_settings
+from yuqei_ai_platform_api.embeddings import build_embedding_provider
 from yuqei_ai_platform_api.repository import (
     AiPlatformRepository,
     InMemoryAiPlatformRepository,
@@ -15,6 +16,16 @@ def get_repository(settings: AiPlatformSettings | None = None) -> AiPlatformRepo
         resolved_settings.repository_backend,
         resolved_settings.database_url,
         resolved_settings.database_auto_create,
+        resolved_settings.embedding_provider,
+        resolved_settings.embedding_model,
+        resolved_settings.embedding_dimension,
+        resolved_settings.embedding_base_url or "",
+        resolved_settings.embedding_api_key or "",
+        resolved_settings.embedding_fallback_enabled,
+        resolved_settings.embedding_request_timeout_seconds,
+        resolved_settings.provider_api_key or "",
+        resolved_settings.provider_base_url or "",
+        resolved_settings.vector_search_backend,
     )
 
 
@@ -23,10 +34,40 @@ def _get_repository(
     repository_backend: Literal["memory", "sqlalchemy"],
     database_url: str,
     database_auto_create: bool,
+    embedding_provider_name: Literal["local-hash", "openai-compatible"],
+    embedding_model: str,
+    embedding_dimension: int,
+    embedding_base_url: str,
+    embedding_api_key: str,
+    embedding_fallback_enabled: bool,
+    embedding_request_timeout_seconds: float,
+    provider_api_key: str,
+    provider_base_url: str,
+    vector_search_backend: Literal["auto", "json", "pgvector"],
 ) -> AiPlatformRepository:
+    settings = get_settings().model_copy(
+        update={
+            "embedding_provider": embedding_provider_name,
+            "embedding_model": embedding_model,
+            "embedding_dimension": embedding_dimension,
+            "embedding_base_url": embedding_base_url or None,
+            "embedding_api_key": embedding_api_key or None,
+            "embedding_fallback_enabled": embedding_fallback_enabled,
+            "embedding_request_timeout_seconds": embedding_request_timeout_seconds,
+            "provider_api_key": provider_api_key or None,
+            "provider_base_url": provider_base_url or None,
+            "vector_search_backend": vector_search_backend,
+        }
+    )
+    embedding_provider = build_embedding_provider(settings)
     if repository_backend == "sqlalchemy":
-        repository = SqlAlchemyAiPlatformRepository(database_url, auto_create=database_auto_create)
+        repository = SqlAlchemyAiPlatformRepository(
+            database_url,
+            auto_create=database_auto_create,
+            embedding_provider=embedding_provider,
+            vector_search_backend=vector_search_backend,
+        )
     else:
-        repository = InMemoryAiPlatformRepository()
+        repository = InMemoryAiPlatformRepository(embedding_provider=embedding_provider)
     repository.seed_defaults()
     return repository

yuqei-ai-platform/services/ai-platform-api/tests/test_embeddings.py

@@ -0,0 +1,92 @@
+import json
+
+import httpx
+
+from yuqei_ai_platform_api.config import AiPlatformSettings
+from yuqei_ai_platform_api.embeddings import (
+    LOCAL_HASH_EMBEDDING_MODEL,
+    OpenAICompatibleEmbeddingProvider,
+    build_embedding_provider,
+)
+
+
+def test_openai_compatible_embedding_provider_calls_embeddings_endpoint() -> None:
+    captured_payload: dict[str, object] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.url.path == "/v1/embeddings"
+        assert request.headers["Authorization"] == "Bearer embedding-key"
+        captured_payload.update(json.loads(request.read().decode("utf-8")))
+        return httpx.Response(
+            200,
+            json={
+                "model": "text-embedding-prod",
+                "data": [
+                    {"index": 0, "embedding": [0.1, 0.2, 0.3]},
+                    {"index": 1, "embedding": [0.3, 0.2, 0.1]},
+                ],
+            },
+        )
+
+    provider = OpenAICompatibleEmbeddingProvider(
+        AiPlatformSettings(
+            environment="test",
+            embedding_provider="openai-compatible",
+            embedding_model="text-embedding-prod",
+            embedding_api_key="embedding-key",
+            embedding_base_url="https://embedding.example.test/v1",
+        ),
+        transport=httpx.MockTransport(handler),
+    )
+
+    vectors = provider.embed_texts(["lease deposit", "probation period"])
+
+    assert captured_payload == {
+        "model": "text-embedding-prod",
+        "input": ["lease deposit", "probation period"],
+    }
+    assert vectors[0].provider == "openai-compatible"
+    assert vectors[0].model == "text-embedding-prod"
+    assert vectors[0].dimension == 3
+    assert vectors[0].vector == [0.1, 0.2, 0.3]
+
+
+def test_openai_compatible_embedding_provider_falls_back_to_local_hash() -> None:
+    provider = build_embedding_provider(
+        AiPlatformSettings(
+            environment="test",
+            embedding_provider="openai-compatible",
+            embedding_model="text-embedding-prod",
+            embedding_api_key=None,
+            embedding_base_url="https://embedding.example.test/v1",
+            embedding_fallback_enabled=True,
+        )
+    )
+
+    vectors = provider.embed_texts(["lease deposit"])
+
+    assert vectors[0].provider == "local-hash"
+    assert vectors[0].model == LOCAL_HASH_EMBEDDING_MODEL
+    assert vectors[0].dimension == 64
+    assert vectors[0].vector
+
+
+def test_openai_compatible_embedding_provider_can_disable_fallback() -> None:
+    provider = build_embedding_provider(
+        AiPlatformSettings(
+            environment="test",
+            embedding_provider="openai-compatible",
+            embedding_model="text-embedding-prod",
+            embedding_dimension=1536,
+            embedding_api_key=None,
+            embedding_base_url="https://embedding.example.test/v1",
+            embedding_fallback_enabled=False,
+        )
+    )
+
+    vectors = provider.embed_texts(["lease deposit"])
+
+    assert vectors[0].provider == "openai-compatible"
+    assert vectors[0].model == "text-embedding-prod"
+    assert vectors[0].dimension == 1536
+    assert vectors[0].vector == []

yuqei-ai-platform/services/ai-platform-api/tests/test_platform_data.py

@@ -1,6 +1,7 @@
 from fastapi.testclient import TestClient
 
 from yuqei_ai_platform_api.config import AiPlatformSettings
+from yuqei_ai_platform_api.embeddings import EmbeddingVector
 from yuqei_ai_platform_api.main import create_app
 from yuqei_ai_platform_api.repository import (
     AiRunAudit,
@@ -13,6 +14,22 @@ from yuqei_ai_platform_api.repository import (
 )
 
 
+class TestEmbeddingProvider:
+    provider_name = "test-embedding"
+    model = "test-embedding-v1"
+
+    def embed_texts(self, texts: list[str]) -> list[EmbeddingVector]:
+        return [
+            EmbeddingVector(
+                vector=[1.0, 0.0, 0.0] if "deposit" in text.lower() else [0.0, 1.0, 0.0],
+                model=self.model,
+                dimension=3,
+                provider=self.provider_name,
+            )
+            for text in texts
+        ]
+
+
 def make_client() -> TestClient:
     repository = InMemoryAiPlatformRepository()
     repository.seed_defaults()
@@ -128,6 +145,7 @@ def test_knowledge_import_batch_chunks_and_scored_search() -> None:
     assert chunks_response.status_code == 200
     chunks = chunks_response.json()
     assert chunks[0]["locator"] == "chunk-1"
+    assert chunks[0]["embedding_provider"] == "local-hash"
     assert chunks[0]["embedding_model"] == "local-hash-v1"
     assert chunks[0]["embedding_dimension"] == 64
 
@@ -151,6 +169,30 @@ def test_knowledge_import_batch_chunks_and_scored_search() -> None:
     assert refreshed_batch["chunks_count"] >= 1
 
 
+def test_repository_uses_injected_embedding_provider_for_chunks_and_search() -> None:
+    repository = InMemoryAiPlatformRepository(embedding_provider=TestEmbeddingProvider())
+    document = repository.add_knowledge_document(
+        KnowledgeDocumentCreate(
+            knowledge_base_id="laws-cn",
+            title="Lease Policy",
+            source_type="policy",
+            reference="Lease-001",
+            content="Deposit refund must be clearly written.",
+        )
+    )
+
+    chunk = repository.list_knowledge_chunks(document.id)[0]
+    assert chunk.embedding_provider == "test-embedding"
+    assert chunk.embedding_model == "test-embedding-v1"
+    assert chunk.embedding_dimension == 3
+    assert chunk.embedding_vector == [1.0, 0.0, 0.0]
+
+    result = repository.search_knowledge_chunks("deposit", strategy="vector")[0]
+    assert result.chunk.id == chunk.id
+    assert result.retrieval_strategy == "vector"
+    assert result.vector_score >= 0.35
+
+
 def test_legal_qa_uses_chunk_search_and_records_audit() -> None:
     client = make_client()
     client.post(

yuqei-ai-sdk-python/src/yuqei_sdk/ai_platform.py

@@ -107,6 +107,7 @@ class KnowledgeChunk(BaseModel):
     content: str
     chunk_index: int
     locator: str
+    embedding_provider: str = "local-hash"
     embedding_model: str = "local-hash-v1"
     embedding_dimension: int = 64
     created_at: str

yuqei-ai-sdk-python/tests/test_ai_platform.py

@@ -211,6 +211,7 @@ def test_ai_platform_client_covers_admin_data_endpoints() -> None:
                         "content": "A lease contract defines rent and use of leased property.",
                         "chunk_index": 1,
                         "locator": "chunk-1",
+                        "embedding_provider": "local-hash",
                         "embedding_model": "local-hash-v1",
                         "embedding_dimension": 64,
                         "created_at": "2026-06-22T00:00:00Z",
@@ -253,6 +254,7 @@ def test_ai_platform_client_covers_admin_data_endpoints() -> None:
                             "content": "A lease contract defines rent and use of leased property.",
                             "chunk_index": 1,
                             "locator": "chunk-1",
+                            "embedding_provider": "local-hash",
                             "embedding_model": "local-hash-v1",
                             "embedding_dimension": 64,
                             "created_at": "2026-06-22T00:00:00Z",
@@ -341,6 +343,7 @@ def test_ai_platform_client_covers_admin_data_endpoints() -> None:
         )
     ).reference == "Article 703"
     assert client.list_knowledge_chunks("doc-1")[0].locator == "chunk-1"
+    assert client.list_knowledge_chunks("doc-1")[0].embedding_provider == "local-hash"
     assert client.list_knowledge_chunks("doc-1")[0].embedding_model == "local-hash-v1"
     assert client.search_knowledge_documents("lease", limit=3)[0].title == "Civil Code"
     chunk_result = client.search_knowledge_chunks("lease", limit=3)