Files

3136 lines
104 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from __future__ import annotations
import json
from types import SimpleNamespace
from typing import Any
import pytest
from fastapi import HTTPException
from langchain_core.messages import HumanMessage, ToolMessage
from openpyxl import load_workbook
from app.gateway.routers.llmwiki import _build_wiki_excel
from deerflow.integrations.weknora.local_index.retrieval import WikiRetrievalService, is_fully_vectorized
from deerflow.integrations.weknora.local_index.search import WikiIndexError
class _Store:
def __init__(self, states):
self.states = states
async def list_index_status(self, mapping_ids=None):
wanted = set(mapping_ids or [])
return [state for state in self.states if not wanted or state["knowledge_base_mapping_id"] in wanted]
class _WikiClient:
def __init__(self):
self.list_calls = []
self.raw_search_called = False
async def list_wiki_pages(self, knowledge_base_id, *, page=1, page_size=50, query=""):
self.list_calls.append((knowledge_base_id, query))
return {"pages": [{"slug": "guides/install", "title": "安装"}], "total": 1, "total_pages": 1}
async def get_wiki_page(self, knowledge_base_id, slug):
return {
"id": f"{knowledge_base_id}:{slug}",
"slug": slug,
"title": "安装",
"summary": "Wiki 摘要",
"content": "# 安装\n这是处理后的 Wiki 正文。",
"status": "published",
"category_path": ["指南"],
}
async def search(self, query, knowledge_base_ids):
self.raw_search_called = True
raise AssertionError("raw knowledge search must never be called")
class _OverviewWikiClient(_WikiClient):
async def list_wiki_pages(self, knowledge_base_id, *, page=1, page_size=50, query=""):
self.list_calls.append((knowledge_base_id, query))
if query:
return {"pages": [], "total": 0, "total_pages": 0}
return {
"pages": [
{"slug": "index", "title": "目录", "page_type": "directory"},
{"slug": "guides/cooling", "title": "设备散热", "page_type": "article"},
],
"total": 2,
"total_pages": 1,
}
async def get_wiki_page(self, knowledge_base_id, slug):
if slug == "index":
return {
"id": f"{knowledge_base_id}:{slug}",
"slug": slug,
"title": "目录",
"page_type": "directory",
"content": "目录正文",
"status": "published",
}
return {
"id": f"{knowledge_base_id}:{slug}",
"slug": slug,
"title": "设备散热",
"page_type": "article",
"summary": "页面摘要",
"content": "设备过热时应先断电,再检查风道和散热风扇。",
"status": "published",
}
class _VectorSearch:
def __init__(self):
self.embedding = SimpleNamespace(fingerprint="fp-current")
self.calls = []
async def search(self, query, mappings, **kwargs):
self.calls.append([mapping["id"] for mapping in mappings])
mapping = mappings[0]
return {
"results": [
{
"kind": "wiki_page",
"id": "local-page",
"wiki_page_id": "local-page",
"wiki_slug": "local/page",
"knowledge_base_id": mapping["id"],
"title": "本地页面",
"content": "本地 Wiki",
"matched_sections": [],
"score": 0.95,
}
],
"warnings": [],
}
def _ready_state(mapping_id: str):
return {
"knowledge_base_mapping_id": mapping_id,
"state": "idle",
"last_success_at": "2026-08-24T00:00:00Z",
"embedding_fingerprint": "fp-current",
"local_page_count": 1,
"ready_page_count": 1,
"failed_page_count": 0,
"vector_count": 3,
}
def test_fully_vectorized_requires_completed_full_consistent_scan() -> None:
state = _ready_state("kb")
assert is_fully_vectorized(state, "fp-current")
assert is_fully_vectorized({**state, "state": "partial"}, "fp-current")
assert is_fully_vectorized({**state, "state": "failed"}, "fp-current")
assert is_fully_vectorized({**state, "state": "failed", "failed_page_count": 1}, "fp-current")
assert not is_fully_vectorized({**state, "state": "rebuild_required"}, "fp-current")
assert not is_fully_vectorized({**state, "ready_page_count": 0}, "fp-current")
assert not is_fully_vectorized(state, "different-fingerprint")
class _KeywordStore(_Store):
async def search_pages_by_keywords(self, mapping_ids, tokens, *, limit, include_drafts):
assert mapping_ids == ["ready"]
assert "散热" in tokens
return [
{
"id": "page-1",
"knowledge_base_mapping_id": "ready",
"slug": "guides/cooling",
"title": "设备散热",
"summary": "散热指南",
"content_md": "设备过热时应先断电并检查风道。",
"page_type": "article",
"status": "published",
}
]
class _EmptyVectorSearch(_VectorSearch):
async def search(self, query, mappings, **kwargs):
self.calls.append([mapping["id"] for mapping in mappings])
return {
"results": [],
"warnings": [],
"diagnostics": {"top_raw_vector_score": 0.21, "threshold_filtered": 7},
}
class _KeywordCandidateStore(_Store):
def __init__(self, pages):
super().__init__([_ready_state("ready")])
self.pages = pages
self.tokens: list[str] = []
async def search_pages_by_keywords(self, mapping_ids, tokens, *, limit, include_drafts):
assert mapping_ids == ["ready"]
self.tokens = list(tokens)
return self.pages[:limit]
@pytest.mark.asyncio
async def test_keyword_channel_recovers_when_vectors_are_below_threshold() -> None:
service = WikiRetrievalService(
_KeywordStore([_ready_state("ready")]),
vector_search=_EmptyVectorSearch(),
client=None,
)
result = await service.search("散热故障怎么处理", [{"id": "ready", "name": "设备库"}])
assert result["mode"] == "keyword"
assert result["results"][0]["title"] == "设备散热"
assert result["results"][0]["retrieval_mode"] == "keyword"
assert result["diagnostics"]["fallback_reason"] == "VECTOR_BELOW_THRESHOLD"
@pytest.mark.asyncio
async def test_keyword_channel_ignores_english_stop_words_and_single_weak_term() -> None:
store = _KeywordCandidateStore(
[
{
"id": "page-1",
"knowledge_base_mapping_id": "ready",
"slug": "controller",
"title": "Controller overview",
"summary": "Annual inspection",
"content_md": "This controller is on standby.",
"page_type": "article",
"status": "published",
}
]
)
service = WikiRetrievalService(store, vector_search=_EmptyVectorSearch(), client=None)
result = await service.search(
"What is the annual rainfall on Neptune north pole?",
[{"id": "ready", "name": "Acceptance"}],
)
assert "what" not in store.tokens
assert "is" not in store.tokens
assert "the" not in store.tokens
assert "on" not in store.tokens
assert result["results"] == []
assert result["diagnostics"]["keyword_candidates"] == 0
assert result["mode"] == "vector"
@pytest.mark.asyncio
async def test_keyword_channel_keeps_precise_code_as_single_strong_match() -> None:
store = _KeywordCandidateStore(
[
{
"id": "page-1",
"knowledge_base_mapping_id": "ready",
"slug": "xuanwu-pump",
"title": "XW-900 maintenance",
"summary": "Reset procedure",
"content_md": "Fault ZXQ-417 requires a ninety second wait.",
"page_type": "article",
"status": "published",
}
]
)
service = WikiRetrievalService(store, vector_search=_EmptyVectorSearch(), client=None)
result = await service.search(
"How do I reset fault ZXQ-417?",
[{"id": "ready", "name": "Acceptance"}],
)
assert result["mode"] == "keyword"
assert result["results"][0]["wiki_slug"] == "xuanwu-pump"
assert result["results"][0]["retrieval_mode"] == "keyword"
@pytest.mark.asyncio
async def test_unvectorized_library_uses_only_wiki_page_search_api() -> None:
client = _WikiClient()
service = WikiRetrievalService(_Store([]), vector_search=_VectorSearch(), client=client)
mapping = {"id": "local-kb", "weknora_id": "remote-kb", "name": "测试库"}
result = await service.search("安装", [mapping])
assert result["mode"] == "wiki_api"
assert result["fallback_knowledge_base_ids"] == ["local-kb"]
assert result["results"][0]["kind"] == "wiki_page"
assert result["results"][0]["content"] == "# 安装\n这是处理后的 Wiki 正文。"
assert "chunk_id" not in result["results"][0]
assert client.list_calls == [("remote-kb", "安装")]
assert not client.raw_search_called
@pytest.mark.asyncio
async def test_mixed_search_vectors_complete_library_and_falls_back_incomplete_library() -> None:
client = _WikiClient()
vector = _VectorSearch()
service = WikiRetrievalService(_Store([_ready_state("ready")]), vector_search=vector, client=client)
mappings = [
{"id": "ready", "weknora_id": "remote-ready", "name": "已完成"},
{"id": "pending", "weknora_id": "remote-pending", "name": "未完成"},
]
result = await service.search("安装", mappings, top_k_pages=8)
assert result["mode"] == "mixed"
assert vector.calls == [["ready"]]
assert client.list_calls == [("remote-pending", "安装")]
assert {item["knowledge_base_id"] for item in result["results"]} == {"ready", "pending"}
@pytest.mark.asyncio
async def test_overview_query_lists_representative_remote_wiki_bodies_without_query_search() -> None:
client = _OverviewWikiClient()
service = WikiRetrievalService(_Store([]), vector_search=_VectorSearch(), client=client)
mapping = {"id": "local-kb", "weknora_id": "remote-kb", "name": "设备库"}
result = await service.search("这个知识空间有哪些内容?", [mapping])
assert result["mode"] == "overview"
assert client.list_calls == [("remote-kb", "")]
assert [item["wiki_slug"] for item in result["results"]] == ["guides/cooling"]
assert result["results"][0]["retrieval_mode"] == "overview"
assert result["results"][0]["content"].startswith("设备过热时应先断电")
def test_excel_export_contains_current_directory_and_every_parent_directory() -> None:
long_content = "正文" * 20_000
pages = [
{"id": "1", "slug": "root", "title": "根专题", "content": "根", "status": "published"},
{"id": "2", "slug": "root/child", "title": "子专题", "parent_slug": "root", "content": "子", "status": "published"},
{
"id": "3",
"slug": "root/child/article",
"title": "文章",
"parent_slug": "root/child",
"content": long_content,
"status": "published",
},
]
workbook = load_workbook(_build_wiki_excel("知识库", pages), read_only=True)
sheet = workbook["Wiki 全量导出"]
rows = list(sheet.iter_rows(values_only=True))
headings = list(rows[0])
article = dict(zip(headings, rows[3], strict=True))
assert article["目录完整路径"] == "根专题 / 子专题"
assert article["所在目录"] == "子专题"
assert article["父级目录完整路径"] == "根专题"
assert article["第1级目录"] == "根专题"
assert article["第2级目录"] == "子专题"
assert article["正文"] + article["正文(续1)"] == long_content
class _TraceMappingStore:
def __init__(self, rows: list[dict[str, Any]]) -> None:
self.rows = {str(row["id"]): dict(row) for row in rows}
self.authorized_calls: list[tuple[str, str, bool, bool]] = []
async def get_authorized(
self,
mapping_id: str,
user_id: str,
*,
write: bool,
is_admin: bool,
) -> dict[str, Any] | None:
self.authorized_calls.append((mapping_id, user_id, write, is_admin))
row = self.rows.get(mapping_id)
return dict(row) if row is not None else None
class _TraceRawClient:
def __init__(
self,
rows: list[dict[str, Any]] | None = None,
*,
error: Exception | None = None,
) -> None:
self.rows = rows or []
self.error = error
self.calls: list[tuple[str, list[str]]] = []
async def search(self, query: str, knowledge_base_ids: list[str]) -> list[dict[str, Any]]:
self.calls.append((query, list(knowledge_base_ids)))
if self.error is not None:
raise self.error
return [dict(row) for row in self.rows]
class _TraceGraphStore:
def __init__(
self,
batches: list[list[dict[str, Any]]] | None = None,
*,
error: Exception | None = None,
) -> None:
self.batches = batches or []
self.error = error
self.calls: list[dict[str, Any]] = []
async def get_adjacent_pages(
self,
mapping_ids: list[str],
seed_slugs: dict[str, list[str]],
*,
direction: str,
limit: int,
include_drafts: bool,
) -> list[dict[str, Any]]:
self.calls.append(
{
"mapping_ids": list(mapping_ids),
"seed_slugs": {key: list(value) for key, value in seed_slugs.items()},
"direction": direction,
"limit": limit,
"include_drafts": include_drafts,
}
)
if self.error is not None:
raise self.error
index = len(self.calls) - 1
rows = self.batches[index] if index < len(self.batches) else []
return [dict(row) for row in rows]
class _TraceRetrievalService:
def __init__(
self,
payload: dict[str, Any],
*,
client: _TraceRawClient | None = None,
error: Exception | None = None,
) -> None:
self.payload = payload
self.client = client
self.error = error
self.calls: list[dict[str, Any]] = []
async def search(
self,
query: str,
mappings: list[dict[str, Any]],
*,
top_k_pages: int,
) -> dict[str, Any]:
self.calls.append(
{
"query": query,
"mapping_ids": [str(row["id"]) for row in mappings],
"top_k_pages": top_k_pages,
}
)
if self.error is not None:
raise self.error
return dict(self.payload)
class _TraceAssistantKnowledgeStore:
def __init__(self, rows: list[dict[str, Any]]) -> None:
self.rows = rows
self.calls: list[dict[str, Any]] = []
async def search_pages(self, **kwargs: Any) -> list[dict[str, Any]]:
self.calls.append(dict(kwargs))
return [dict(row) for row in self.rows]
class _TraceAssistantGraphStore(_TraceAssistantKnowledgeStore):
def __init__(
self,
rows: list[dict[str, Any]],
*,
bases: list[dict[str, Any]],
graph_batches: list[list[dict[str, Any]]],
) -> None:
super().__init__(rows)
self.bases = {str(base["id"]): dict(base) for base in bases}
self.graph = _TraceGraphStore(graph_batches)
async def get_base(self, base_id: str) -> dict[str, Any] | None:
base = self.bases.get(base_id)
return dict(base) if base is not None else None
async def get_adjacent_pages(
self,
base_ids: list[str],
seed_pages: dict[str, list[str]],
*,
direction: str,
limit: int,
include_drafts: bool,
) -> list[dict[str, Any]]:
return await self.graph.get_adjacent_pages(
base_ids,
seed_pages,
direction=direction,
limit=limit,
include_drafts=include_drafts,
)
def _trace_request(
*,
mapping_store: _TraceMappingStore,
retrieval_service: _TraceRetrievalService,
assistant_store: _TraceAssistantKnowledgeStore | None = None,
graph_store: _TraceGraphStore | None = None,
) -> SimpleNamespace:
local_index = SimpleNamespace(enabled=True, internal_search_enabled=True)
state = SimpleNamespace(
config=SimpleNamespace(
llmwiki=SimpleNamespace(local_wiki_index=local_index),
),
llmwiki_store=mapping_store,
llmwiki_retrieval_service=retrieval_service,
llmwiki_index_store=graph_store,
assistant_knowledge_store=assistant_store,
llmwiki_embedding=SimpleNamespace(fingerprint="fp-current"),
)
return SimpleNamespace(app=SimpleNamespace(state=state))
def _trace_body(
*,
llmwiki_ids: list[str] | None = None,
assistant_ids: list[str] | None = None,
) -> SimpleNamespace:
return SimpleNamespace(
assistant_id=None,
context={
"llmwiki_knowledge_base_ids": llmwiki_ids or [],
"assistant_knowledge_base_ids": assistant_ids or [],
},
metadata={},
)
def _selected_spaces_json(trace: dict[str, Any]) -> str:
"""Allow space items to grow without fixing their exact representation."""
return json.dumps(trace["selected_spaces"], ensure_ascii=False, sort_keys=True)
def _assert_non_negative_timings(timings: dict[str, Any]) -> None:
assert timings
for value in timings.values():
if isinstance(value, (int, float)):
assert value >= 0
def _ordinary_trace_results(count: int, *, knowledge_base_id: str = "kb-mixed") -> list[dict[str, Any]]:
return [
{
"kind": "wiki_page",
"id": f"ordinary-{index}",
"wiki_page_id": f"ordinary-{index}",
"wiki_slug": f"ordinary/{index}",
"knowledge_base_id": knowledge_base_id,
"knowledge_base_name": "普通 Wiki",
"title": f"普通结果 {index}",
"content": f"普通知识内容 {index}",
"score": 0.01 * index,
"retrieval_mode": "vector",
"matched_sections": [
{
"heading": f"普通章节 {index}",
"content": f"普通知识内容 {index}",
"score": 0.9 - index / 100,
}
],
}
for index in range(1, count + 1)
]
def _graph_row(
*,
mapping_id: str,
seed_slug: str,
neighbor_slug: str,
title: str,
content: str,
direction: str = "outgoing",
edge_type: str = "wikilink",
status: str = "published",
) -> dict[str, Any]:
source_slug, target_slug = (
(seed_slug, neighbor_slug)
if direction == "outgoing"
else (neighbor_slug, seed_slug)
)
return {
"knowledge_base_mapping_id": mapping_id,
"seed_slug": seed_slug,
"neighbor_slug": neighbor_slug,
"direction": direction,
"edge_type": edge_type,
"source_slug": source_slug,
"target_slug": target_slug,
"page": {
"id": f"page:{mapping_id}:{neighbor_slug}",
"knowledge_base_mapping_id": mapping_id,
"slug": neighbor_slug,
"title": title,
"summary": f"{title}摘要",
"content_md": content,
"page_type": "article",
"status": status,
"source_refs": [{"chunk_id": f"chunk:{neighbor_slug}"}],
},
}
def _assistant_graph_row(
*,
base_id: str,
seed_slug: str,
neighbor_slug: str,
title: str,
content: str,
) -> dict[str, Any]:
return {
"base_id": base_id,
"seed_slug": seed_slug,
"neighbor_slug": neighbor_slug,
"direction": "outgoing",
"edge_type": "references",
"page": {
"id": f"assistant-page:{base_id}:{neighbor_slug}",
"base_id": base_id,
"slug": neighbor_slug,
"title": title,
"summary": f"{title}摘要",
"content_markdown": content,
"page_type": "article",
"status": "active",
},
}
def _assistant_trace_results(count: int, *, base_id: str = "assistant-mixed") -> list[dict[str, Any]]:
return [
{
"id": f"assistant-{index}",
"base_id": base_id,
"slug": f"assistant/{index}",
"title": f"助手结果 {index}",
"content": f"助手知识内容 {index}",
"score": 0.8 - index / 100,
"retrieval_mode": "vector",
"matched_section": {
"heading": f"助手章节 {index}",
"content": f"助手知识内容 {index}",
"score": 0.8 - index / 100,
},
}
for index in range(1, count + 1)
]
def test_local_wiki_section_source_is_bounded_safe_and_vector_diagnostic() -> None:
from app.gateway import llmwiki_rag
source = llmwiki_rag._source_from_result(
{
"kind": "wiki_section",
"id": "page-cooling:0",
"wiki_page_id": "page-cooling",
"weknora_knowledge_base_id": "remote-equipment-secret",
"weknora_wiki_page_id": "remote-page-secret",
"knowledge_base_id": "kb-equipment",
"wiki_slug": "guides/cooling",
"title": "设备散热",
"heading": "故障处理",
"content": "设备过热时应先断电。" + "检修步骤" * 500,
"page_content": "不应持久化的完整页面" * 1000,
"score": 0.016,
"channel_score": 0.82,
"rank_score_semantics": "rrf",
"retrieval_mode": "vector",
"source_refs": [
{
"knowledge_id": "doc-one",
"knowledge_title": "设备手册.pdf",
"internal_metadata": "must-not-leak",
}
],
"chunk_refs": ["chunk-one"],
},
{"id": "kb-equipment", "name": "设备知识库"},
1,
)
assert source["type"] == "llmwiki_wiki"
assert len(source["content"]) <= 1400
assert "knowledge_base_id=kb-equipment" in source["url"]
assert "wiki_slug=guides%2Fcooling" in source["url"]
serialized = json.dumps(source, ensure_ascii=False)
assert "remote-equipment-secret" not in serialized
assert "remote-page-secret" not in serialized
assert "不应持久化的完整页面" not in serialized
assert "internal_metadata" not in serialized
trace_result = llmwiki_rag._trace_result(source)
assert trace_result["knowledge_base_id"] == "kb-equipment"
assert trace_result["matched_heading"] == "故障处理"
assert trace_result["vector_score"] == pytest.approx(0.82)
assert trace_result["rank_score_semantics"] == "rrf"
def test_prompt_keeps_partial_retrieval_diagnostics_internal_when_sources_exist() -> None:
from app.gateway import llmwiki_rag
prompt = llmwiki_rag._build_prompt(
"海湖庄园",
[
{
"index": 1,
"id": "wiki-page",
"title": "海湖庄园",
"type": "llmwiki_wiki",
"content": "格式化后的 Wiki 内容",
"source": "测试知识空间",
"retrieval_mode": "vector",
"raw": {
"knowledge_base_id": "kb-test",
"wiki_slug": "wiki/mar-a-lago",
},
}
],
{
"partial": True,
"warnings": ["VECTOR_UNAVAILABLE"],
"fallback_reason": "VECTOR_UNAVAILABLE",
},
)
assert "VECTOR_UNAVAILABLE" not in prompt
assert "只有部分检索通道成功" not in prompt
assert "明确说明检索结果可能不完整" not in prompt
assert "不要在回答中展示检索通道状态" in prompt
@pytest.mark.asyncio
async def test_rag_context_exposes_successful_hybrid_retrieval_trace(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
mapping_store = _TraceMappingStore(
[
{
"id": "kb-equipment",
"weknora_id": "remote-equipment",
"name": "设备知识库",
"owner_user_id": "user-a",
}
]
)
retrieval = _TraceRetrievalService(
{
"query": "设备散热故障怎么处理?",
"mode": "hybrid",
"results": [
{
"kind": "wiki_page",
"id": "page-cooling",
"wiki_page_id": "page-cooling",
"wiki_slug": "guides/cooling",
"knowledge_base_id": "kb-equipment",
"knowledge_base_name": "设备知识库",
"title": "设备散热",
"content": "设备过热时应先断电,再检查风道和散热风扇。",
"snippet": "设备过热时应先断电",
"matched_sections": [
{
"section_index": 0,
"heading": "故障处理",
"content": "设备过热时应先断电,再检查风道和散热风扇。",
"score": 0.82,
}
],
"score": 0.032,
"channel_score": 0.82,
"rank_score_semantics": "rrf",
"retrieval_mode": "hybrid",
"url": "/weknora-source-preview?knowledge_base_id=kb-equipment",
}
],
"knowledge_base_count": 1,
"partial": False,
"warnings": [],
"fallback_knowledge_base_ids": [],
"diagnostics": {
"top_raw_vector_score": 0.82,
"min_similarity": 0.35,
"threshold_filtered": 0,
"vector_candidates": 1,
"keyword_candidates": 1,
"wiki_api_candidates": 0,
"fallback_reason": None,
},
}
)
request = _trace_request(mapping_store=mapping_store, retrieval_service=retrieval)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=True),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(llmwiki_ids=["kb-equipment"]),
graph_input={
"messages": [
HumanMessage(
id="human-one",
content="设备散热故障怎么处理?",
)
]
},
)
assert retrieval.calls == [
{
"query": "设备散热故障怎么处理?",
"mapping_ids": ["kb-equipment"],
"top_k_pages": 8,
}
]
assert mapping_store.authorized_calls == [("kb-equipment", "user-a", False, False)]
assert rag.source_count == 1
assert rag.sources[0]["raw"]["wiki_slug"] == "guides/cooling"
trace = rag.trace
assert trace["retrieval_trace_id"]
assert trace["trigger"] == "pre_retrieval"
assert trace["query"] == "设备散热故障怎么处理?"
assert "kb-equipment" in _selected_spaces_json(trace)
assert trace["authorized_space_count"] == 1
assert trace["mode"] == "hybrid"
assert trace["source_count"] == 1
assert trace["partial"] is False
assert trace["warnings"] == []
assert trace["fallback_reason"] is None
assert trace["fallback_knowledge_base_ids"] == []
assert trace["diagnostics"]["top_raw_vector_score"] == pytest.approx(0.82)
assert trace["diagnostics"]["vector_candidates"] == 1
assert trace["results"][0]["knowledge_base_id"] == "kb-equipment"
assert trace["results"][0]["wiki_slug"] == "guides/cooling"
# Hybrid uses an RRF/channel score; the section score is not guaranteed
# to be cosine similarity and must not be exposed as vector_score.
assert trace["results"][0]["rank_score_semantics"] == "rrf"
assert trace["results"][0]["vector_score"] is None
assert trace["zero_result_reason"] is None
assert isinstance(trace["capabilities"], (dict, list))
_assert_non_negative_timings(trace["timings_ms"])
@pytest.mark.asyncio
async def test_rag_context_fuses_authorized_raw_chunks_with_wiki_results(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
mapping_store = _TraceMappingStore(
[
{
"id": "kb-equipment",
"weknora_id": "remote-equipment",
"name": "设备知识库",
"owner_user_id": "user-a",
}
]
)
long_content = "TIGER-928 关闭旁通阀后等待 75 秒,再长按绿色复位键 6 秒。" + "检修说明" * 400
raw_client = _TraceRawClient(
[
{
"chunk_id": "chunk-tiger",
"knowledge_id": "doc-tiger",
"knowledge_base_id": "remote-equipment",
"title": "星河泵站编码模型验收",
"content": long_content,
"score": 0.91,
"chunk_index": 0,
"metadata": {"must_not_leak": "internal"},
},
{
"chunk_id": "chunk-other",
"knowledge_id": "doc-other",
"knowledge_base_id": "remote-not-authorized",
"title": "越权片段",
"content": "SECRET-OTHER-KB",
"score": 0.99,
},
]
)
ordinary_results = _ordinary_trace_results(8, knowledge_base_id="kb-equipment")
ordinary_results[0]["weknora_knowledge_base_id"] = "remote-equipment"
ordinary_results[0]["weknora_wiki_page_id"] = "remote-page-secret"
retrieval = _TraceRetrievalService(
{
"mode": "vector",
"results": ordinary_results,
"partial": False,
"warnings": [],
"fallback_knowledge_base_ids": [],
"diagnostics": {"vector_candidates": 8},
},
client=raw_client,
)
request = _trace_request(mapping_store=mapping_store, retrieval_service=retrieval)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=True),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(llmwiki_ids=["kb-equipment"]),
graph_input={"messages": [HumanMessage(content="TIGER-928 告警后如何复位?")]},
)
assert raw_client.calls == [("TIGER-928 告警后如何复位?", ["remote-equipment"])]
assert rag.source_count == 8
assert rag.sources[0]["id"] == "chunk-tiger"
assert len(rag.sources[0]["content"]) <= 1400
assert "TIGER-928" in rag.prompt
assert "SECRET-OTHER-KB" not in rag.prompt
assert "SECRET-OTHER-KB" not in json.dumps(rag.trace, ensure_ascii=False)
assert "metadata" not in rag.sources[0]["raw"]
assert "remote-equipment" not in json.dumps(rag.sources, ensure_ascii=False)
assert "remote-equipment" not in json.dumps(rag.trace, ensure_ascii=False)
assert "remote-page-secret" not in json.dumps(rag.sources, ensure_ascii=False)
assert "remote-page-secret" not in json.dumps(rag.trace, ensure_ascii=False)
trace = rag.trace
assert trace["mode"] == "wiki_raw_hybrid"
assert trace["capabilities"]["raw_document_chunk_retrieval"] is True
assert trace["diagnostics"]["raw_chunk_candidates"] == 2
assert trace["diagnostics"]["raw_chunk_considered"] == 1
assert trace["diagnostics"]["raw_chunk_rejected"] == 1
assert trace["diagnostics"]["raw_chunk_top_provider_score"] == pytest.approx(0.91)
assert trace["diagnostics"]["final_raw_chunk_source_count"] >= 1
assert trace["diagnostics"]["final_wiki_page_source_count"] >= 1
assert {source["type"] for source in rag.sources} >= {"llmwiki", "llmwiki_wiki"}
assert trace["results"][0]["retrieval_mode"] == "raw_chunk"
assert trace["results"][0]["provider_score"] == pytest.approx(0.91)
assert trace["results"][0]["rank_score_semantics"] == "rrf"
assert trace["results"][0]["vector_score"] is None
@pytest.mark.asyncio
async def test_raw_chunk_hit_survives_wiki_channel_failure(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
mapping_store = _TraceMappingStore(
[
{
"id": "kb-equipment",
"weknora_id": "remote-equipment",
"name": "设备知识库",
"owner_user_id": "user-a",
}
]
)
raw_client = _TraceRawClient(
[
{
"chunk_id": "chunk-tiger",
"knowledge_id": "doc-tiger",
"knowledge_base_id": "remote-equipment",
"title": "复位规程",
"content": "TIGER-928:等待 75 秒并长按绿色复位键 6 秒。",
"score": 0.73,
}
]
)
retrieval = _TraceRetrievalService(
{},
client=raw_client,
error=WikiIndexError("WIKI_API_UNAVAILABLE", "wiki unavailable", status_code=502),
)
request = _trace_request(mapping_store=mapping_store, retrieval_service=retrieval)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=True),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(llmwiki_ids=["kb-equipment"]),
graph_input={"messages": [HumanMessage(content="TIGER-928 如何复位?")]},
)
assert rag.source_count == 1
assert rag.sources[0]["id"] == "chunk-tiger"
assert rag.trace["mode"] == "raw_chunk"
assert rag.trace["partial"] is True
assert "WIKI_API_UNAVAILABLE" in rag.trace["warnings"]
assert rag.trace["zero_result_reason"] is None
assert "TIGER-928" in rag.prompt
assert "不能判断库内是否存在相关资料" not in rag.prompt
@pytest.mark.asyncio
async def test_raw_chunk_failure_preserves_wiki_results(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
mapping_store = _TraceMappingStore(
[
{
"id": "kb-equipment",
"weknora_id": "remote-equipment",
"name": "设备知识库",
"owner_user_id": "user-a",
}
]
)
retrieval = _TraceRetrievalService(
{
"mode": "keyword",
"results": _ordinary_trace_results(1, knowledge_base_id="kb-equipment"),
"partial": False,
"warnings": [],
"fallback_knowledge_base_ids": [],
"diagnostics": {"keyword_candidates": 1},
},
client=_TraceRawClient(error=RuntimeError("raw unavailable")),
)
request = _trace_request(mapping_store=mapping_store, retrieval_service=retrieval)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=True),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(llmwiki_ids=["kb-equipment"]),
graph_input={"messages": [HumanMessage(content="普通 Wiki 命中")]} ,
)
assert rag.source_count == 1
assert rag.sources[0]["id"] == "ordinary-1"
assert rag.trace["mode"] == "keyword"
assert rag.trace["partial"] is True
assert "RAW_CHUNK_RETRIEVAL_FAILED" in rag.trace["warnings"]
@pytest.mark.asyncio
async def test_rag_expands_authorized_wiki_graph_with_bounded_per_library_quota(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
mapping_store = _TraceMappingStore(
[
{
"id": "kb-equipment",
"weknora_id": "remote-equipment",
"name": "设备知识库",
"owner_user_id": "user-a",
}
]
)
graph_rows = [
_graph_row(
mapping_id="kb-equipment",
seed_slug="ordinary/1",
neighbor_slug=f"related/{index}",
title=f"关联页面 {index}",
content=f"关联页面 {index} 的真实正文和操作依据。",
)
for index in range(1, 6)
]
graph_rows.extend(
[
_graph_row(
mapping_id="kb-not-authorized",
seed_slug="ordinary/1",
neighbor_slug="secret/page",
title="越权页面",
content="SECRET-GRAPH-CONTENT",
),
_graph_row(
mapping_id="kb-equipment",
seed_slug="ordinary/1",
neighbor_slug="empty/page",
title="空正文",
content="",
),
]
)
graph_store = _TraceGraphStore([graph_rows])
retrieval = _TraceRetrievalService(
{
"mode": "hybrid",
"results": _ordinary_trace_results(1, knowledge_base_id="kb-equipment"),
"partial": False,
"warnings": [],
"fallback_knowledge_base_ids": [],
"diagnostics": {
"vector_candidates": 1,
"keyword_candidates": 1,
"top_raw_vector_score": 0.82,
"min_similarity": 0.35,
},
}
)
request = _trace_request(
mapping_store=mapping_store,
retrieval_service=retrieval,
graph_store=graph_store,
)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=False),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(llmwiki_ids=["kb-equipment"]),
graph_input={"messages": [HumanMessage(content="泵站故障还关联哪些操作?")]},
)
assert graph_store.calls == [
{
"mapping_ids": ["kb-equipment"],
"seed_slugs": {"kb-equipment": ["ordinary/1"]},
"direction": "both",
"limit": 24,
"include_drafts": False,
}
]
assert rag.trace["mode"] == "wiki_graph_hybrid"
assert rag.trace["capabilities"]["wiki_graph_expansion"] is True
assert rag.trace["diagnostics"]["graph_seed_count"] == 1
assert rag.trace["diagnostics"]["graph_candidates"] == 7
assert rag.trace["diagnostics"]["graph_unique_candidates"] == 5
assert rag.trace["diagnostics"]["graph_accepted"] == 3
assert rag.trace["diagnostics"]["graph_sources_used"] == 2
assert rag.trace["diagnostics"]["graph_rejected"] == 2
assert rag.trace["diagnostics"]["graph_hops_requested"] == 1
assert rag.trace["diagnostics"]["graph_hops_executed"] == 1
assert rag.trace["timings_ms"]["graph_expansion"] >= 0
assert "SECRET-GRAPH-CONTENT" not in rag.prompt
assert "SECRET-GRAPH-CONTENT" not in json.dumps(rag.trace, ensure_ascii=False)
graph_results = [
result for result in rag.trace["results"]
if result["retrieval_mode"] == "graph_expansion"
]
assert len(graph_results) == 2
assert graph_results[0]["graph_hop"] == 1
assert graph_results[0]["graph_score"] == pytest.approx(0.58)
assert graph_results[0]["graph_path"] == [
{"slug": "ordinary/1", "title": "普通结果 1"},
{
"slug": "related/1",
"title": "关联页面 1",
"direction": "outgoing",
"edge_type": "wikilink",
},
]
assert graph_results[0]["source_refs"][0]["chunk_id"] == "chunk:related/1"
@pytest.mark.asyncio
async def test_rag_graph_expansion_uses_two_hops_only_for_explicit_or_low_confidence_queries(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
mapping_store = _TraceMappingStore(
[
{
"id": "kb-equipment",
"weknora_id": "remote-equipment",
"name": "设备知识库",
"owner_user_id": "user-a",
}
]
)
graph_store = _TraceGraphStore(
[
[
_graph_row(
mapping_id="kb-equipment",
seed_slug="ordinary/1",
neighbor_slug="nodes/middle",
title="中继控制站",
content="中继控制站连接上游控制中心。",
)
],
[
_graph_row(
mapping_id="kb-equipment",
seed_slug="nodes/middle",
neighbor_slug="nodes/upstream",
title="上游控制中心",
content="上游控制中心交接口令为 CINDER-884。",
)
],
]
)
retrieval = _TraceRetrievalService(
{
"mode": "vector",
"results": _ordinary_trace_results(1, knowledge_base_id="kb-equipment"),
"partial": False,
"warnings": [],
"fallback_knowledge_base_ids": [],
"diagnostics": {
"vector_candidates": 1,
"top_raw_vector_score": 0.74,
"min_similarity": 0.35,
},
}
)
request = _trace_request(
mapping_store=mapping_store,
retrieval_service=retrieval,
graph_store=graph_store,
)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=False),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(llmwiki_ids=["kb-equipment"]),
graph_input={"messages": [HumanMessage(content="沿完整链路做多跳追溯,找上游交接口令")]},
)
assert len(graph_store.calls) == 2
assert graph_store.calls[0]["seed_slugs"] == {"kb-equipment": ["ordinary/1"]}
assert graph_store.calls[0]["limit"] == 12
assert graph_store.calls[1]["seed_slugs"] == {"kb-equipment": ["nodes/middle"]}
assert graph_store.calls[1]["limit"] == 12
assert rag.trace["diagnostics"]["graph_hops_requested"] == 2
assert rag.trace["diagnostics"]["graph_hops_executed"] == 2
assert rag.trace["diagnostics"]["graph_depth_reason"] == "explicit_multi_hop"
upstream = next(
result for result in rag.trace["results"]
if result["wiki_slug"] == "nodes/upstream"
)
assert upstream["graph_hop"] == 2
assert upstream["graph_score"] == pytest.approx(0.58 * 0.58)
assert [node["slug"] for node in upstream["graph_path"]] == [
"ordinary/1",
"nodes/middle",
"nodes/upstream",
]
assert "CINDER-884" in rag.prompt
assert llmwiki_rag._graph_hop_budget(
"普通提问",
{
"top_raw_vector_score": 0.2,
"min_similarity": 0.35,
"vector_candidates": 0,
"threshold_filtered": 4,
},
) == (2, "low_confidence")
@pytest.mark.asyncio
async def test_rag_graph_expansion_traverses_direct_non_seed_without_duplicate_output(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
mapping_store = _TraceMappingStore(
[
{
"id": "kb-equipment",
"weknora_id": "remote-equipment",
"name": "设备知识库",
"owner_user_id": "user-a",
}
]
)
direct_results = _ordinary_trace_results(3, knowledge_base_id="kb-equipment")
direct_results[2].update(
{
"id": "direct-middle",
"wiki_page_id": "direct-middle",
"wiki_slug": "nodes/middle",
"title": "中继控制站",
"content": "中继控制站也被向量通道直接召回。",
}
)
graph_store = _TraceGraphStore(
[
[
_graph_row(
mapping_id="kb-equipment",
seed_slug="ordinary/1",
neighbor_slug="nodes/middle",
title="中继控制站",
content="中继控制站连接上游控制中心。",
)
],
[
_graph_row(
mapping_id="kb-equipment",
seed_slug="nodes/middle",
neighbor_slug="nodes/upstream",
title="上游控制中心",
content="上游控制中心交接口令为 NEBULA-642。",
)
],
]
)
retrieval = _TraceRetrievalService(
{
"mode": "vector",
"results": direct_results,
"partial": False,
"warnings": [],
"fallback_knowledge_base_ids": [],
"diagnostics": {
"vector_candidates": 3,
"top_raw_vector_score": 0.74,
"min_similarity": 0.35,
},
}
)
request = _trace_request(
mapping_store=mapping_store,
retrieval_service=retrieval,
graph_store=graph_store,
)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=False),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(llmwiki_ids=["kb-equipment"]),
graph_input={"messages": [HumanMessage(content="沿完整链路做二跳追溯")]},
)
assert len(graph_store.calls) == 2
assert graph_store.calls[0]["seed_slugs"] == {
"kb-equipment": ["ordinary/1", "ordinary/2"]
}
assert graph_store.calls[1]["seed_slugs"] == {"kb-equipment": ["nodes/middle"]}
assert rag.trace["diagnostics"]["graph_candidates"] == 2
assert rag.trace["diagnostics"]["graph_unique_candidates"] == 1
assert rag.trace["diagnostics"]["graph_accepted"] == 1
assert rag.trace["diagnostics"]["graph_hops_executed"] == 2
middle_results = [
result for result in rag.trace["results"]
if result["wiki_slug"] == "nodes/middle"
]
assert len(middle_results) == 1
assert middle_results[0]["retrieval_mode"] == "vector"
upstream = next(
result for result in rag.trace["results"]
if result["wiki_slug"] == "nodes/upstream"
)
assert upstream["retrieval_mode"] == "graph_expansion"
assert upstream["graph_hop"] == 2
assert [node["slug"] for node in upstream["graph_path"]] == [
"ordinary/1",
"nodes/middle",
"nodes/upstream",
]
assert "NEBULA-642" in rag.prompt
@pytest.mark.asyncio
async def test_rag_expands_assistant_graph_two_hops_and_reports_assistant_trace(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
base_id = "assistant-graph"
direct_results = _assistant_trace_results(3, base_id=base_id)
direct_results[2].update(
{
"id": "assistant-middle",
"slug": "nodes/middle",
"title": "助手中继页",
"content": "助手中继页也被向量通道直接召回。",
"matched_section": {
"heading": "助手中继页",
"content": "助手中继页也被向量通道直接召回。",
"score": 0.61,
},
}
)
assistant_store = _TraceAssistantGraphStore(
direct_results,
bases=[{"id": base_id, "name": "关系助手库", "status": "active"}],
graph_batches=[
[
_assistant_graph_row(
base_id=base_id,
seed_slug="assistant/1",
neighbor_slug="nodes/middle",
title="助手中继页",
content="助手中继页连接二跳答案页。",
)
],
[
_assistant_graph_row(
base_id=base_id,
seed_slug="nodes/middle",
neighbor_slug="nodes/answer",
title="助手二跳答案页",
content="助手图二跳答案为 AURORA-915。",
)
],
],
)
request = _trace_request(
mapping_store=_TraceMappingStore([]),
retrieval_service=_TraceRetrievalService({"results": []}),
assistant_store=assistant_store,
)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=False),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(assistant_ids=[base_id]),
graph_input={"messages": [HumanMessage(content="沿关系做二跳追溯")]},
)
assert rag.trace["capabilities"]["wiki_graph_expansion"] is True
assert rag.trace["capabilities"]["raw_document_chunk_retrieval"] is True
assert rag.trace["mode"] == "assistant_graph_hybrid"
assert len(assistant_store.graph.calls) == 2
assert assistant_store.graph.calls[0]["seed_slugs"] == {
base_id: ["assistant/1", "assistant/2"]
}
assert assistant_store.graph.calls[1]["seed_slugs"] == {
base_id: ["nodes/middle"]
}
assert rag.trace["diagnostics"]["graph_channels"] == ["assistant"]
assert rag.trace["diagnostics"]["graph_candidates"] == 2
assert rag.trace["diagnostics"]["graph_unique_candidates"] == 1
assert rag.trace["diagnostics"]["graph_accepted"] == 1
assert rag.trace["diagnostics"]["graph_hops_requested"] == 2
assert rag.trace["diagnostics"]["graph_hops_executed"] == 2
middle_results = [
result for result in rag.trace["results"]
if result["wiki_slug"] == "nodes/middle"
]
assert len(middle_results) == 1
assert middle_results[0]["kind"] == "assistant"
assert middle_results[0]["retrieval_mode"] == "vector"
answer = next(
result for result in rag.trace["results"]
if result["wiki_slug"] == "nodes/answer"
)
assert answer["kind"] == "assistant"
assert answer["retrieval_mode"] == "graph_expansion"
assert answer["graph_hop"] == 2
assert answer["graph_score"] == pytest.approx(0.58 * 0.58)
assert [node["slug"] for node in answer["graph_path"]] == [
"assistant/1",
"nodes/middle",
"nodes/answer",
]
assert "AURORA-915" in rag.prompt
assert "知识空间标识=assistant-graph" in rag.prompt
assert "Wiki 标识=nodes/answer" in rag.prompt
assert "图扩展=第2跳" in rag.prompt
assert "助手二跳答案页 [知识空间=assistant-graph; Wiki=nodes/answer]" in rag.prompt
assert "--outgoing/references-->" in rag.prompt
assert "应视为该末节点已检索" in rag.prompt
def test_prompt_graph_path_preserves_incoming_direction_and_space_identity() -> None:
from app.gateway import llmwiki_rag
rendered = llmwiki_rag._prompt_graph_path(
{
"graph_path": [
{"slug": "seed", "title": "种子页"},
{
"slug": "upstream",
"title": "上游页",
"direction": "incoming",
"edge_type": "wikilink",
},
]
},
"kb-a",
)
assert rendered == (
"种子页 [知识空间=kb-a; Wiki=seed] "
"<--incoming/wikilink-- 上游页 [知识空间=kb-a; Wiki=upstream]"
)
@pytest.mark.asyncio
async def test_rag_reports_wiki_graph_raw_hybrid_mode_when_all_channels_contribute(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
mapping_store = _TraceMappingStore(
[
{
"id": "kb-equipment",
"weknora_id": "remote-equipment",
"name": "设备知识库",
"owner_user_id": "user-a",
}
]
)
raw_client = _TraceRawClient(
[
{
"chunk_id": "raw-one",
"knowledge_base_id": "remote-equipment",
"title": "原始手册",
"content": "原始手册正文。",
"score": 0.8,
}
]
)
graph_store = _TraceGraphStore(
[
[
_graph_row(
mapping_id="kb-equipment",
seed_slug="ordinary/1",
neighbor_slug="related/one",
title="关联规程",
content="关联规程正文。",
)
]
]
)
retrieval = _TraceRetrievalService(
{
"mode": "vector",
"results": _ordinary_trace_results(1, knowledge_base_id="kb-equipment"),
"partial": False,
"warnings": [],
"fallback_knowledge_base_ids": [],
"diagnostics": {"vector_candidates": 1},
},
client=raw_client,
)
request = _trace_request(
mapping_store=mapping_store,
retrieval_service=retrieval,
graph_store=graph_store,
)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=True),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(llmwiki_ids=["kb-equipment"]),
graph_input={"messages": [HumanMessage(content="联合检索设备操作")]} ,
)
assert rag.trace["mode"] == "wiki_graph_raw_hybrid"
assert {
result["retrieval_mode"] for result in rag.trace["results"]
} == {"raw_chunk", "vector", "graph_expansion"}
assert rag.trace["diagnostics"]["graph_sources_used"] == 1
@pytest.mark.asyncio
async def test_graph_expansion_failure_is_partial_and_keeps_direct_wiki_results(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
mapping_store = _TraceMappingStore(
[
{
"id": "kb-equipment",
"weknora_id": "remote-equipment",
"name": "设备知识库",
"owner_user_id": "user-a",
}
]
)
graph_store = _TraceGraphStore(error=RuntimeError("graph unavailable"))
retrieval = _TraceRetrievalService(
{
"mode": "vector",
"results": _ordinary_trace_results(1, knowledge_base_id="kb-equipment"),
"partial": False,
"warnings": [],
"fallback_knowledge_base_ids": [],
"diagnostics": {"vector_candidates": 1},
}
)
request = _trace_request(
mapping_store=mapping_store,
retrieval_service=retrieval,
graph_store=graph_store,
)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=False),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(llmwiki_ids=["kb-equipment"]),
graph_input={"messages": [HumanMessage(content="普通 Wiki 命中,图通道失败")]},
)
assert rag.source_count == 1
assert rag.sources[0]["id"] == "ordinary-1"
assert rag.trace["mode"] == "vector"
assert rag.trace["partial"] is True
assert rag.trace["capabilities"]["wiki_graph_expansion"] is True
assert "WIKI_GRAPH_EXPANSION_FAILED" in rag.trace["warnings"]
assert rag.trace["zero_result_reason"] is None
@pytest.mark.asyncio
async def test_zero_hit_trace_preserves_vector_threshold_diagnostics(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
mapping_store = _TraceMappingStore(
[
{
"id": "kb-equipment",
"weknora_id": "remote-equipment",
"name": "设备知识库",
"owner_user_id": "user-a",
}
]
)
retrieval = _TraceRetrievalService(
{
"query": "冷却泵异常代码 ZX-9",
"mode": "keyword",
"results": [],
"knowledge_base_count": 1,
"partial": False,
"warnings": [],
"fallback_knowledge_base_ids": [],
"diagnostics": {
"top_raw_vector_score": 0.21,
"min_similarity": 0.35,
"threshold_filtered": 7,
"vector_candidates": 0,
"keyword_candidates": 0,
"wiki_api_candidates": 0,
"fallback_reason": "VECTOR_BELOW_THRESHOLD",
},
}
)
request = _trace_request(mapping_store=mapping_store, retrieval_service=retrieval)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=True),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(llmwiki_ids=["kb-equipment"]),
graph_input={"messages": [HumanMessage(content="冷却泵异常代码 ZX-9")]},
)
assert rag.source_count == 0
assert rag.sources == []
assert "暂未找到相关资料" in rag.prompt
trace = rag.trace
assert trace["mode"] == "keyword"
assert trace["source_count"] == 0
assert trace["results"] == []
assert trace["fallback_reason"] == "VECTOR_BELOW_THRESHOLD"
assert trace["zero_result_reason"] == "VECTOR_BELOW_THRESHOLD"
assert trace["diagnostics"]["top_raw_vector_score"] == pytest.approx(0.21)
assert trace["diagnostics"]["min_similarity"] == pytest.approx(0.35)
assert trace["diagnostics"]["threshold_filtered"] == 7
assert trace["diagnostics"]["vector_candidates"] == 0
assert trace["diagnostics"]["keyword_candidates"] == 0
@pytest.mark.asyncio
async def test_trace_records_ordinary_and_assistant_space_selection(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
mapping_store = _TraceMappingStore(
[
{
"id": "kb-ordinary",
"weknora_id": "remote-ordinary",
"name": "普通 Wiki",
"owner_user_id": "user-a",
}
]
)
retrieval = _TraceRetrievalService(
{
"query": "联合检索",
"mode": "vector",
"results": [],
"knowledge_base_count": 1,
"partial": False,
"warnings": [],
"fallback_knowledge_base_ids": [],
"diagnostics": {
"top_raw_vector_score": 0.1,
"min_similarity": 0.35,
"threshold_filtered": 1,
"fallback_reason": "VECTOR_BELOW_THRESHOLD",
},
}
)
assistant_store = _TraceAssistantKnowledgeStore(
[
{
"id": "assistant-page",
"base_id": "assistant-base",
"slug": "assistant/page",
"title": "助手知识页",
"content": "助手库命中的内容",
"score": 0.76,
"retrieval_mode": "vector",
"matched_section": {
"heading": "助手知识章节",
"content": "助手库命中的内容",
"score": 0.76,
},
}
]
)
request = _trace_request(
mapping_store=mapping_store,
retrieval_service=retrieval,
assistant_store=assistant_store,
)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=True),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(
llmwiki_ids=["kb-ordinary"],
assistant_ids=["assistant-base"],
),
graph_input={"messages": [HumanMessage(content="联合检索")]},
)
selected = _selected_spaces_json(rag.trace)
assert "kb-ordinary" in selected
assert "assistant-base" in selected
assert rag.trace["authorized_space_count"] == 2
assert rag.trace["source_count"] == 1
assert rag.trace["results"][0]["title"] == "助手知识页"
assert rag.trace["results"][0]["matched_heading"] == "助手知识章节"
assert rag.trace["results"][0]["vector_score"] == pytest.approx(0.76)
assert assistant_store.calls[0]["base_ids"] == ["assistant-base"]
@pytest.mark.asyncio
async def test_attach_and_reference_response_round_trip_retrieval_trace(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
from app.gateway.routers.threads import (
ReferenceBatchesResponse,
_extract_knowledge_retrieval_traces,
)
mapping_store = _TraceMappingStore(
[
{
"id": "kb-one",
"weknora_id": "remote-one",
"name": "知识库一",
"owner_user_id": "user-a",
}
]
)
retrieval = _TraceRetrievalService(
{
"query": "测试追踪",
"mode": "wiki_api",
"results": [],
"knowledge_base_count": 1,
"partial": True,
"warnings": ["kb-one:WIKI_API_SEARCH_FAILED"],
"fallback_knowledge_base_ids": ["kb-one"],
"diagnostics": {
"vector_candidates": 0,
"keyword_candidates": 0,
"wiki_api_candidates": 0,
"fallback_reason": "RETRIEVAL_PROVIDER_UNAVAILABLE",
},
}
)
request = _trace_request(mapping_store=mapping_store, retrieval_service=retrieval)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=True),
)
original = HumanMessage(id="human-trace", content="测试追踪")
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(llmwiki_ids=["kb-one"]),
graph_input={"messages": [original]},
)
attached = llmwiki_rag.attach_llmwiki_rag_to_graph_input(
{"messages": [original]},
rag,
)
attached_message = attached["messages"][0]
assert attached_message.additional_kwargs["knowledge_retrieval_trace"] == rag.trace
traces = _extract_knowledge_retrieval_traces(attached["messages"])
assert traces == [{**rag.trace, "message_id": "human-trace"}]
# Persisted checkpoints use dictionaries. The helper must support that
# shape, ignore malformed values and de-duplicate by trace id.
serialized_messages = [
{
"type": "human",
"content": "测试追踪",
"additional_kwargs": {"knowledge_retrieval_trace": rag.trace},
},
{
"type": "human",
"content": "重复追踪",
"additional_kwargs": {"knowledge_retrieval_trace": dict(rag.trace)},
},
{
"type": "human",
"content": "坏数据",
"additional_kwargs": {"knowledge_retrieval_trace": "not-an-object"},
},
]
assert _extract_knowledge_retrieval_traces(serialized_messages) == [rag.trace]
response = ReferenceBatchesResponse(
batches=[],
retrieval_traces=_extract_knowledge_retrieval_traces(serialized_messages),
)
payload = response.model_dump()
assert payload["batches"] == []
assert payload["retrieval_traces"] == [rag.trace]
@pytest.mark.asyncio
async def test_retrieval_debug_reuses_answer_retrieval_and_marks_debug_trigger(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway.routers import llmwiki
captured: dict[str, Any] = {}
original_trace = {
"retrieval_trace_id": "trace-debug",
"trigger": "pre_retrieval",
"query": "泵故障",
"selected_spaces": [],
"source_count": 0,
"capabilities": {"vector_retrieval": True},
}
async def fake_build_llmwiki_rag_context(**kwargs: Any) -> SimpleNamespace:
captured.update(kwargs)
return SimpleNamespace(trace=original_trace)
async def admin_actor(_: object) -> tuple[str, bool]:
return "admin", True
monkeypatch.setattr(llmwiki, "_actor", admin_actor)
monkeypatch.setattr(
llmwiki,
"build_llmwiki_rag_context",
fake_build_llmwiki_rag_context,
)
request = SimpleNamespace()
body = llmwiki.RetrievalDebugRequest(
query=" 泵故障 ",
knowledge_base_ids=[" kb-one ", "kb-one"],
assistant_knowledge_base_ids=[" assistant-one ", "assistant-one"],
)
response = await llmwiki.debug_llmwiki_retrieval(request, body)
assert captured["request"] is request
assert captured["body"].context == {
"llmwiki_knowledge_base_ids": ["kb-one"],
"assistant_knowledge_base_ids": ["assistant-one"],
}
message = captured["graph_input"]["messages"][0]
assert isinstance(message, HumanMessage)
assert message.content == "泵故障"
assert response == {**original_trace, "trigger": "debug"}
assert original_trace["trigger"] == "pre_retrieval"
@pytest.mark.asyncio
async def test_retrieval_debug_requires_at_least_one_selected_space(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway.routers import llmwiki
async def admin_actor(_: object) -> tuple[str, bool]:
return "admin", True
monkeypatch.setattr(llmwiki, "_actor", admin_actor)
with pytest.raises(HTTPException) as exc_info:
await llmwiki.debug_llmwiki_retrieval(
SimpleNamespace(),
llmwiki.RetrievalDebugRequest(
query="泵故障",
knowledge_base_ids=[],
assistant_knowledge_base_ids=[],
),
)
assert exc_info.value.status_code == 422
@pytest.mark.asyncio
async def test_combined_retrieval_debug_rejects_non_admin_before_retrieval(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway.routers import llmwiki
retrieval_started = False
async def ordinary_actor(_: object) -> tuple[str, bool]:
return "ordinary-user", False
async def forbidden_retrieval(**_: Any) -> SimpleNamespace:
nonlocal retrieval_started
retrieval_started = True
return SimpleNamespace(trace={})
monkeypatch.setattr(llmwiki, "_actor", ordinary_actor)
monkeypatch.setattr(llmwiki, "build_llmwiki_rag_context", forbidden_retrieval)
with pytest.raises(HTTPException) as exc_info:
await llmwiki.debug_llmwiki_retrieval(
SimpleNamespace(),
llmwiki.RetrievalDebugRequest(
query="泵故障",
knowledge_base_ids=["kb-one"],
),
)
assert exc_info.value.status_code == 403
assert retrieval_started is False
@pytest.mark.asyncio
async def test_single_base_retrieval_debug_rejects_non_admin_before_services(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway.routers import llmwiki_index
services_started = False
async def ordinary_actor(_: object) -> tuple[str, bool]:
return "ordinary-user", False
async def forbidden_services(_: object) -> None:
nonlocal services_started
services_started = True
monkeypatch.setattr(llmwiki_index, "_actor", ordinary_actor)
monkeypatch.setattr(llmwiki_index, "_services", forbidden_services)
with pytest.raises(HTTPException) as exc_info:
await llmwiki_index.retrieval_debug(
SimpleNamespace(),
"kb-one",
llmwiki_index.WikiRetrievalDebugRequest(query="泵故障"),
)
assert exc_info.value.status_code == 403
assert services_started is False
@pytest.mark.asyncio
async def test_rag_context_restores_both_selection_kinds_from_persisted_metadata_when_keys_are_omitted(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
mapping_store = _TraceMappingStore(
[
{
"id": "kb-persisted",
"weknora_id": "remote-persisted",
"name": "持久化 Wiki",
"owner_user_id": "user-a",
}
]
)
retrieval = _TraceRetrievalService(
{
"mode": "vector",
"results": [],
"partial": False,
"warnings": [],
"fallback_knowledge_base_ids": [],
"diagnostics": {},
}
)
assistant_store = _TraceAssistantKnowledgeStore([])
request = _trace_request(
mapping_store=mapping_store,
retrieval_service=retrieval,
assistant_store=assistant_store,
)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=True),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=SimpleNamespace(assistant_id=None, context={}, metadata={}),
graph_input={"messages": [HumanMessage(content="继续问持久化知识")]},
fallback_metadata={
"llmwiki_knowledge_base_ids": ["kb-persisted"],
"assistant_knowledge_base_ids": ["assistant-persisted"],
},
)
assert rag.knowledge_base_ids == ["kb-persisted"]
assert rag.assistant_knowledge_base_ids == ["assistant-persisted"]
assert retrieval.calls == [
{
"query": "继续问持久化知识",
"mapping_ids": ["kb-persisted"],
"top_k_pages": 8,
}
]
assert assistant_store.calls[0]["base_ids"] == ["assistant-persisted"]
assert rag.trace["authorized_space_count"] == 2
selected_spaces = _selected_spaces_json(rag.trace)
assert "kb-persisted" in selected_spaces
assert "assistant-persisted" in selected_spaces
@pytest.mark.asyncio
@pytest.mark.parametrize(
("explicit_empty_key", "expected_llmwiki_ids", "expected_assistant_ids"),
[
("llmwiki_knowledge_base_ids", [], ["assistant-persisted"]),
("assistant_knowledge_base_ids", ["kb-persisted"], []),
],
)
async def test_explicit_empty_selection_is_not_restored_from_persisted_metadata(
monkeypatch: pytest.MonkeyPatch,
explicit_empty_key: str,
expected_llmwiki_ids: list[str],
expected_assistant_ids: list[str],
) -> None:
from app.gateway import llmwiki_rag
mapping_store = _TraceMappingStore(
[
{
"id": "kb-persisted",
"weknora_id": "remote-persisted",
"name": "持久化 Wiki",
"owner_user_id": "user-a",
}
]
)
retrieval = _TraceRetrievalService(
{
"mode": "vector",
"results": [],
"partial": False,
"warnings": [],
"fallback_knowledge_base_ids": [],
"diagnostics": {},
}
)
assistant_store = _TraceAssistantKnowledgeStore([])
request = _trace_request(
mapping_store=mapping_store,
retrieval_service=retrieval,
assistant_store=assistant_store,
)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=True),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=SimpleNamespace(
assistant_id=None,
context={explicit_empty_key: []},
metadata={},
),
graph_input={"messages": [HumanMessage(content="继续问持久化知识")]},
fallback_metadata={
"llmwiki_knowledge_base_ids": ["kb-persisted"],
"assistant_knowledge_base_ids": ["assistant-persisted"],
},
)
assert rag.knowledge_base_ids == expected_llmwiki_ids
assert rag.assistant_knowledge_base_ids == expected_assistant_ids
assert [call["mapping_ids"] for call in retrieval.calls] == (
[["kb-persisted"]] if expected_llmwiki_ids else []
)
assert [call["base_ids"] for call in assistant_store.calls] == (
[["assistant-persisted"]] if expected_assistant_ids else []
)
@pytest.mark.asyncio
async def test_explicitly_clearing_all_spaces_emits_a_durable_no_search_trace() -> None:
from app.gateway import llmwiki_rag
retrieval = _TraceRetrievalService(
{
"mode": "vector",
"results": [],
"partial": False,
"warnings": [],
"fallback_knowledge_base_ids": [],
"diagnostics": {},
}
)
assistant_store = _TraceAssistantKnowledgeStore([])
request = _trace_request(
mapping_store=_TraceMappingStore([]),
retrieval_service=retrieval,
assistant_store=assistant_store,
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=SimpleNamespace(
assistant_id=None,
context={
"llmwiki_knowledge_base_ids": [],
"assistant_knowledge_base_ids": [],
},
metadata={},
),
graph_input={"messages": [HumanMessage(content="明确清空后不要检索")]},
fallback_metadata={
"llmwiki_knowledge_base_ids": ["kb-old"],
"assistant_knowledge_base_ids": ["assistant-old"],
},
)
assert rag.knowledge_base_ids == []
assert rag.assistant_knowledge_base_ids == []
assert rag.sources == []
assert rag.prompt == ""
assert rag.trace["mode"] == "disabled"
assert rag.trace["selected_spaces"] == []
assert rag.trace["source_count"] == 0
assert rag.trace["zero_result_reason"] == "NO_KNOWLEDGE_SELECTED"
assert retrieval.calls == []
assert assistant_store.calls == []
def test_disabled_selection_trace_does_not_reenable_inherited_rag_mode() -> None:
from app.gateway.llmwiki_rag import LlmWikiRagContext
from app.gateway.services import _effective_rag_mode_after_llmwiki
body = SimpleNamespace(
context={
"llmwiki_knowledge_base_ids": [],
"assistant_knowledge_base_ids": [],
},
metadata={},
)
disabled_rag = LlmWikiRagContext(
[],
[],
0,
"",
[],
"明确清空后普通回答",
{"mode": "disabled", "zero_result_reason": "NO_KNOWLEDGE_SELECTED"},
)
assert _effective_rag_mode_after_llmwiki(
body=body,
rag=disabled_rag,
inherited=True,
) is False
def test_resolved_space_or_explicit_generic_rag_flag_still_enables_rag_mode() -> None:
from app.gateway.llmwiki_rag import LlmWikiRagContext
from app.gateway.services import _effective_rag_mode_after_llmwiki
selected_rag = LlmWikiRagContext(["kb-1"], [], 0, "", [], "问题", {})
assert _effective_rag_mode_after_llmwiki(
body=SimpleNamespace(context={}, metadata={}),
rag=selected_rag,
inherited=False,
) is True
explicit_generic_rag = LlmWikiRagContext([], [], 0, "", [], "问题", {"mode": "disabled"})
assert _effective_rag_mode_after_llmwiki(
body=SimpleNamespace(
context={
"llmwiki_knowledge_base_ids": [],
"assistant_knowledge_base_ids": [],
"rag_mode_enabled": True,
},
metadata={},
),
rag=explicit_generic_rag,
inherited=True,
) is True
@pytest.mark.asyncio
async def test_llmwiki_actor_uses_regular_runtime_user_without_admin_escalation(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
from app.gateway.auth.models import User
from deerflow.runtime.user_context import reset_current_user, set_current_user
async def no_http_user(_: object) -> None:
return None
context_user = User(email="regular@example.com", system_role="user")
token = set_current_user(context_user)
try:
monkeypatch.setattr(llmwiki_rag, "get_optional_user_from_request", no_http_user)
# Even disabled-auth mode must not override a real request-scoped user.
monkeypatch.setattr(llmwiki_rag, "is_auth_disabled", lambda: True)
user_id, is_admin = await llmwiki_rag._actor(SimpleNamespace())
finally:
reset_current_user(token)
assert user_id == str(context_user.id)
assert is_admin is False
@pytest.mark.asyncio
async def test_llmwiki_actor_without_any_user_is_not_admin_when_auth_is_enabled(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
async def no_http_user(_: object) -> None:
return None
monkeypatch.setattr(llmwiki_rag, "get_optional_user_from_request", no_http_user)
monkeypatch.setattr(llmwiki_rag, "get_current_user", lambda: None)
monkeypatch.setattr(llmwiki_rag, "get_effective_user_id", lambda: "default-user")
monkeypatch.setattr(llmwiki_rag, "is_auth_disabled", lambda: False)
assert await llmwiki_rag._actor(SimpleNamespace()) == ("default-user", False)
@pytest.mark.asyncio
async def test_fail_unstarted_run_marks_run_and_thread_error_and_releases_lease_once() -> None:
from app.gateway.services import _fail_unstarted_run
from deerflow.runtime.runs.manager import RunManager
from deerflow.runtime.runs.schemas import RunStatus
class ThreadStore:
def __init__(self) -> None:
self.calls: list[tuple[str, str]] = []
async def update_status(self, thread_id: str, status: str) -> None:
self.calls.append((thread_id, status))
class ConcurrencyGate:
def __init__(self) -> None:
self.releases: list[dict[str, Any]] = []
async def release(self, **kwargs: Any) -> None:
self.releases.append(dict(kwargs))
run_mgr = RunManager()
record = await run_mgr.create("thread-unstarted")
thread_store = ThreadStore()
gate = ConcurrencyGate()
lease_metadata = {
"run_id": record.run_id,
"thread_id": record.thread_id,
"model_name": "test-model",
"user_id": "user-a",
}
await _fail_unstarted_run(
run_mgr=run_mgr,
run_ctx=SimpleNamespace(thread_store=thread_store),
record=record,
thread_id=record.thread_id,
concurrency_gate=gate,
redis_lease_acquired=True,
lease_metadata=lease_metadata,
exc=RuntimeError("retrieval setup failed"),
)
assert record.status == RunStatus.error
assert record.error == "RuntimeError: retrieval setup failed"
assert thread_store.calls == [("thread-unstarted", "error")]
assert gate.releases == [lease_metadata]
@pytest.mark.asyncio
async def test_mixed_selection_backfills_all_eight_ordinary_results_when_assistant_has_no_hits(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
mapping_store = _TraceMappingStore(
[
{
"id": "kb-mixed",
"weknora_id": "remote-mixed",
"name": "普通 Wiki",
"owner_user_id": "user-a",
}
]
)
retrieval = _TraceRetrievalService(
{
"mode": "vector",
"results": _ordinary_trace_results(8),
"partial": False,
"warnings": [],
"fallback_knowledge_base_ids": [],
"diagnostics": {},
}
)
assistant_store = _TraceAssistantKnowledgeStore([])
request = _trace_request(
mapping_store=mapping_store,
retrieval_service=retrieval,
assistant_store=assistant_store,
)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=True),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(
llmwiki_ids=["kb-mixed"],
assistant_ids=["assistant-mixed"],
),
graph_input={"messages": [HumanMessage(content="混合检索无助手命中")]},
)
assert retrieval.calls[0]["top_k_pages"] == 8
assert assistant_store.calls[0]["limit"] == 8
assert rag.source_count == 8
assert [source["id"] for source in rag.sources] == [
f"ordinary-{index}" for index in range(1, 9)
]
assert [source["index"] for source in rag.sources] == list(range(1, 9))
assert [result["rank"] for result in rag.trace["results"]] == list(range(1, 9))
@pytest.mark.asyncio
async def test_mixed_selection_reserves_four_results_per_populated_channel_and_renumbers(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
mapping_store = _TraceMappingStore(
[
{
"id": "kb-mixed",
"weknora_id": "remote-mixed",
"name": "普通 Wiki",
"owner_user_id": "user-a",
}
]
)
retrieval = _TraceRetrievalService(
{
"mode": "vector",
"results": _ordinary_trace_results(8),
"partial": False,
"warnings": [],
"fallback_knowledge_base_ids": [],
"diagnostics": {},
}
)
assistant_store = _TraceAssistantKnowledgeStore(_assistant_trace_results(8))
request = _trace_request(
mapping_store=mapping_store,
retrieval_service=retrieval,
assistant_store=assistant_store,
)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=True),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(
llmwiki_ids=["kb-mixed"],
assistant_ids=["assistant-mixed"],
),
graph_input={"messages": [HumanMessage(content="混合检索双通道命中")]},
)
assert rag.trace["mode"] == "mixed"
assert rag.source_count == 8
assert [source["id"] for source in rag.sources] == [
"ordinary-1",
"ordinary-2",
"ordinary-3",
"ordinary-4",
"assistant-1",
"assistant-2",
"assistant-3",
"assistant-4",
]
assert [source["index"] for source in rag.sources] == list(range(1, 9))
assert [result["rank"] for result in rag.trace["results"]] == list(range(1, 9))
assert [result["kind"] for result in rag.trace["results"]] == [
"llmwiki",
"llmwiki",
"llmwiki",
"llmwiki",
"assistant",
"assistant",
"assistant",
"assistant",
]
@pytest.mark.asyncio
async def test_mixed_selection_keeps_raw_and_formatted_wiki_in_ordinary_quota(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.gateway import llmwiki_rag
mapping_store = _TraceMappingStore(
[
{
"id": "kb-mixed",
"weknora_id": "remote-mixed",
"name": "普通 Wiki",
"owner_user_id": "user-a",
}
]
)
raw_client = _TraceRawClient(
[
{
"chunk_id": f"raw-{index}",
"knowledge_id": f"document-{index}",
"knowledge_base_id": "remote-mixed",
"title": "海湖庄园资料",
"content": f"原始文档切片 {index}",
"score": 0.99 - index / 100,
}
for index in range(1, 9)
]
)
retrieval = _TraceRetrievalService(
{
"mode": "vector",
"results": _ordinary_trace_results(8),
"partial": False,
"warnings": [],
"fallback_knowledge_base_ids": [],
"diagnostics": {},
},
client=raw_client,
)
assistant_store = _TraceAssistantKnowledgeStore(_assistant_trace_results(8))
request = _trace_request(
mapping_store=mapping_store,
retrieval_service=retrieval,
assistant_store=assistant_store,
)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=True),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(
llmwiki_ids=["kb-mixed"],
assistant_ids=["assistant-mixed"],
),
graph_input={"messages": [HumanMessage(content="海湖庄园")]},
)
ordinary_head = rag.sources[:4]
assert [source["type"] for source in ordinary_head] == [
"llmwiki",
"llmwiki_wiki",
"llmwiki",
"llmwiki_wiki",
]
assert [source["type"] for source in rag.sources[4:]] == [
"assistant_knowledge"
] * 4
assert rag.trace["diagnostics"]["final_raw_chunk_source_count"] == 2
assert sum(source["type"] == "llmwiki_wiki" for source in rag.sources) == 2
@pytest.mark.asyncio
@pytest.mark.parametrize(
("failure_kind", "expected_warning"),
[
("missing", "ASSISTANT_RETRIEVAL_PROVIDER_UNAVAILABLE"),
("raises", "ASSISTANT_RETRIEVAL_FAILED"),
],
)
async def test_assistant_provider_failure_uses_unavailable_prompt_instead_of_no_match_prompt(
monkeypatch: pytest.MonkeyPatch,
failure_kind: str,
expected_warning: str,
) -> None:
from app.gateway import llmwiki_rag
class FailingAssistantStore:
async def search_pages(self, **_: Any) -> list[dict[str, Any]]:
raise RuntimeError("assistant index unavailable")
mapping_store = _TraceMappingStore([])
retrieval = _TraceRetrievalService({"results": []})
assistant_store: Any | None = None if failure_kind == "missing" else FailingAssistantStore()
request = _trace_request(
mapping_store=mapping_store,
retrieval_service=retrieval,
assistant_store=assistant_store,
)
async def actor(_: object) -> tuple[str, bool]:
return "user-a", False
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
monkeypatch.setattr(
llmwiki_rag,
"get_resolved_llmwiki_runtime",
lambda _config: SimpleNamespace(weknora_enabled=True),
)
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=request,
body=_trace_body(assistant_ids=["assistant-failing"]),
graph_input={"messages": [HumanMessage(content="助手知识检索失败")]},
)
assert rag.sources == []
assert rag.trace["partial"] is True
assert expected_warning in rag.trace["warnings"]
assert rag.trace["zero_result_reason"] == "RETRIEVAL_PROVIDER_UNAVAILABLE"
assert "检索不可用或未完整执行" not in rag.prompt
assert "不能判断库内是否存在相关资料" not in rag.prompt
assert "暂未找到相关资料" in rag.prompt
assert expected_warning not in rag.prompt
assert "不要向用户展示检索通道状态" in rag.prompt
@pytest.mark.asyncio
async def test_query_empty_trace_only_context_attaches_to_latest_human_and_round_trips() -> None:
from app.gateway import llmwiki_rag
from app.gateway.routers.threads import _extract_knowledge_retrieval_traces
graph_input = {
"messages": [
HumanMessage(id="human-old", content="旧问题"),
HumanMessage(
id="human-latest",
content=" ",
additional_kwargs={"client_marker": "keep"},
),
]
}
rag = await llmwiki_rag.build_llmwiki_rag_context(
request=SimpleNamespace(),
body=_trace_body(llmwiki_ids=["kb-query-empty"]),
graph_input=graph_input,
)
assert rag.prompt == ""
assert rag.sources == []
assert rag.trace["zero_result_reason"] == "QUERY_EMPTY"
attached = llmwiki_rag.attach_llmwiki_rag_to_graph_input(graph_input, rag)
old_message, latest_message = attached["messages"]
assert "knowledge_retrieval_trace" not in old_message.additional_kwargs
assert latest_message.additional_kwargs["client_marker"] == "keep"
assert latest_message.additional_kwargs["knowledge_retrieval_trace"] == rag.trace
assert "prompt_prefix" not in latest_message.additional_kwargs
assert _extract_knowledge_retrieval_traces(attached["messages"]) == [
{**rag.trace, "message_id": "human-latest"}
]
def test_tool_trace_parses_zero_result_error_and_clears_stale_reason_for_hits() -> None:
from app.gateway.routers.threads import _tool_trace_from_message
error_message = ToolMessage(
name="llmwiki_search",
tool_call_id="tool-error",
content=json.dumps(
{
"retrieval_trace_id": "trace-tool-error",
"query": "故障代码",
"selection": ["kb-one"],
"assistant_selection": [],
"mode": "empty",
"knowledge_base_count": 1,
"results": [],
"partial": True,
"warnings": ["RETRIEVAL_PROVIDER_UNAVAILABLE"],
"zero_result_reason": "RETRIEVAL_PROVIDER_UNAVAILABLE",
},
ensure_ascii=False,
),
)
error_trace = _tool_trace_from_message(error_message)
assert error_trace is not None
assert error_trace["trigger"] == "tool"
assert error_trace["source_count"] == 0
assert error_trace["results"] == []
assert error_trace["zero_result_reason"] == "RETRIEVAL_PROVIDER_UNAVAILABLE"
success_message = ToolMessage(
name="llmwiki_search",
tool_call_id="tool-success",
content=json.dumps(
{
"retrieval_trace_id": "trace-tool-success",
"query": "设备散热",
"selection": ["kb-one"],
"mode": "vector",
"knowledge_base_count": 1,
"results": [
{
"id": "page-one",
"knowledge_base_id": "kb-one",
"title": "设备散热",
"content": "先断电,再检查风道。",
"retrieval_mode": "vector",
"score": 0.91,
}
],
# A defensive parser must not carry a stale failure reason
# into a trace that contains usable evidence.
"zero_result_reason": "RETRIEVAL_PROVIDER_UNAVAILABLE",
},
ensure_ascii=False,
),
)
success_trace = _tool_trace_from_message(success_message)
assert success_trace is not None
assert success_trace["source_count"] == 1
assert success_trace["results"][0]["id"] == "page-one"
assert success_trace["zero_result_reason"] is None
class _ToolTraceMappingStore:
def __init__(self, rows: dict[str, dict[str, Any] | None]) -> None:
self.rows = rows
async def get_authorized(self, mapping_id: str, *_: Any, **__: Any) -> dict[str, Any] | None:
return self.rows.get(mapping_id)
class _ToolTraceOrdinaryService:
def __init__(self, result: dict[str, Any] | Exception) -> None:
self.result = result
self.calls: list[list[str]] = []
async def search(self, _query: str, rows: list[dict[str, Any]]) -> dict[str, Any]:
self.calls.append([str(row["id"]) for row in rows])
if isinstance(self.result, Exception):
raise self.result
return self.result
class _ToolTraceAssistantStore:
def __init__(
self,
*,
bases: dict[str, dict[str, Any] | None],
result: list[dict[str, Any]] | Exception,
) -> None:
self.bases = bases
self.result = result
async def get_base(self, base_id: str) -> dict[str, Any] | None:
return self.bases.get(base_id)
async def search_pages(self, *, base_ids: list[str], **_: Any) -> list[dict[str, Any]]:
if isinstance(self.result, Exception):
raise self.result
return self.result
def _tool_trace_ordinary_hit() -> dict[str, Any]:
return {
"id": "ordinary-page",
"knowledge_base_id": "ordinary-kb",
"title": "普通 Wiki 命中",
"content": "普通 Wiki 检索证据",
"retrieval_mode": "vector",
"score": 0.91,
}
def _tool_trace_assistant_hit() -> dict[str, Any]:
return {
"id": "assistant-page",
"base_id": "assistant-kb",
"title": "助手库命中",
"content": "助手知识检索证据",
"retrieval_mode": "vector",
"score": 0.87,
}
def _configure_search_tool_trace(
monkeypatch: pytest.MonkeyPatch,
*,
mapping_store: _ToolTraceMappingStore,
ordinary_service: _ToolTraceOrdinaryService,
assistant_store: _ToolTraceAssistantStore | None,
) -> Any:
from deerflow.persistence import assistant_knowledge
from deerflow.tools.builtins import llmwiki_search_tool as tool_module
monkeypatch.setattr(
tool_module,
"get_app_config",
lambda: SimpleNamespace(
llmwiki=SimpleNamespace(
local_wiki_index=SimpleNamespace(enabled=True, internal_search_enabled=True)
)
),
)
monkeypatch.setattr(
tool_module,
"get_resolved_llmwiki_runtime",
lambda: SimpleNamespace(weknora_enabled=True),
)
monkeypatch.setattr(tool_module, "get_effective_user_id", lambda: "user-a")
monkeypatch.setattr(
tool_module,
"get_current_user",
lambda: SimpleNamespace(system_role="user"),
)
monkeypatch.setattr(tool_module, "_agent_allowed_ids", lambda _context: None)
monkeypatch.setattr(tool_module, "get_llmwiki_store", lambda: mapping_store)
monkeypatch.setattr(
tool_module,
"get_local_wiki_retrieval_service",
lambda: ordinary_service,
)
monkeypatch.setattr(
tool_module,
"get_local_wiki_search_service",
lambda: SimpleNamespace(embedding=object()),
)
monkeypatch.setattr(
assistant_knowledge,
"get_assistant_knowledge_store",
lambda: assistant_store,
)
return tool_module
async def _run_mixed_search_tool(tool_module: Any) -> dict[str, Any]:
runtime = SimpleNamespace(
context={
"llmwiki_knowledge_base_ids": ["ordinary-kb"],
"assistant_knowledge_base_ids": ["assistant-kb"],
}
)
content = await tool_module.llmwiki_search_tool.coroutine("散热故障", runtime)
return json.loads(content)
@pytest.mark.asyncio
async def test_mixed_tool_keeps_assistant_hit_when_ordinary_authorization_fails(
monkeypatch: pytest.MonkeyPatch,
) -> None:
ordinary_service = _ToolTraceOrdinaryService({"mode": "vector", "results": []})
tool_module = _configure_search_tool_trace(
monkeypatch,
mapping_store=_ToolTraceMappingStore({"ordinary-kb": None}),
ordinary_service=ordinary_service,
assistant_store=_ToolTraceAssistantStore(
bases={"assistant-kb": {"id": "assistant-kb"}},
result=[_tool_trace_assistant_hit()],
),
)
trace = await _run_mixed_search_tool(tool_module)
assert ordinary_service.calls == []
assert trace["selection_count"] == 2
assert trace["authorized_space_count"] == 1
assert trace["partial"] is True
assert trace["warnings"] == ["SPACE_NOT_FOUND_OR_UNAUTHORIZED"]
assert [result["id"] for result in trace["results"]] == ["assistant-page"]
assert trace["zero_result_reason"] is None
@pytest.mark.asyncio
async def test_mixed_tool_keeps_assistant_hit_when_ordinary_search_fails(
monkeypatch: pytest.MonkeyPatch,
) -> None:
tool_module = _configure_search_tool_trace(
monkeypatch,
mapping_store=_ToolTraceMappingStore(
{"ordinary-kb": {"id": "ordinary-kb", "name": "普通 Wiki"}}
),
ordinary_service=_ToolTraceOrdinaryService(
WikiIndexError("WIKI_NOT_FINALIZED", "not ready", status_code=409)
),
assistant_store=_ToolTraceAssistantStore(
bases={"assistant-kb": {"id": "assistant-kb"}},
result=[_tool_trace_assistant_hit()],
),
)
trace = await _run_mixed_search_tool(tool_module)
assert trace["selection_count"] == 2
assert trace["authorized_space_count"] == 2
assert trace["partial"] is True
assert "WIKI_NOT_FINALIZED" in trace["warnings"]
assert [result["id"] for result in trace["results"]] == ["assistant-page"]
assert trace["zero_result_reason"] is None
@pytest.mark.asyncio
async def test_classic_tool_scopes_one_library_and_assigns_stable_global_ranks(
monkeypatch: pytest.MonkeyPatch,
) -> None:
ordinary_service = _ToolTraceOrdinaryService(
{
"mode": "vector",
"results": [
{
"id": "lower-score",
"knowledge_base_id": "ordinary-two",
"title": "次相关页面",
"content": "次相关证据",
"score": 0.71,
},
{
"id": "higher-score",
"knowledge_base_id": "ordinary-two",
"title": "最相关页面",
"content": "最相关证据",
"score": 0.95,
},
],
"partial": False,
"warnings": [],
}
)
tool_module = _configure_search_tool_trace(
monkeypatch,
mapping_store=_ToolTraceMappingStore(
{
"ordinary-one": {"id": "ordinary-one", "name": "一号库"},
"ordinary-two": {"id": "ordinary-two", "name": "二号库"},
}
),
ordinary_service=ordinary_service,
assistant_store=None,
)
runtime = SimpleNamespace(
context={
"knowledge_retrieval_mode": "classic",
"llmwiki_knowledge_base_ids": ["ordinary-one", "ordinary-two"],
"assistant_knowledge_base_ids": [],
}
)
first = json.loads(
await tool_module.llmwiki_search_tool.coroutine(
"治理进展",
runtime,
knowledge_base_id="ordinary-two",
)
)
second = json.loads(
await tool_module.llmwiki_search_tool.coroutine(
"治理进展",
runtime,
knowledge_base_id="ordinary-two",
)
)
assert ordinary_service.calls == [["ordinary-two"], ["ordinary-two"]]
assert first["selection"] == ["ordinary-two"]
assert [item["id"] for item in first["results"]] == [
"higher-score",
"lower-score",
]
assert [item["relevance_rank"] for item in first["results"]] == [1, 2]
assert [item["citation_id"] for item in first["results"]] == [
item["citation_id"] for item in second["results"]
]
assert all(
item["citation_id"].startswith("llmwiki:") for item in first["results"]
)
@pytest.mark.asyncio
async def test_classic_search_tool_rejects_smart_mode_even_if_called_directly(
monkeypatch: pytest.MonkeyPatch,
) -> None:
ordinary_service = _ToolTraceOrdinaryService(
{"mode": "vector", "results": [_tool_trace_ordinary_hit()]}
)
tool_module = _configure_search_tool_trace(
monkeypatch,
mapping_store=_ToolTraceMappingStore(
{"ordinary-kb": {"id": "ordinary-kb", "name": "普通 Wiki"}}
),
ordinary_service=ordinary_service,
assistant_store=None,
)
runtime = SimpleNamespace(
context={
"knowledge_retrieval_mode": "smart_reasoning",
"llmwiki_knowledge_base_ids": ["ordinary-kb"],
"assistant_knowledge_base_ids": [],
}
)
trace = json.loads(
await tool_module.llmwiki_search_tool.coroutine("散热故障", runtime)
)
assert trace["results"] == []
assert trace["zero_result_reason"] == "CLASSIC_MODE_REQUIRED"
assert ordinary_service.calls == []
@pytest.mark.asyncio
@pytest.mark.parametrize("provider_missing", [False, True])
async def test_mixed_tool_keeps_ordinary_hit_when_assistant_channel_fails(
monkeypatch: pytest.MonkeyPatch,
provider_missing: bool,
) -> None:
assistant_store = (
None
if provider_missing
else _ToolTraceAssistantStore(
bases={"assistant-kb": {"id": "assistant-kb"}},
result=RuntimeError("assistant index unavailable"),
)
)
tool_module = _configure_search_tool_trace(
monkeypatch,
mapping_store=_ToolTraceMappingStore(
{"ordinary-kb": {"id": "ordinary-kb", "name": "普通 Wiki"}}
),
ordinary_service=_ToolTraceOrdinaryService(
{
"mode": "vector",
"results": [_tool_trace_ordinary_hit()],
"partial": False,
"warnings": [],
}
),
assistant_store=assistant_store,
)
trace = await _run_mixed_search_tool(tool_module)
expected_warning = (
"ASSISTANT_RETRIEVAL_PROVIDER_UNAVAILABLE"
if provider_missing
else "ASSISTANT_RETRIEVAL_FAILED"
)
assert trace["selection_count"] == 2
assert trace["authorized_space_count"] == (1 if provider_missing else 2)
assert trace["partial"] is True
assert trace["warnings"] == [expected_warning]
assert [result["id"] for result in trace["results"]] == ["ordinary-page"]
assert trace["zero_result_reason"] is None
@pytest.mark.parametrize(
("extra_fields", "expected_authorized"),
[
({"selection_count": 2, "authorized_space_count": 1}, 1),
({"knowledge_base_count": 2}, 2),
],
)
def test_tool_trace_parser_prefers_explicit_authorized_count_and_supports_legacy(
extra_fields: dict[str, int],
expected_authorized: int,
) -> None:
from app.gateway.routers.threads import _tool_trace_from_message
message = ToolMessage(
name="llmwiki_search",
tool_call_id="call-counts",
content=json.dumps(
{
"query": "散热故障",
"selection": ["ordinary-kb"],
"assistant_selection": ["assistant-kb"],
"results": [],
**extra_fields,
},
ensure_ascii=False,
),
)
trace = _tool_trace_from_message(message)
assert trace is not None
assert trace["selection_count"] == 2
assert trace["authorized_space_count"] == expected_authorized