yuqei/ai-contract-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add knowledge search explain endpoint

Browse Source
This commit is contained in:
yuqei
2026-06-22 23:42:01 +08:00
parent cf03b0eafd
commit 27894cc713
9 changed files with 610 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
project-docs/V2-ai-platform-embedding-pgvector-2026-06-22.md
+8
View File
@@ -44,6 +44,13 @@
- PostgreSQL + pgvector 环境会重新写入 `knowledge_chunks.embedding`。
- Python SDK 新增 `reindex_knowledge_documents(...)`。
7. 检索 explain 闭环。
- 新增 `GET /api/v1/knowledge-chunks/search/explain`。
- 返回 normalized terms、候选 chunk 数、命中数、keyword/vector/hybrid 分数。
- 返回每个候选项的 `included`、`rank`、`filter_reason`。
- Python SDK 新增 `explain_knowledge_search(...)`。
- JSON fallback 的向量分改为 cosine 口径，便于和 pgvector cosine 距离保持一致。
## 生产配置建议
```powershell
@@ -73,3 +80,4 @@ python -m alembic upgrade head
- provider 不可用且允许 fallback 时：chunk 仍可写入，检索仍可用，但 provider 显示 `local-hash`。
- pgvector 可用时：优先使用数据库向量距离排序。
- pgvector 不可用时：回退 JSON embedding，不影响法律问答主链路。
- explain 接口可说明命中依据和过滤原因，用于法律问答可信度展示和检索调参。
yuqei-ai-platform/packages/ai-openapi/openapi.yaml
+135
View File
@@ -373,6 +373,52 @@ paths:
type: array
items:
$ref: "#/components/schemas/KnowledgeSearchResult"
/api/v1/knowledge-chunks/search/explain:
get:
operationId: explainKnowledgeSearch
summary: Explain knowledge chunk retrieval scoring and filtering
parameters:
- name: keyword
in: query
required: false
schema:
type: string
default: ""
- name: knowledge_base_id
in: query
required: false
schema:
type: string
- name: limit
in: query
required: false
schema:
type: integer
default: 5
minimum: 1
maximum: 20
- name: candidate_limit
in: query
required: false
schema:
type: integer
default: 20
minimum: 1
maximum: 100
- name: strategy
in: query
required: false
schema:
type: string
enum: [keyword, vector, hybrid]
default: hybrid
responses:
"200":
description: Retrieval explain result
content:
application/json:
schema:
$ref: "#/components/schemas/KnowledgeSearchExplain"
/api/v1/audit/ai-runs:
get:
operationId: listAiRuns
@@ -969,6 +1015,95 @@ components:
type: array
items:
type: string
KnowledgeSearchExplainItem:
type: object
required:
- chunk
- included
properties:
chunk:
$ref: "#/components/schemas/KnowledgeChunk"
score:
type: number
format: float
keyword_score:
type: number
format: float
vector_score:
type: number
format: float
hybrid_score:
type: number
format: float
retrieval_strategy:
type: string
enum: [keyword, vector, hybrid]
matched_terms:
type: array
items:
type: string
included:
type: boolean
rank:
type: integer
nullable: true
filter_reason:
type: string
nullable: true
KnowledgeSearchExplain:
type: object
required:
- keyword
- retrieval_strategy
- embedding_provider
- embedding_model
- results
properties:
keyword:
type: string
normalized_terms:
type: array
items:
type: string
retrieval_strategy:
type: string
enum: [keyword, vector, hybrid]
knowledge_base_ids:
type: array
items:
type: string
limit:
type: integer
candidate_limit:
type: integer
candidate_count:
type: integer
included_count:
type: integer
retrieval_backend:
type: string
embedding_provider:
type: string
embedding_model:
type: string
query_embedding_dimension:
type: integer
keyword_threshold:
type: number
format: float
vector_threshold:
type: number
format: float
hybrid_keyword_weight:
type: number
format: float
hybrid_vector_weight:
type: number
format: float
results:
type: array
items:
$ref: "#/components/schemas/KnowledgeSearchExplainItem"
RetrievalSource:
type: object
required:
yuqei-ai-platform/services/ai-platform-api/README.md
+5
View File
@@ -71,6 +71,7 @@ Core MVP endpoints:
- `GET /api/v1/knowledge-documents/{document_id}/chunks`
- `GET /api/v1/knowledge-documents/search`
- `GET /api/v1/knowledge-chunks/search?strategy=keyword|vector|hybrid`
- `GET /api/v1/knowledge-chunks/search/explain?strategy=keyword|vector|hybrid`
- `POST /api/v1/legal/qa`
- `GET /api/v1/audit/ai-runs`
@@ -78,6 +79,10 @@ Knowledge chunk search returns keyword, vector, and hybrid scores. Legal QA uses
hybrid retrieval by default and records retrieved chunk ids plus structured source
metadata in AI run audits.
Use `GET /api/v1/knowledge-chunks/search/explain` to inspect normalized query
terms, keyword/vector/hybrid scores, inclusion rank, and filter reasons for the
candidate chunks considered by retrieval.
After switching from `local-hash-v1` to a real embedding model, call
`POST /api/v1/knowledge-documents/reindex` with a `document_id`,
`knowledge_base_id`, or `import_batch_id` filter to rebuild historical chunks
yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/main.py
+21
View File
@@ -21,6 +21,7 @@ from yuqei_ai_platform_api.repository import (
KnowledgeImportBatchCreate,
KnowledgeReindexRequest,
KnowledgeReindexResult,
KnowledgeSearchExplain,
KnowledgeSearchResult,
PromptTemplate,
PromptTemplateCreate,
@@ -180,6 +181,26 @@ def create_app(
strategy=strategy,
)
@app.get(
f"{resolved_settings.api_prefix}/knowledge-chunks/search/explain",
response_model=KnowledgeSearchExplain,
tags=["knowledge"],
)
def explain_knowledge_search(
keyword: str = "",
knowledge_base_id: str | None = None,
limit: int = Query(5, ge=1, le=20),
candidate_limit: int = Query(20, ge=1, le=100),
strategy: str = Query("hybrid", pattern="^(keyword|vector|hybrid)$"),
) -> KnowledgeSearchExplain:
return store.explain_knowledge_search(
keyword,
knowledge_base_ids=[knowledge_base_id] if knowledge_base_id else None,
limit=limit,
candidate_limit=candidate_limit,
strategy=strategy,
)
@app.get(
f"{resolved_settings.api_prefix}/audit/ai-runs",
response_model=list[AiRunAudit],
yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/repository.py
+265 -10
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import json
import math
from dataclasses import dataclass, field
from datetime import UTC, datetime
from threading import Lock
@@ -124,6 +125,39 @@ class KnowledgeSearchResult(BaseModel):
matched_terms: list[str] = Field(default_factory=list)
class KnowledgeSearchExplainItem(BaseModel):
chunk: KnowledgeChunk
score: float = 0.0
keyword_score: float = 0.0
vector_score: float = 0.0
hybrid_score: float = 0.0
retrieval_strategy: str = "hybrid"
matched_terms: list[str] = Field(default_factory=list)
included: bool = False
rank: int | None = None
filter_reason: str | None = None
class KnowledgeSearchExplain(BaseModel):
keyword: str
normalized_terms: list[str] = Field(default_factory=list)
retrieval_strategy: str = "hybrid"
knowledge_base_ids: list[str] = Field(default_factory=list)
limit: int = 5
candidate_limit: int = 20
candidate_count: int = 0
included_count: int = 0
retrieval_backend: str = "json-explain"
embedding_provider: str
embedding_model: str
query_embedding_dimension: int = 0
keyword_threshold: float = 0.0
vector_threshold: float = MIN_VECTOR_RECALL_SCORE
hybrid_keyword_weight: float = 0.55
hybrid_vector_weight: float = 0.45
results: list[KnowledgeSearchExplainItem] = Field(default_factory=list)
class KnowledgeReindexRequest(BaseModel):
document_id: str | None = None
knowledge_base_id: str | None = None
@@ -216,6 +250,16 @@ class AiPlatformRepository(Protocol):
strategy: str = "hybrid",
) -> list[KnowledgeSearchResult]: ...
def explain_knowledge_search(
self,
keyword: str,
*,
knowledge_base_ids: list[str] | None = None,
limit: int = 5,
candidate_limit: int = 20,
strategy: str = "hybrid",
) -> KnowledgeSearchExplain: ...
def reindex_knowledge_documents(self, payload: KnowledgeReindexRequest) -> KnowledgeReindexResult: ...
def add_ai_run_audit(self, audit: AiRunAudit) -> AiRunAudit: ...
@@ -397,6 +441,27 @@ class InMemoryAiPlatformRepository:
embedding_provider=self.embedding_provider,
)
def explain_knowledge_search(
self,
keyword: str,
*,
knowledge_base_ids: list[str] | None = None,
limit: int = 5,
candidate_limit: int = 20,
strategy: str = "hybrid",
) -> KnowledgeSearchExplain:
with self._lock:
chunks = list(self.knowledge_chunks.values())
return _explain_knowledge_chunks(
chunks,
keyword,
knowledge_base_ids=knowledge_base_ids,
limit=limit,
candidate_limit=candidate_limit,
strategy=strategy,
embedding_provider=self.embedding_provider,
)
def reindex_knowledge_documents(self, payload: KnowledgeReindexRequest) -> KnowledgeReindexResult:
with self._lock:
documents = [
@@ -683,6 +748,30 @@ class SqlAlchemyAiPlatformRepository:
embedding_provider=self._embedding_provider,
)
def explain_knowledge_search(
self,
keyword: str,
*,
knowledge_base_ids: list[str] | None = None,
limit: int = 5,
candidate_limit: int = 20,
strategy: str = "hybrid",
) -> KnowledgeSearchExplain:
with self._session_factory() as session:
statement = select(KnowledgeChunkModel).order_by(KnowledgeChunkModel.created_at.desc())
if knowledge_base_ids:
statement = statement.where(KnowledgeChunkModel.knowledge_base_id.in_(knowledge_base_ids))
rows = session.scalars(statement.limit(candidate_limit)).all()
return _explain_knowledge_chunks(
[_knowledge_chunk_from_model(row) for row in rows],
keyword,
knowledge_base_ids=knowledge_base_ids,
limit=limit,
candidate_limit=candidate_limit,
strategy=strategy,
embedding_provider=self._embedding_provider,
)
def reindex_knowledge_documents(self, payload: KnowledgeReindexRequest) -> KnowledgeReindexResult:
with self._session_factory.begin() as session:
statement = select(KnowledgeDocumentModel).order_by(KnowledgeDocumentModel.created_at.desc())
@@ -1207,8 +1296,7 @@ def _rank_knowledge_chunks(
for chunk in chunks:
if allowed_base_ids and chunk.knowledge_base_id not in allowed_base_ids:
continue
haystack = f"{chunk.title}\n{chunk.reference}\n{chunk.content}".lower()
matched_terms = [term for term in terms if term in haystack]
matched_terms = _matched_terms(chunk, terms)
keyword_score = _score_chunk(chunk, matched_terms, terms_count=len(terms))
vector_score = _vector_score(query_vector, _chunk_embedding(chunk))
if terms and not _passes_retrieval_threshold(
@@ -1217,13 +1305,13 @@ def _rank_knowledge_chunks(
strategy=retrieval_strategy,
):
continue
hybrid_score = round(keyword_score * 0.55 + vector_score * 0.45, 4)
if retrieval_strategy == "keyword":
score = keyword_score
elif retrieval_strategy == "vector":
score = vector_score
else:
score = hybrid_score
hybrid_score = _hybrid_score(keyword_score=keyword_score, vector_score=vector_score)
score = _retrieval_score(
keyword_score=keyword_score,
vector_score=vector_score,
hybrid_score=hybrid_score,
strategy=retrieval_strategy,
)
if terms and score <= 0:
continue
results.append(
@@ -1240,10 +1328,112 @@ def _rank_knowledge_chunks(
return sorted(results, key=lambda item: (item.score, item.chunk.created_at), reverse=True)[:limit]
def _explain_knowledge_chunks(
chunks: list[KnowledgeChunk],
keyword: str,
*,
knowledge_base_ids: list[str] | None = None,
limit: int = 5,
candidate_limit: int = 20,
strategy: str = "hybrid",
embedding_provider: EmbeddingProvider | None = None,
) -> KnowledgeSearchExplain:
allowed_base_ids = set(knowledge_base_ids or [])
retrieval_strategy = strategy if strategy in RETRIEVAL_STRATEGIES else "hybrid"
terms = _tokenize(keyword)
provider = embedding_provider or LocalHashEmbeddingProvider()
query_vector = _embed_query(keyword, provider)
candidates = sorted(chunks, key=lambda item: item.created_at, reverse=True)[:candidate_limit]
items: list[KnowledgeSearchExplainItem] = []
for chunk in candidates:
base_filtered = bool(allowed_base_ids and chunk.knowledge_base_id not in allowed_base_ids)
matched_terms = _matched_terms(chunk, terms) if not base_filtered else []
keyword_score = 0.0 if base_filtered else _score_chunk(chunk, matched_terms, terms_count=len(terms))
vector_score = 0.0 if base_filtered else _vector_score(query_vector, _chunk_embedding(chunk))
hybrid_score = _hybrid_score(keyword_score=keyword_score, vector_score=vector_score)
score = _retrieval_score(
keyword_score=keyword_score,
vector_score=vector_score,
hybrid_score=hybrid_score,
strategy=retrieval_strategy,
)
included = not base_filtered and _passes_explain_threshold(
terms=terms,
score=score,
keyword_score=keyword_score,
vector_score=vector_score,
strategy=retrieval_strategy,
)
items.append(
KnowledgeSearchExplainItem(
chunk=chunk,
score=score,
keyword_score=keyword_score,
vector_score=vector_score,
hybrid_score=hybrid_score,
retrieval_strategy=retrieval_strategy,
matched_terms=matched_terms,
included=included,
filter_reason=_explain_filter_reason(
base_filtered=base_filtered,
terms=terms,
score=score,
keyword_score=keyword_score,
vector_score=vector_score,
strategy=retrieval_strategy,
),
)
)
included_items = sorted(
[item for item in items if item.included],
key=lambda item: (item.score, item.chunk.created_at),
reverse=True,
)
included_ids: set[str] = set()
for rank, item in enumerate(included_items[:limit], start=1):
item.rank = rank
included_ids.add(item.chunk.id)
for item in items:
if item.included and item.chunk.id not in included_ids:
item.included = False
item.filter_reason = "below_limit"
ordered_items = sorted(
items,
key=lambda item: (
1 if item.included else 0,
item.score,
item.chunk.created_at,
),
reverse=True,
)
return KnowledgeSearchExplain(
keyword=keyword,
normalized_terms=terms,
retrieval_strategy=retrieval_strategy,
knowledge_base_ids=knowledge_base_ids or [],
limit=limit,
candidate_limit=candidate_limit,
candidate_count=len(candidates),
included_count=len(included_ids),
embedding_provider=provider.provider_name,
embedding_model=provider.model,
query_embedding_dimension=len(query_vector),
results=ordered_items,
)
def _tokenize(keyword: str) -> list[str]:
return tokenize_embedding_text(keyword)
def _matched_terms(chunk: KnowledgeChunk, terms: list[str]) -> list[str]:
haystack = f"{chunk.title}\n{chunk.reference}\n{chunk.content}".lower()
return [term for term in terms if term in haystack]
def _score_chunk(chunk: KnowledgeChunk, matched_terms: list[str], *, terms_count: int) -> float:
if terms_count == 0:
return 0.1
@@ -1256,6 +1446,24 @@ def _score_chunk(chunk: KnowledgeChunk, matched_terms: list[str], *, terms_count
return round(min(coverage * 0.65 + density * 0.2 + title_bonus + reference_bonus, 1.0), 4)
def _hybrid_score(*, keyword_score: float, vector_score: float) -> float:
return round(keyword_score * 0.55 + vector_score * 0.45, 4)
def _retrieval_score(
*,
keyword_score: float,
vector_score: float,
hybrid_score: float,
strategy: str,
) -> float:
if strategy == "keyword":
return keyword_score
if strategy == "vector":
return vector_score
return hybrid_score
def _passes_retrieval_threshold(
*,
keyword_score: float,
@@ -1269,6 +1477,49 @@ def _passes_retrieval_threshold(
return keyword_score > 0 or vector_score >= MIN_VECTOR_RECALL_SCORE
def _passes_explain_threshold(
*,
terms: list[str],
score: float,
keyword_score: float,
vector_score: float,
strategy: str,
) -> bool:
if terms and score <= 0:
return False
if not terms:
return score > 0
return _passes_retrieval_threshold(
keyword_score=keyword_score,
vector_score=vector_score,
strategy=strategy,
)
def _explain_filter_reason(
*,
base_filtered: bool,
terms: list[str],
score: float,
keyword_score: float,
vector_score: float,
strategy: str,
) -> str | None:
if base_filtered:
return "knowledge_base_filtered"
if not terms:
return None if score > 0 else "empty_query_no_score"
if strategy == "keyword" and keyword_score <= 0:
return "keyword_score_zero"
if strategy == "vector" and vector_score < MIN_VECTOR_RECALL_SCORE:
return "vector_below_threshold"
if strategy == "hybrid" and keyword_score <= 0 and vector_score < MIN_VECTOR_RECALL_SCORE:
return "keyword_and_vector_below_threshold"
if score <= 0:
return "score_zero"
return None
def _embed_query(text: str, embedding_provider: EmbeddingProvider) -> list[float]:
embeddings = embedding_provider.embed_texts([text])
return embeddings[0].vector if embeddings else []
@@ -1285,7 +1536,11 @@ def _vector_score(query_vector: list[float], chunk_vector: list[float]) -> float
return 0.0
length = min(len(query_vector), len(chunk_vector))
dot = sum(query_vector[index] * chunk_vector[index] for index in range(length))
return round(max(dot, 0.0), 4)
query_norm = math.sqrt(sum(value * value for value in query_vector[:length]))
chunk_norm = math.sqrt(sum(value * value for value in chunk_vector[:length]))
if query_norm <= 0 or chunk_norm <= 0:
return 0.0
return round(max(dot / (query_norm * chunk_norm), 0.0), 4)
def _encode_embedding(vector: list[float]) -> str | None:
yuqei-ai-platform/services/ai-platform-api/tests/test_platform_data.py
+53
View File
@@ -194,6 +194,59 @@ def test_repository_uses_injected_embedding_provider_for_chunks_and_search() ->
assert result.vector_score >= 0.35
def test_knowledge_search_explain_returns_scores_rank_and_filter_reason() -> None:
repository = InMemoryAiPlatformRepository(embedding_provider=TestEmbeddingProvider())
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.",
)
)
repository.add_knowledge_document(
KnowledgeDocumentCreate(
knowledge_base_id="enterprise",
title="Employment Policy",
source_type="policy",
reference="EMP-001",
content="Probation period must be approved by HR.",
)
)
app = create_app(AiPlatformSettings(environment="test"), repository=repository)
client = TestClient(app)
response = client.get(
"/api/v1/knowledge-chunks/search/explain",
params={
"keyword": "deposit refund",
"knowledge_base_id": "laws-cn",
"strategy": "hybrid",
"limit": 1,
"candidate_limit": 10,
},
)
assert response.status_code == 200
payload = response.json()
assert payload["keyword"] == "deposit refund"
assert payload["retrieval_strategy"] == "hybrid"
assert payload["embedding_provider"] == "test-embedding"
assert payload["embedding_model"] == "test-embedding-v1"
assert payload["query_embedding_dimension"] == 3
assert payload["candidate_count"] == 2
assert payload["included_count"] == 1
included = next(item for item in payload["results"] if item["included"])
filtered = next(item for item in payload["results"] if not item["included"])
assert included["rank"] == 1
assert included["keyword_score"] > 0
assert included["vector_score"] >= 0.35
assert included["matched_terms"]
assert included["filter_reason"] is None
assert filtered["filter_reason"] == "knowledge_base_filtered"
def test_knowledge_reindex_rebuilds_chunks_with_current_embedding_provider() -> None:
repository = InMemoryAiPlatformRepository()
app = create_app(AiPlatformSettings(environment="test"), repository=repository)
yuqei-ai-sdk-python/src/yuqei_sdk/__init__.py
+4
View File
@@ -8,6 +8,8 @@ from yuqei_sdk.ai_platform import (
KnowledgeImportBatchCreate,
KnowledgeReindexRequest,
KnowledgeReindexResult,
KnowledgeSearchExplain,
KnowledgeSearchExplainItem,
KnowledgeSearchResult,
LegalCitation,
LegalQaRequest,
@@ -30,6 +32,8 @@ __all__ = [
"KnowledgeImportBatchCreate",
"KnowledgeReindexRequest",
"KnowledgeReindexResult",
"KnowledgeSearchExplain",
"KnowledgeSearchExplainItem",
"KnowledgeSearchResult",
"LegalCitation",
"LegalQaRequest",
yuqei-ai-sdk-python/src/yuqei_sdk/ai_platform.py
+59
View File
@@ -123,6 +123,39 @@ class KnowledgeSearchResult(BaseModel):
matched_terms: list[str] = Field(default_factory=list)
class KnowledgeSearchExplainItem(BaseModel):
chunk: KnowledgeChunk
score: float = 0.0
keyword_score: float = 0.0
vector_score: float = 0.0
hybrid_score: float = 0.0
retrieval_strategy: str = "hybrid"
matched_terms: list[str] = Field(default_factory=list)
included: bool = False
rank: int | None = None
filter_reason: str | None = None
class KnowledgeSearchExplain(BaseModel):
keyword: str
normalized_terms: list[str] = Field(default_factory=list)
retrieval_strategy: str = "hybrid"
knowledge_base_ids: list[str] = Field(default_factory=list)
limit: int = 5
candidate_limit: int = 20
candidate_count: int = 0
included_count: int = 0
retrieval_backend: str = "json-explain"
embedding_provider: str
embedding_model: str
query_embedding_dimension: int = 0
keyword_threshold: float = 0.0
vector_threshold: float = 0.35
hybrid_keyword_weight: float = 0.55
hybrid_vector_weight: float = 0.45
results: list[KnowledgeSearchExplainItem] = Field(default_factory=list)
class KnowledgeReindexRequest(BaseModel):
document_id: str | None = None
knowledge_base_id: str | None = None
@@ -368,6 +401,32 @@ class AiPlatformClient:
response.raise_for_status()
return [KnowledgeSearchResult.model_validate(item) for item in response.json()]
def explain_knowledge_search(
self,
keyword: str = "",
*,
knowledge_base_id: str | None = None,
limit: int = 5,
candidate_limit: int = 20,
strategy: str = "hybrid",
context: RequestContext | None = None,
) -> KnowledgeSearchExplain:
params: dict[str, Any] = {
"keyword": keyword,
"limit": limit,
"candidate_limit": candidate_limit,
"strategy": strategy,
}
if knowledge_base_id:
params["knowledge_base_id"] = knowledge_base_id
response = self._client.get(
f"{self._api_prefix}/knowledge-chunks/search/explain",
params=params,
headers=(context or RequestContext()).to_headers(),
)
response.raise_for_status()
return KnowledgeSearchExplain.model_validate(response.json())
def list_ai_run_audits(
self,
*,
yuqei-ai-sdk-python/tests/test_ai_platform.py
+60
View File
@@ -284,6 +284,62 @@ def test_ai_platform_client_covers_admin_data_endpoints() -> None:
}
],
)
if request.url.path == "/api/v1/knowledge-chunks/search/explain":
query = parse_qs(request.url.query.decode())
assert query["keyword"] == ["lease"]
assert query["limit"] == ["3"]
assert query["candidate_limit"] == ["9"]
assert query["strategy"] == ["hybrid"]
return httpx.Response(
200,
json={
"keyword": "lease",
"normalized_terms": ["lease"],
"retrieval_strategy": "hybrid",
"knowledge_base_ids": [],
"limit": 3,
"candidate_limit": 9,
"candidate_count": 1,
"included_count": 1,
"retrieval_backend": "json-explain",
"embedding_provider": "local-hash",
"embedding_model": "local-hash-v1",
"query_embedding_dimension": 64,
"keyword_threshold": 0.0,
"vector_threshold": 0.35,
"hybrid_keyword_weight": 0.55,
"hybrid_vector_weight": 0.45,
"results": [
{
"chunk": {
"id": "chunk-1",
"document_id": "doc-1",
"import_batch_id": "kb-batch-1",
"knowledge_base_id": "laws-cn",
"title": "Civil Code",
"source_type": "law",
"reference": "Article 703",
"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",
},
"score": 0.82,
"keyword_score": 0.76,
"vector_score": 0.89,
"hybrid_score": 0.82,
"retrieval_strategy": "hybrid",
"matched_terms": ["lease"],
"included": True,
"rank": 1,
"filter_reason": None,
}
],
},
)
if request.url.path == "/api/v1/audit/ai-runs":
return httpx.Response(
200,
@@ -369,6 +425,10 @@ def test_ai_platform_client_covers_admin_data_endpoints() -> None:
chunk_result = client.search_knowledge_chunks("lease", limit=3)
assert chunk_result[0].score == 0.82
assert chunk_result[0].hybrid_score == 0.82
explain = client.explain_knowledge_search("lease", limit=3, candidate_limit=9)
assert explain.embedding_provider == "local-hash"
assert explain.results[0].included is True
assert explain.results[0].rank == 1
audit = client.list_ai_run_audits(limit=5)[0]
assert audit.operation == "legal_qa"
assert audit.retrieved_chunk_ids == ["chunk-1"]