"""LLMWiki retrieval helpers for normal DeerFlow chat runs. The public chat API accepts DeerFlow-local LLMWiki knowledge-base mapping ids. When WeKnora mode is enabled, this module resolves those local ids under the current DeerFlow user's permissions, searches WeKnora once before the model call, and attaches both: * a hidden prompt with numbered context snippets, and * structured citation sources for the existing reference panel. The helper lives in ``app.gateway`` because it needs Gateway auth state and the DeerFlow-owned mapping store. Harness code remains app-import free. """ from __future__ import annotations import asyncio import logging from collections.abc import Mapping from dataclasses import dataclass, field from time import perf_counter from typing import Any from urllib.parse import urlencode from uuid import uuid4 from fastapi import HTTPException, Request from langchain_core.messages import HumanMessage from app.gateway.auth.disabled_mode import is_auth_disabled from app.gateway.deps import get_optional_user_from_request from app.gateway.utils import sanitize_log_param from deerflow.config.agents_config import load_agent_config, validate_agent_id from deerflow.integrations.weknora.local_index.overview import is_wiki_body_page from deerflow.integrations.weknora.local_index.ranking import ( title_boosted_rrf_score, title_match_strength, ) from deerflow.integrations.weknora.local_index.search import WikiIndexError from deerflow.integrations.weknora.runtime import get_resolved_llmwiki_runtime from deerflow.persistence.llmwiki import LlmWikiStore from deerflow.runtime.user_context import get_current_user, get_effective_user_id logger = logging.getLogger(__name__) _MAX_SOURCES = 8 _MAX_SOURCE_CHARS = 1400 _MAX_REFERENCE_ITEMS = 20 _MAX_MATCHED_SECTIONS = 8 _GRAPH_MAX_SEEDS = 4 _GRAPH_MAX_SEEDS_PER_KB = 2 _GRAPH_MAX_CANDIDATES = 24 _GRAPH_MAX_FRONTIER = 8 _GRAPH_MAX_ACCEPTED = 6 _GRAPH_MAX_ACCEPTED_PER_KB = 3 _GRAPH_DECAY = 0.58 _GRAPH_DEFAULT_SOURCE_SLOTS = 2 _GRAPH_DEEP_SOURCE_SLOTS = 3 _GRAPH_MULTI_HOP_MARKERS = ( "多跳", "二跳", "两跳", "上上游", "下下游", "上游的上游", "下游的下游", "完整链路", "完整路径", "关系链", "关系路径", "追溯链路", "追溯路径", "跨页关联", "间接关联", "间接影响", "沿链路", "沿关系", "multi-hop", "multihop", "two-hop", "two hop", "indirect relation", "indirect impact", "dependency path", "full chain", ) @dataclass(frozen=True) class LlmWikiRagContext: knowledge_base_ids: list[str] assistant_knowledge_base_ids: list[str] source_count: int prompt: str sources: list[dict[str, Any]] query: str trace: dict[str, Any] = field(default_factory=dict) def _store(request: Request) -> LlmWikiStore | None: return getattr(request.app.state, "llmwiki_store", None) def _assistant_store(request: Request) -> Any | None: return getattr(request.app.state, "assistant_knowledge_store", None) async def _actor(request: Request) -> tuple[str, bool]: user = await get_optional_user_from_request(request) if user is None: # In-process callers (roundtable/workflow gateways) intentionally use # a synthetic Request without cookies while carrying the real actor in # the request-scoped contextvar. Never turn that ordinary user into an # administrator merely because HTTP credentials are absent. context_user = get_current_user() if context_user is not None: return str(context_user.id), getattr(context_user, "system_role", None) == "admin" return get_effective_user_id(), is_auth_disabled() return str(user.id), getattr(user, "system_role", None) == "admin" def _as_id_list(value: Any) -> list[str]: if value is None: return [] if isinstance(value, str): parts = [value] elif isinstance(value, (list, tuple, set)): parts = list(value) else: return [] return list(dict.fromkeys(str(item).strip() for item in parts if str(item).strip()))[:100] def _body_selected_ids(body: Any) -> list[str]: for source in (getattr(body, "context", None), getattr(body, "metadata", None)): if isinstance(source, Mapping) and "llmwiki_knowledge_base_ids" in source: return _as_id_list(source.get("llmwiki_knowledge_base_ids")) return [] def _body_selected_assistant_ids(body: Any) -> list[str]: for source in (getattr(body, "context", None), getattr(body, "metadata", None)): if isinstance(source, Mapping) and "assistant_knowledge_base_ids" in source: return _as_id_list(source.get("assistant_knowledge_base_ids")) return [] def _body_declares_selection(body: Any, key: str) -> bool: return any( isinstance(source, Mapping) and key in source for source in (getattr(body, "context", None), getattr(body, "metadata", None)) ) def llmwiki_selection_declared(body: Any, key: str | None = None) -> bool: """Return whether a client explicitly supplied one or both selection keys. Presence is intentionally different from truthiness: an explicit empty list means "clear this retrieval scope" and must be persisted as such. """ if key is not None: return _body_declares_selection(body, key) return _body_declares_selection(body, "llmwiki_knowledge_base_ids") or _body_declares_selection( body, "assistant_knowledge_base_ids", ) def _assistant_agent_id(body: Any) -> str | None: assistant_id = validate_agent_id(getattr(body, "assistant_id", None)) if assistant_id: return assistant_id context = getattr(body, "context", None) if isinstance(context, Mapping): return validate_agent_id(str(context.get("agent_id") or "")) if context.get("agent_id") else None metadata = getattr(body, "metadata", None) if isinstance(metadata, Mapping): return validate_agent_id(str(metadata.get("agent_id") or "")) if metadata.get("agent_id") else None return None def _agent_bound_ids(body: Any) -> list[str]: agent_id = _assistant_agent_id(body) if not agent_id: return [] try: config = load_agent_config(agent_id) except (FileNotFoundError, ValueError): return [] return _as_id_list(config.llmwiki_knowledge_base_ids if config else None) def resolve_requested_llmwiki_knowledge_base_ids(body: Any) -> list[str]: """Return knowledge bases explicitly selected for the current conversation.""" return _body_selected_ids(body) def resolve_requested_assistant_knowledge_base_ids(body: Any) -> list[str]: """Return assistant knowledge bases explicitly selected for the run.""" return _body_selected_assistant_ids(body) def llmwiki_selection_requested(body: Any) -> bool: return bool( resolve_requested_llmwiki_knowledge_base_ids(body) or resolve_requested_assistant_knowledge_base_ids(body) ) def _latest_human_query(graph_input: dict[str, Any]) -> str: messages = graph_input.get("messages") if not isinstance(messages, list): return "" for message in reversed(messages): if isinstance(message, HumanMessage): return _message_text(getattr(message, "content", "")) if isinstance(message, dict) and str(message.get("type") or message.get("role") or "") in {"human", "user"}: return _message_text(message.get("content")) return "" def _message_text(content: Any) -> str: if isinstance(content, str): return content.strip() if isinstance(content, list): parts: list[str] = [] for item in content: if isinstance(item, str): parts.append(item) elif isinstance(item, Mapping): text = item.get("text") or item.get("content") if isinstance(text, str): parts.append(text) return "\n".join(part.strip() for part in parts if part and part.strip()).strip() return str(content or "").strip() def _truncate(text: str, max_chars: int) -> str: value = text.strip() if len(value) <= max_chars: return value return f"{value[: max_chars - 1].rstrip()}…" def _bounded_reference_items(value: Any) -> list[Any]: """Keep citation lineage useful without persisting arbitrary upstream data.""" if not isinstance(value, list): return [] allowed_keys = { "id", "chunk_id", "knowledge_id", "document_id", "title", "knowledge_title", "filename", "heading", "url", } bounded: list[Any] = [] for item in value[:_MAX_REFERENCE_ITEMS]: if isinstance(item, Mapping): safe: dict[str, Any] = {} for key in allowed_keys: field_value = item.get(key) if isinstance(field_value, str): safe[key] = _truncate(field_value, 256) elif isinstance(field_value, (int, float, bool)) or field_value is None: safe[key] = field_value if safe: bounded.append(safe) elif isinstance(item, (str, int, float, bool)): bounded.append(_truncate(str(item), 256)) return bounded def _bounded_matched_sections(value: Any) -> list[dict[str, Any]]: if not isinstance(value, list): return [] sections: list[dict[str, Any]] = [] for item in value[:_MAX_MATCHED_SECTIONS]: if not isinstance(item, Mapping): continue safe: dict[str, Any] = {} for key in ("section_index", "heading", "score"): field_value = item.get(key) if isinstance(field_value, str): safe[key] = _truncate(field_value, 256) elif isinstance(field_value, (int, float)): safe[key] = field_value if safe: sections.append(safe) return sections def _source_from_result(result: dict[str, Any], mapping: dict[str, Any], index: int) -> dict[str, Any]: if result.get("kind") in {"wiki_page", "wiki_section"}: sections = list(result.get("matched_sections") or []) content = "\n\n".join(str(section.get("content") or "") for section in sections if section.get("content")) content = _truncate( content or str(result.get("content") or result.get("content_markdown") or ""), _MAX_SOURCE_CHARS, ) wiki_slug = str(result.get("wiki_slug") or result.get("slug") or "") url = str(result.get("url") or "") if not url and wiki_slug: url = f"/weknora-source-preview?{urlencode({'knowledge_base_id': mapping['id'], 'wiki_slug': wiki_slug})}" raw = { "kind": str(result.get("kind") or "wiki_page"), "id": str(result.get("id") or ""), "wiki_page_id": str(result.get("wiki_page_id") or result.get("id") or ""), "knowledge_base_id": str(mapping["id"]), "knowledge_base_name": str(mapping.get("name") or ""), "title": str(result.get("title") or ""), "wiki_slug": wiki_slug, "page_type": result.get("page_type"), "heading": result.get("heading"), "matched_heading": result.get("matched_heading") or result.get("heading"), "matched_sections": _bounded_matched_sections(sections), "title_match_score": float(result.get("title_match_score") or 0), "summary": _truncate(str(result.get("summary") or ""), 360), "retrieval_mode": str(result.get("retrieval_mode") or "wiki_api"), "rank_score_semantics": result.get("rank_score_semantics"), "channel_score": result.get("channel_score"), "vector_score": result.get("vector_score"), "source_refs": _bounded_reference_items(result.get("source_refs")), "chunk_refs": _bounded_reference_items(result.get("chunk_refs")), "source": str(mapping.get("name") or "LLMWiki"), "url": url, } return { "index": index, "id": result.get("wiki_page_id") or result.get("id"), "title": result.get("title") or mapping.get("name") or f"Wiki 页面 {index}", "type": "llmwiki_wiki", "snippet": _truncate(content, 360), "content": content, "source": mapping.get("name") or "LLMWiki", "score": float(result.get("score") or 0), "title_match_score": float(result.get("title_match_score") or 0), "retrieval_mode": str(result.get("retrieval_mode") or "wiki_api"), "url": url, "skillName": "llmwiki_search", "mode": "citation", "raw": raw, } chunk_id = str(result.get("chunk_id") or "") title = str(result.get("title") or result.get("filename") or mapping.get("name") or f"LLMWiki 片段 {index}") content = _truncate(str(result.get("content") or ""), _MAX_SOURCE_CHARS) params = urlencode({"knowledge_base_id": mapping["id"], "chunk_id": chunk_id}) raw = { "kind": str(result.get("kind") or "chunk"), "id": str(result.get("id") or chunk_id), "chunk_id": chunk_id, "knowledge_id": str(result.get("knowledge_id") or ""), "knowledge_base_id": str(mapping["id"]), "knowledge_base_name": str(mapping.get("name") or ""), "title": title, "filename": str(result.get("filename") or ""), "chunk_index": int(result.get("chunk_index") or 0), "retrieval_mode": str(result.get("retrieval_mode") or "wiki_api"), "rank_score_semantics": result.get("rank_score_semantics"), "channel_score": result.get("channel_score"), "vector_score": result.get("vector_score"), "source_refs": _bounded_reference_items(result.get("source_refs")), "chunk_refs": _bounded_reference_items(result.get("chunk_refs")), "title_match_score": float(result.get("title_match_score") or 0), "source": str(mapping.get("name") or "LLMWiki"), "url": f"/weknora-source-preview?{params}", } return { "index": index, "id": chunk_id or str(result.get("knowledge_id") or f"{mapping['id']}:{index}"), "title": title, "type": "llmwiki", "snippet": _truncate(content, 360), "content": content, "source": mapping.get("name") or "LLMWiki", "score": float(result.get("score") or 0), "retrieval_mode": str(result.get("retrieval_mode") or "wiki_api"), "url": f"/weknora-source-preview?{params}", "skillName": "llmwiki_search", "mode": "citation", "raw": raw, } def _source_from_raw_chunk_result( result: Mapping[str, Any], mapping: Mapping[str, Any], index: int, ) -> dict[str, Any]: """Convert one authorized WeKnora chunk into a bounded citation source. The raw knowledge-search score is provider-defined (it may include hybrid ranking or reranking), so it is deliberately exposed as a provider/channel score rather than as cosine similarity. Keep the raw payload allowlisted: upstream chunk metadata can be arbitrarily large or contain internal data. """ chunk_id = str(result.get("chunk_id") or "") content = _truncate(str(result.get("content") or ""), _MAX_SOURCE_CHARS) title = str( result.get("title") or result.get("filename") or mapping.get("name") or f"LLMWiki 原文片段 {index}" ) try: provider_score = float(result.get("score") or 0) except (TypeError, ValueError): provider_score = 0.0 params = urlencode({"knowledge_base_id": str(mapping["id"]), "chunk_id": chunk_id}) url = f"/weknora-source-preview?{params}" raw = { "chunk_id": chunk_id, "knowledge_id": str(result.get("knowledge_id") or ""), "knowledge_base_id": str(mapping["id"]), "knowledge_base_name": str(mapping.get("name") or ""), "title": title, "filename": str(result.get("filename") or ""), "chunk_index": int(result.get("chunk_index") or 0), "retrieval_mode": "raw_chunk", "provider_score": provider_score, "channel_score": provider_score, "rank_score_semantics": "provider_relevance", "title_match_score": float(result.get("title_match_score") or 0), "source": str(mapping.get("name") or "LLMWiki"), "url": url, } return { "index": index, "id": chunk_id or str(result.get("knowledge_id") or f"{mapping['id']}:{index}"), "title": title, "type": "llmwiki", "snippet": _truncate(content, 360), "content": content, "source": str(mapping.get("name") or "LLMWiki"), "score": provider_score, "title_match_score": float(result.get("title_match_score") or 0), "retrieval_mode": "raw_chunk", "url": url, "skillName": "llmwiki_search", "mode": "citation", "raw": raw, } def _rrf_merge_source_channels( channels: list[list[dict[str, Any]]], *, limit: int, query: str = "", title_boost_enabled: bool = True, title_boost_weight: float = 0.25, min_items_per_channel: int = 0, ) -> list[dict[str, Any]]: """Fuse independently ranked source channels without starving a channel. Title boosting is intentionally allowed to reorder sources, but it must not let repeated chunks from one raw document consume every final citation slot while formatted Wiki pages are available. ``min_items_per_channel`` keeps a bounded quota from each populated channel and rank-interleaves the chosen items; unused quota is filled from the global boosted ranking. """ fused: dict[tuple[str, str, str], dict[str, Any]] = {} contributions: dict[tuple[str, str, str], float] = {} first_seen: dict[tuple[str, str, str], int] = {} title_scores: dict[tuple[str, str, str], float] = {} channel_keys: list[list[tuple[str, str, str]]] = [] position = 0 for channel in channels: keys_in_channel: list[tuple[str, str, str]] = [] for rank, source in enumerate(channel, start=1): raw = source.get("raw") if isinstance(source.get("raw"), Mapping) else {} key = ( str(source.get("type") or ""), str(raw.get("knowledge_base_id") or source.get("source") or ""), str(source.get("id") or ""), ) if key in keys_in_channel: continue keys_in_channel.append(key) contributions[key] = contributions.get(key, 0.0) + 1.0 / (60 + rank) raw_title_score = raw.get("title_match_score") try: raw_title_score = float(raw_title_score) except (TypeError, ValueError): raw_title_score = 0.0 title_score = max( float(source.get("title_match_score") or 0), raw_title_score, title_match_strength( query, source.get("title") or raw.get("title") or raw.get("filename"), raw.get("aliases") if isinstance(raw.get("aliases"), list) else None, ) if query else 0.0, ) title_scores[key] = max(title_scores.get(key, 0.0), title_score) if key not in fused: fused[key] = dict(source) first_seen[key] = position position += 1 if keys_in_channel: channel_keys.append(keys_in_channel) ranking_scores = { key: title_boosted_rrf_score( contributions[key], title_scores[key], enabled=title_boost_enabled, weight=title_boost_weight, ) for key in fused } ranked_keys = sorted(fused, key=lambda key: (-ranking_scores[key], first_seen[key])) output_keys = ranked_keys[:limit] if min_items_per_channel > 0 and len(channel_keys) > 1 and limit > 1: quota = min(min_items_per_channel, max(1, limit // len(channel_keys))) rank_position = {key: index for index, key in enumerate(ranked_keys)} ranked_channel_keys = [ sorted(set(keys), key=lambda key: rank_position[key]) for keys in channel_keys ] selected: set[tuple[str, str, str]] = set() for keys in ranked_channel_keys: selected.update(keys[:quota]) for key in ranked_keys: if len(selected) >= limit: break selected.add(key) # Keep both evidence layers visible in every prefix (the ordinary # channel may later be sliced again when assistant knowledge is mixed). channel_order = sorted( range(len(ranked_channel_keys)), key=lambda index: rank_position[ranked_channel_keys[index][0]], ) output_keys = [] offsets = [0] * len(ranked_channel_keys) while len(output_keys) < min(limit, len(selected)): appended = False for channel_index in channel_order: keys = ranked_channel_keys[channel_index] while offsets[channel_index] < len(keys): key = keys[offsets[channel_index]] offsets[channel_index] += 1 if key not in selected or key in output_keys: continue output_keys.append(key) appended = True break if len(output_keys) >= min(limit, len(selected)): break if not appended: break for key in ranked_keys: if len(output_keys) >= limit: break if key in selected and key not in output_keys: output_keys.append(key) merged: list[dict[str, Any]] = [] for key in output_keys: source = fused[key] raw = source.get("raw") if isinstance(source.get("raw"), Mapping) else {} source["rrf_score"] = contributions[key] source["title_ranking_score"] = ranking_scores[key] source["score"] = contributions[key] source["title_match_score"] = title_scores[key] source["raw"] = { **raw, "rrf_score": contributions[key], "title_ranking_score": ranking_scores[key], "rank_score_semantics": "rrf", "title_match_score": title_scores[key], } merged.append(source) return merged def _graph_hop_budget(query: str, diagnostics: Mapping[str, Any]) -> tuple[int, str]: """Choose the bounded Wiki-graph depth without an extra model call.""" normalized = " ".join(query.lower().split()) if any(marker in normalized for marker in _GRAPH_MULTI_HOP_MARKERS): return 2, "explicit_multi_hop" fallback_reason = str(diagnostics.get("fallback_reason") or "") if fallback_reason in {"VECTOR_BELOW_THRESHOLD", "VECTOR_UNAVAILABLE"}: return 2, "low_confidence" top_score = diagnostics.get("top_raw_vector_score") min_similarity = diagnostics.get("min_similarity") if ( isinstance(top_score, (int, float)) and isinstance(min_similarity, (int, float)) and float(top_score) < float(min_similarity) + 0.05 ): return 2, "low_confidence" if diagnostics.get("threshold_filtered") and not diagnostics.get("vector_candidates"): return 2, "low_confidence" return 1, "default" def _graph_path_node( *, slug: str, title: str, direction: str | None = None, edge_type: str | None = None, ) -> dict[str, Any]: node: dict[str, Any] = { "slug": _truncate(slug, 256), "title": _truncate(title, 256), } if direction: node["direction"] = _truncate(direction, 32) if edge_type: node["edge_type"] = _truncate(edge_type, 64) return node async def _expand_wiki_graph( *, index_store: Any, mappings: list[dict[str, Any]], wiki_sources: list[dict[str, Any]], query: str, retrieval_diagnostics: Mapping[str, Any], source_kind: str = "llmwiki", ) -> tuple[list[dict[str, Any]], dict[str, Any]]: """Expand authorized Wiki results over a tightly bounded local graph. ``get_adjacent_pages`` is an optional persistence capability. Its output is treated as untrusted even though the store also scopes it: every row is checked again against the already-authorized mapping ids and the current frontier before any body reaches the model or trace. """ get_adjacent_pages = getattr(index_store, "get_adjacent_pages", None) hop_budget, depth_reason = _graph_hop_budget(query, retrieval_diagnostics) normalized_source_kind = "assistant" if source_kind == "assistant" else "llmwiki" details: dict[str, Any] = { "graph_source_kind": normalized_source_kind, "graph_seed_count": 0, "graph_seeds": [], "graph_candidates": 0, "graph_unique_candidates": 0, "graph_accepted": 0, "graph_sources_used": 0, "graph_rejected": 0, "graph_hops_requested": hop_budget, "graph_hops_executed": 0, "graph_depth_reason": depth_reason, "graph_paths": [], } if not callable(get_adjacent_pages): return [], details mapping_by_id = {str(mapping.get("id") or ""): mapping for mapping in mappings} authorized_mapping_ids = [mapping_id for mapping_id in mapping_by_id if mapping_id] if not authorized_mapping_ids: return [], details # All directly recalled Wiki pages are excluded from expansion output, not # just the selected seeds. Keep that output de-duplication separate from # traversal state: a directly recalled non-seed page may still be the only # bridge to a useful page on the next hop. direct_page_keys: set[tuple[str, str]] = set() seeds: list[dict[str, Any]] = [] seed_counts: dict[str, int] = {} for rank, source in enumerate(wiki_sources, start=1): raw = source.get("raw") if isinstance(source.get("raw"), Mapping) else {} mapping_id = str(raw.get("knowledge_base_id") or raw.get("base_id") or "") slug = str(raw.get("wiki_slug") or raw.get("slug") or "").strip("/ ") if mapping_id not in mapping_by_id or not slug: continue direct_page_keys.add((mapping_id, slug)) if len(seeds) >= _GRAPH_MAX_SEEDS: continue if seed_counts.get(mapping_id, 0) >= _GRAPH_MAX_SEEDS_PER_KB: continue seed_counts[mapping_id] = seed_counts.get(mapping_id, 0) + 1 seeds.append( { "knowledge_base_id": mapping_id, "wiki_slug": slug, "title": str(source.get("title") or slug), "rank": rank, "score": 1.0 / rank, } ) details["graph_seed_count"] = len(seeds) details["graph_seeds"] = [ { "kind": normalized_source_kind, "knowledge_base_id": seed["knowledge_base_id"], "wiki_slug": seed["wiki_slug"], "title": _truncate(str(seed["title"]), 256), "rank": seed["rank"], } for seed in seeds ] if not seeds: return [], details # A state carries only local page identity and an auditable path. Scores # form a bounded personalized walk: each hop decays the parent's mass and # multiple independent paths add support, capped at 1. frontier: dict[tuple[str, str], dict[str, Any]] = {} for seed in seeds: key = (str(seed["knowledge_base_id"]), str(seed["wiki_slug"])) frontier[key] = { "score": float(seed["score"]), "path": [ _graph_path_node( slug=str(seed["wiki_slug"]), title=str(seed["title"]), ) ], } traversed_page_keys = set(frontier) candidates: dict[tuple[str, str], dict[str, Any]] = {} candidate_rows = 0 rejected = 0 for hop in range(1, hop_budget + 1): if not frontier or candidate_rows >= _GRAPH_MAX_CANDIDATES: break frontier_by_mapping: dict[str, list[str]] = {} for mapping_id, slug in frontier: frontier_by_mapping.setdefault(mapping_id, []).append(slug) for slugs in frontier_by_mapping.values(): slugs.sort() remaining = _GRAPH_MAX_CANDIDATES - candidate_rows call_limit = remaining if hop_budget == 1 else min(12, remaining) rows = await get_adjacent_pages( authorized_mapping_ids, frontier_by_mapping, direction="both", limit=call_limit, include_drafts=False, ) details["graph_hops_executed"] = hop bounded_rows = list(rows or [])[:call_limit] if isinstance(rows, list) else [] candidate_rows += len(bounded_rows) next_frontier: dict[tuple[str, str], dict[str, Any]] = {} for row in bounded_rows: if not isinstance(row, Mapping): rejected += 1 continue page_value = row.get("page") page = dict(page_value) if isinstance(page_value, Mapping) else {} mapping_id = str( page.get("knowledge_base_mapping_id") or page.get("knowledge_base_id") or page.get("base_id") or row.get("knowledge_base_mapping_id") or row.get("base_id") or "" ) seed_slug = str(row.get("seed_slug") or "").strip("/ ") neighbor_slug = str( page.get("slug") or page.get("wiki_slug") or row.get("neighbor_slug") or "" ).strip("/ ") declared_neighbor = str(row.get("neighbor_slug") or "").strip("/ ") parent = frontier.get((mapping_id, seed_slug)) if ( mapping_id not in mapping_by_id or parent is None or not neighbor_slug or (declared_neighbor and declared_neighbor != neighbor_slug) ): rejected += 1 continue status = str(page.get("status") or "published").strip().lower() content = str( page.get("content_md") or page.get("content_markdown") or page.get("content") or "" ).strip() if ( status in {"archived", "deleted", "draft"} or not is_wiki_body_page(page) or not content ): rejected += 1 continue key = (mapping_id, neighbor_slug) if key in traversed_page_keys: rejected += 1 continue direction = str(row.get("direction") or "related") edge_type = str(row.get("edge_type") or "link") title = str(page.get("title") or neighbor_slug) contribution = min(1.0, float(parent["score"]) * _GRAPH_DECAY) path = [ *list(parent["path"]), _graph_path_node( slug=neighbor_slug, title=title, direction=direction, edge_type=edge_type, ), ][:3] current = next_frontier.get(key) if current is None: next_frontier[key] = { "knowledge_base_id": mapping_id, "slug": neighbor_slug, "title": title, "page": page, "hop": hop, "score": contribution, "best_contribution": contribution, "path": path, } else: current["score"] = min(1.0, float(current["score"]) + contribution) if contribution > float(current["best_contribution"]): current["best_contribution"] = contribution current["path"] = path current["page"] = page for key, state in next_frontier.items(): if key not in direct_page_keys: candidates[key] = state traversed_page_keys.update(next_frontier) ranked_frontier = sorted( next_frontier.items(), key=lambda item: ( -float(item[1]["score"]), item[0][0], item[0][1], ), )[:_GRAPH_MAX_FRONTIER] frontier = {key: state for key, state in ranked_frontier} details["graph_candidates"] = candidate_rows details["graph_unique_candidates"] = len(candidates) details["graph_rejected"] = rejected ranked_candidates = sorted( candidates.values(), key=lambda item: ( -float(item["score"]), int(item["hop"]), str(item["knowledge_base_id"]), str(item["slug"]), ), ) accepted: list[dict[str, Any]] = [] accepted_per_mapping: dict[str, int] = {} for candidate in ranked_candidates: mapping_id = str(candidate["knowledge_base_id"]) if accepted_per_mapping.get(mapping_id, 0) >= _GRAPH_MAX_ACCEPTED_PER_KB: continue accepted.append(candidate) accepted_per_mapping[mapping_id] = accepted_per_mapping.get(mapping_id, 0) + 1 if len(accepted) >= _GRAPH_MAX_ACCEPTED: break sources: list[dict[str, Any]] = [] paths: list[dict[str, Any]] = [] for candidate in accepted: mapping_id = str(candidate["knowledge_base_id"]) mapping = mapping_by_id[mapping_id] page = candidate["page"] graph_score = float(candidate["score"]) content = _truncate( str( page.get("content_md") or page.get("content_markdown") or page.get("content") or "" ), _MAX_SOURCE_CHARS, ) if normalized_source_kind == "assistant": result = { "kind": "assistant", "id": page.get("id") or f"{mapping_id}:{candidate['slug']}", "base_id": mapping_id, "knowledge_base_name": mapping.get("name") or "", "slug": candidate["slug"], "title": candidate["title"], "summary": page.get("summary") or "", "page_type": page.get("page_type") or "article", "content_markdown": content, "content": content, "score": graph_score, "retrieval_mode": "graph_expansion", "rank_score_semantics": "graph_decay", "source_refs": page.get("source_refs") or [], "chunk_refs": page.get("chunk_refs") or [], } source = _source_from_assistant_result(result, len(sources) + 1) else: result = { "kind": "wiki_page", "id": page.get("id") or f"{mapping_id}:{candidate['slug']}", "wiki_page_id": page.get("id") or f"{mapping_id}:{candidate['slug']}", "wiki_slug": candidate["slug"], "knowledge_base_id": mapping_id, "knowledge_base_name": mapping.get("name") or "", "title": candidate["title"], "summary": page.get("summary") or "", "page_type": page.get("page_type") or "article", "content": content, "score": graph_score, "retrieval_mode": "graph_expansion", "rank_score_semantics": "graph_decay", "source_refs": page.get("source_refs") or [], "chunk_refs": page.get("chunk_refs") or [], } source = _source_from_result(result, mapping, len(sources) + 1) raw = source.get("raw") if isinstance(source.get("raw"), Mapping) else {} source["raw"] = { **raw, "graph_hop": int(candidate["hop"]), "graph_score": graph_score, "graph_path": list(candidate["path"]), } sources.append(source) paths.append( { "kind": normalized_source_kind, "knowledge_base_id": mapping_id, "wiki_slug": candidate["slug"], "hop": int(candidate["hop"]), "score": round(graph_score, 8), "path": list(candidate["path"]), } ) details["graph_accepted"] = len(sources) details["graph_paths"] = paths return sources, details def _combine_graph_details(batches: list[Mapping[str, Any]]) -> dict[str, Any]: """Combine ordinary and assistant graph diagnostics without hiding either.""" usable = [batch for batch in batches if batch] if not usable: return {} count_keys = ( "graph_seed_count", "graph_candidates", "graph_unique_candidates", "graph_accepted", "graph_rejected", ) combined: dict[str, Any] = { key: sum(int(batch.get(key) or 0) for batch in usable) for key in count_keys } combined["graph_seeds"] = [ seed for batch in usable for seed in batch.get("graph_seeds") or [] if isinstance(seed, Mapping) ] combined["graph_paths"] = [ path for batch in usable for path in batch.get("graph_paths") or [] if isinstance(path, Mapping) ] combined["graph_hops_requested"] = max( int(batch.get("graph_hops_requested") or 0) for batch in usable ) combined["graph_hops_executed"] = max( int(batch.get("graph_hops_executed") or 0) for batch in usable ) depth_reasons = { str(batch.get("graph_depth_reason") or "default") for batch in usable } combined["graph_depth_reason"] = next( ( reason for reason in ("explicit_multi_hop", "low_confidence", "default") if reason in depth_reasons ), "default", ) combined["graph_channels"] = list( dict.fromkeys( str(batch.get("graph_source_kind") or "llmwiki") for batch in usable ) ) return combined def _merge_graph_sources( direct_sources: list[dict[str, Any]], graph_sources: list[dict[str, Any]], *, deep: bool, limit: int, ) -> list[dict[str, Any]]: """Reserve bounded graph slots while keeping direct retrieval dominant.""" if not graph_sources: return direct_sources[:limit] if not direct_sources: return graph_sources[:limit] requested_slots = _GRAPH_DEEP_SOURCE_SLOTS if deep else _GRAPH_DEFAULT_SOURCE_SLOTS graph_slots = min(requested_slots, len(graph_sources), max(0, limit - 1)) selected_graph = graph_sources[:graph_slots] selected_direct = direct_sources[: limit - len(selected_graph)] # Keep the two strongest direct results first, then make graph evidence # visible before the remaining backfill. This is stable and never compares # unrelated vector/provider/graph score scales. head_size = min(2, len(selected_direct)) return [ *selected_direct[:head_size], *selected_graph, *selected_direct[head_size:], ][:limit] def _source_from_assistant_result(result: dict[str, Any], index: int) -> dict[str, Any]: title = str(result.get("title") or f"助手知识库片段 {index}") matched_section = result.get("matched_section") if isinstance(result.get("matched_section"), Mapping) else {} content = str( matched_section.get("content") or result.get("content_markdown") or result.get("content") or result.get("summary") or "" ) # The retrieval citation must expose the matched knowledge itself. A page # summary is display metadata and can omit the fact that answered the query. snippet = content raw = { **result, "source": "助手知识库", "url": "/page/workspace/knowledge/assistant", } return { "index": index, "id": result.get("id") or result.get("slug") or f"assistant:{index}", "title": title, "type": "assistant_knowledge", "snippet": _truncate(snippet, 360), "content": content, "source": "助手知识库", "score": float(result.get("score") or 0), "retrieval_mode": str(result.get("retrieval_mode") or "keyword"), "url": "/page/workspace/knowledge/assistant", "skillName": "assistant_knowledge_search", "mode": "citation", "raw": raw, } def _prompt_graph_node_label(node: Mapping[str, Any], fallback_knowledge_base_id: str) -> str: title = _truncate(str(node.get("title") or "").strip(), 96) slug = _truncate(str(node.get("slug") or "").strip(), 96) knowledge_base_id = _truncate( str( node.get("knowledge_base_id") or node.get("base_id") or fallback_knowledge_base_id or "" ).strip(), 128, ) identity = [] if knowledge_base_id: identity.append(f"知识空间={knowledge_base_id}") if slug: identity.append(f"Wiki={slug}") label = title or slug or "未命名节点" return f"{label} [{'; '.join(identity)}]" if identity else label def _prompt_graph_path(raw: Mapping[str, Any], knowledge_base_id: str) -> str: path = raw.get("graph_path") if isinstance(raw.get("graph_path"), list) else [] nodes = [node for node in path[:3] if isinstance(node, Mapping)] if not nodes: return "" rendered = _prompt_graph_node_label(nodes[0], knowledge_base_id) for node in nodes[1:]: direction = str(node.get("direction") or "related").strip().lower() edge_type = _truncate(str(node.get("edge_type") or "link").strip(), 48) edge_label = f"{direction}/{edge_type}" node_label = _prompt_graph_node_label(node, knowledge_base_id) if direction == "incoming": rendered += f" <--{edge_label}-- {node_label}" else: rendered += f" --{edge_label}--> {node_label}" return _truncate(rendered, 768) def _build_prompt( query: str, sources: list[dict[str, Any]], trace: Mapping[str, Any] | None = None, ) -> str: if not sources: return ( "当前用户选择了知识空间，本轮只能使用所选知识空间，禁止使用 Web、MCP、技能或其他外部检索来源。" "DeerFlow 未检索到可用片段；请直接说明所选知识空间中暂未找到相关资料，不要用外部知识补答，" "不要编造知识库内容或引用编号。不要向用户展示检索通道状态、内部错误码或系统诊断信息。" ) lines = [ "当前用户选择了知识库。DeerFlow 已在模型回答前自动检索到以下背景资料。", "本轮只能依据所选知识空间回答，禁止使用 Web、MCP、技能或其他外部检索来源；不得编造或用常识补齐。", "不要在回答中展示检索通道状态、内部错误码或系统诊断信息，也不要单独生成“资料不足与检索局限说明”等技术性提示。", "引用规则：正文关键结论必须使用 [1]、[2] 这样的编号引用下方资料；只能引用存在的编号；不要在正文末尾重复输出完整参考文献列表，前端会展示引用卡片。", "下方每个编号条目都已被检索并放入本轮上下文；页面标题与 Wiki 标识可能不同。若图扩展路径末节点的（知识空间标识，Wiki 标识）与某条目相同，应视为该末节点已检索，禁止声称其内容未被检索。", "", f"用户问题：{query}", "", f'', ] for source in sources: raw = source.get("raw") if isinstance(source.get("raw"), Mapping) else {} score = source.get("score") score_label = None if isinstance(score, (int, float)): semantics = str(raw.get("rank_score_semantics") or "") label = ( "融合排序分" if semantics == "rrf" else "图扩展分" if semantics == "graph_decay" else "上游通道分" if semantics == "provider_relevance" else "检索分" ) score_label = f"{label}={score:.4f}" meta = " / ".join( str(item) for item in ( source.get("source"), score_label, ) if item ) lines.append(f"[{source['index']}] {source.get('title') or '知识库片段'}") if meta: lines.append(f"来源：{meta}") knowledge_base_id = str( raw.get("knowledge_base_id") or raw.get("base_id") or "" ).strip() wiki_slug = str(raw.get("wiki_slug") or raw.get("slug") or "").strip() graph_hop = raw.get("graph_hop") graph_path = _prompt_graph_path(raw, knowledge_base_id) retrieval_meta = [] if knowledge_base_id: retrieval_meta.append(f"知识空间标识={_truncate(knowledge_base_id, 128)}") if wiki_slug: retrieval_meta.append(f"Wiki 标识={_truncate(wiki_slug, 128)}") if isinstance(graph_hop, (int, float)): retrieval_meta.append(f"图扩展=第{int(graph_hop)}跳") if graph_path: retrieval_meta.append(f"图扩展路径={graph_path}") if retrieval_meta: lines.append("检索元数据：" + " / ".join(retrieval_meta)) lines.append(_truncate(str(source.get("content") or source.get("snippet") or ""), _MAX_SOURCE_CHARS)) lines.append("") lines.append("") return "\n".join(lines).strip() def _trace_result(source: dict[str, Any]) -> dict[str, Any]: """Return the bounded evidence record exposed by retrieval diagnostics. ``score`` on a Wiki result can be an RRF rank score. Keep it separate from the channel/vector scores so the diagnostic UI never presents the fused rank as a cosine similarity percentage. """ raw = source.get("raw") if isinstance(source.get("raw"), Mapping) else {} sections = [item for item in raw.get("matched_sections") or [] if isinstance(item, Mapping)] matched_section = raw.get("matched_section") if isinstance(matched_section, Mapping): sections.append(matched_section) retrieval_mode = str(source.get("retrieval_mode") or raw.get("retrieval_mode") or "unknown") # A hybrid RRF result keeps the winning channel's raw section score, which # may be lexical rather than vector similarity. Only label a per-result # vector score when the source says it is vector-only (or exposes an # explicit vector_score in a future result contract). explicit_vector_score = raw.get("vector_score") section_vector_score = ( raw.get("channel_score") if retrieval_mode == "vector" and raw.get("kind") == "wiki_section" else None ) vector_scores = ( [float(item.get("score")) for item in sections if isinstance(item.get("score"), (int, float))] if retrieval_mode == "vector" else [] ) knowledge_base_id = raw.get("knowledge_base_id") or raw.get("base_id") return { "rank": int(source.get("index") or 0), "id": source.get("id"), "kind": "assistant" if source.get("type") == "assistant_knowledge" else "llmwiki", "knowledge_base_id": knowledge_base_id, "knowledge_base_name": raw.get("knowledge_base_name") or source.get("source"), "title": source.get("title"), "wiki_slug": raw.get("wiki_slug") or raw.get("slug"), "page_type": raw.get("page_type"), "matched_heading": ( raw.get("matched_heading") or raw.get("heading") or (sections[0].get("heading") if sections else None) ), "retrieval_mode": retrieval_mode, "rank_score": float(source.get("score") or 0), "rank_score_semantics": raw.get("rank_score_semantics"), "provider_score": ( float(raw.get("provider_score")) if isinstance(raw.get("provider_score"), (int, float)) else None ), "channel_score": float(raw.get("channel_score")) if isinstance(raw.get("channel_score"), (int, float)) else None, "vector_score": ( float(explicit_vector_score) if isinstance(explicit_vector_score, (int, float)) else float(section_vector_score) if isinstance(section_vector_score, (int, float)) else max(vector_scores) if vector_scores else None ), "graph_hop": ( int(raw.get("graph_hop")) if isinstance(raw.get("graph_hop"), (int, float)) else None ), "graph_score": ( float(raw.get("graph_score")) if isinstance(raw.get("graph_score"), (int, float)) else None ), "graph_path": raw.get("graph_path") if isinstance(raw.get("graph_path"), list) else [], "snippet": source.get("snippet") or "", # This is the exact bounded text made available to the answer prompt. "content": _truncate(str(source.get("content") or source.get("snippet") or ""), _MAX_SOURCE_CHARS), "url": source.get("url") or raw.get("url") or "", "source_refs": raw.get("source_refs") or [], "chunk_refs": raw.get("chunk_refs") or [], } def _new_trace( *, query: str, selected_ids: list[str], assistant_selected_ids: list[str], ) -> dict[str, Any]: return { "retrieval_trace_id": str(uuid4()), "trigger": "pre_retrieval", "query": query, "query_variants": [query] if query else [], "selected_spaces": [ *({"kind": "llmwiki", "id": item} for item in selected_ids), *({"kind": "assistant", "id": item} for item in assistant_selected_ids), ], "authorized_space_count": 0, "mode": "empty", "source_count": 0, "partial": False, "warnings": [], "fallback_reason": None, "fallback_knowledge_base_ids": [], "diagnostics": { "vector_candidates": 0, "keyword_candidates": 0, "wiki_api_candidates": 0, "raw_chunk_candidates": 0, "raw_chunk_considered": 0, "raw_chunk_rejected": 0, "raw_chunk_top_provider_score": None, "final_wiki_page_source_count": 0, "final_raw_chunk_source_count": 0, "assistant_candidates": 0, "graph_candidates": 0, "graph_unique_candidates": 0, "graph_seed_count": 0, "graph_seeds": [], "graph_accepted": 0, "graph_sources_used": 0, "graph_rejected": 0, "graph_hops_requested": 1, "graph_hops_executed": 0, "graph_depth_reason": "default", "graph_paths": [], "graph_channels": [], "rerank_candidates": 0, "top_raw_vector_score": None, "min_similarity": None, "threshold_filtered": 0, }, "timings_ms": { "ordinary": 0.0, "raw_chunks": 0.0, "graph_expansion": 0.0, "assistant": 0.0, "total": 0.0, }, "results": [], "zero_result_reason": None, "capabilities": { "wiki_page_retrieval": True, "vector_retrieval": True, "keyword_retrieval": True, "raw_document_chunk_retrieval": False, "wiki_graph_expansion": False, }, } def _finish_trace(trace: dict[str, Any], sources: list[dict[str, Any]], started_at: float) -> dict[str, Any]: trace["source_count"] = len(sources) trace["results"] = [_trace_result(source) for source in sources] diagnostics = trace.setdefault("diagnostics", {}) diagnostics["final_raw_chunk_source_count"] = sum( 1 for source in sources if str(source.get("retrieval_mode") or "").strip().lower() == "raw_chunk" or str( (source.get("raw") if isinstance(source.get("raw"), Mapping) else {}).get( "retrieval_mode" ) or "" ).strip().lower() == "raw_chunk" ) diagnostics["final_wiki_page_source_count"] = sum( 1 for source in sources if str(source.get("retrieval_mode") or "").strip().lower() != "raw_chunk" and ( str(source.get("type") or "") == "llmwiki_wiki" or bool( str( (source.get("raw") if isinstance(source.get("raw"), Mapping) else {}).get( "wiki_slug" ) or (source.get("raw") if isinstance(source.get("raw"), Mapping) else {}).get( "slug" ) or "" ).strip() ) ) ) trace["timings_ms"]["total"] = round((perf_counter() - started_at) * 1000, 2) trace["warnings"] = list(dict.fromkeys(str(item) for item in trace.get("warnings") or [])) if sources: trace["zero_result_reason"] = None elif not trace.get("zero_result_reason"): diagnostics = trace.get("diagnostics") or {} trace["zero_result_reason"] = ( diagnostics.get("fallback_reason") or ("VECTOR_BELOW_THRESHOLD" if diagnostics.get("threshold_filtered") else None) or ("RETRIEVAL_PROVIDER_UNAVAILABLE" if trace.get("warnings") else None) or "KEYWORD_NO_MATCH" ) return trace async def build_llmwiki_rag_context( *, request: Request, body: Any, graph_input: dict[str, Any], fallback_metadata: Mapping[str, Any] | None = None, ) -> LlmWikiRagContext: selection_declared = _body_declares_selection( body, "llmwiki_knowledge_base_ids", ) or _body_declares_selection(body, "assistant_knowledge_base_ids") selected_ids = resolve_requested_llmwiki_knowledge_base_ids(body) assistant_selected_ids = resolve_requested_assistant_knowledge_base_ids(body) # A normal frontend repeats the selection on every turn. If an older or # embedded client omits the keys, recover the thread's persisted scope; # an explicit empty list still means "search no knowledge spaces". if not _body_declares_selection(body, "llmwiki_knowledge_base_ids") and fallback_metadata: selected_ids = _as_id_list(fallback_metadata.get("llmwiki_knowledge_base_ids")) if not _body_declares_selection(body, "assistant_knowledge_base_ids") and fallback_metadata: assistant_selected_ids = _as_id_list(fallback_metadata.get("assistant_knowledge_base_ids")) if not selected_ids and not assistant_selected_ids: if selection_declared: query = _latest_human_query(graph_input) started_at = perf_counter() trace = _new_trace( query=query, selected_ids=[], assistant_selected_ids=[], ) trace["mode"] = "disabled" trace["zero_result_reason"] = "NO_KNOWLEDGE_SELECTED" return LlmWikiRagContext( [], [], 0, "", [], query, _finish_trace(trace, [], started_at), ) return LlmWikiRagContext([], [], 0, "", [], "") query = _latest_human_query(graph_input) started_at = perf_counter() trace = _new_trace( query=query, selected_ids=selected_ids, assistant_selected_ids=assistant_selected_ids, ) if not query: trace["zero_result_reason"] = "QUERY_EMPTY" return LlmWikiRagContext( selected_ids, assistant_selected_ids, 0, "", [], "", _finish_trace(trace, [], started_at), ) ordinary_sources: list[dict[str, Any]] = [] wiki_recall_sources: list[dict[str, Any]] = [] raw_chunk_sources: list[dict[str, Any]] = [] graph_sources: list[dict[str, Any]] = [] assistant_sources: list[dict[str, Any]] = [] assistant_graph_sources: list[dict[str, Any]] = [] graph_detail_batches: list[Mapping[str, Any]] = [] ordinary_payload: dict[str, Any] = {} ordinary_rows: list[dict[str, Any]] = [] raw_chunk_task: asyncio.Task[list[dict[str, Any]]] | None = None wiki_sources_present = False raw_chunk_sources_present = False graph_sources_present = False runtime = get_resolved_llmwiki_runtime(getattr(request.app.state, "config", None)) local_config = request.app.state.config.llmwiki.local_wiki_index title_boost_enabled = bool(getattr(local_config, "title_boost_enabled", True)) title_boost_weight = float(getattr(local_config, "title_boost_weight", 0.25)) user_id, is_admin = await _actor(request) if selected_ids: llmwiki_available = (local_config.enabled and local_config.internal_search_enabled) or runtime.weknora_enabled store = _store(request) if not llmwiki_available or store is None: trace["partial"] = True trace["warnings"].append("RETRIEVAL_PROVIDER_UNAVAILABLE") trace["zero_result_reason"] = "RETRIEVAL_PROVIDER_UNAVAILABLE" if not assistant_selected_ids: prompt = "Wiki 检索暂不可用。请明确说明系统检索不可用，不要表述为资料不存在。" return LlmWikiRagContext( selected_ids, assistant_selected_ids, 0, prompt, [], query, _finish_trace(trace, [], started_at), ) else: for mapping_id in selected_ids: row = await store.get_authorized(mapping_id, user_id, write=False, is_admin=is_admin) if row is None: raise HTTPException(status_code=404, detail=f"Knowledge base {mapping_id} not found") ordinary_rows.append(row) ordinary_by_id = {str(row["id"]): row for row in ordinary_rows} for space in trace["selected_spaces"]: if space["kind"] == "llmwiki" and space["id"] in ordinary_by_id: space["name"] = ordinary_by_id[space["id"]].get("name") or "" trace["authorized_space_count"] += len(ordinary_rows) search_service = getattr(request.app.state, "llmwiki_retrieval_service", None) graph_store = getattr(request.app.state, "llmwiki_index_store", None) graph_available = callable(getattr(graph_store, "get_adjacent_pages", None)) trace["capabilities"]["wiki_graph_expansion"] = graph_available raw_client = getattr(search_service, "client", None) remote_mapping_by_id = { str(row.get("weknora_id") or ""): row for row in ordinary_rows if str(row.get("weknora_id") or "") } if runtime.weknora_enabled and raw_client is not None and remote_mapping_by_id: trace["capabilities"]["raw_document_chunk_retrieval"] = True remote_ids = list(remote_mapping_by_id) async def search_raw_chunks() -> list[dict[str, Any]]: raw_started_at = perf_counter() try: return await raw_client.search(query, remote_ids) finally: trace["timings_ms"]["raw_chunks"] = round( (perf_counter() - raw_started_at) * 1000, 2, ) raw_chunk_task = asyncio.create_task(search_raw_chunks()) if search_service is None: trace["partial"] = True trace["warnings"].append("RETRIEVAL_PROVIDER_UNAVAILABLE") trace["zero_result_reason"] = "RETRIEVAL_PROVIDER_UNAVAILABLE" if not assistant_selected_ids and raw_chunk_task is None: prompt = "Wiki 检索暂不可用。请明确说明系统检索不可用，不要表述为资料不存在。" return LlmWikiRagContext( selected_ids, assistant_selected_ids, 0, prompt, [], query, _finish_trace(trace, [], started_at), ) ordinary_started_at = perf_counter() try: ordinary_payload = ( await search_service.search(query, ordinary_rows, top_k_pages=_MAX_SOURCES) if search_service is not None else {"results": []} ) except asyncio.CancelledError: if raw_chunk_task is not None: if not raw_chunk_task.done(): raw_chunk_task.cancel() await asyncio.gather(raw_chunk_task, return_exceptions=True) raise except WikiIndexError as exc: logger.warning( "Wiki-only pre-retrieval failed user=%s kb_count=%s code=%s", sanitize_log_param(user_id), len(ordinary_rows), exc.code, ) trace["partial"] = True trace["warnings"].append(exc.code) trace["zero_result_reason"] = exc.code if not assistant_selected_ids and raw_chunk_task is None: prompt = f"Wiki 检索暂不可用（{exc.code}）。请明确说明系统检索状态，不要声称知识库没有资料。" return LlmWikiRagContext( selected_ids, assistant_selected_ids, 0, prompt, [], query, _finish_trace(trace, [], started_at), ) ordinary_payload = {"results": []} except Exception: # noqa: BLE001 - expose provider failure as retrieval diagnostics logger.exception( "Wiki pre-retrieval failed unexpectedly user=%s kb_count=%s", sanitize_log_param(user_id), len(ordinary_rows), ) trace["partial"] = True trace["warnings"].append("RETRIEVAL_PROVIDER_UNAVAILABLE") trace["zero_result_reason"] = "RETRIEVAL_PROVIDER_UNAVAILABLE" if not assistant_selected_ids and raw_chunk_task is None: prompt = ( "Wiki 检索暂不可用（RETRIEVAL_PROVIDER_UNAVAILABLE）。" "请明确说明系统检索状态，不要声称知识库没有资料。" ) return LlmWikiRagContext( selected_ids, assistant_selected_ids, 0, prompt, [], query, _finish_trace(trace, [], started_at), ) ordinary_payload = {"results": []} finally: trace["timings_ms"]["ordinary"] = round((perf_counter() - ordinary_started_at) * 1000, 2) trace["mode"] = str(ordinary_payload.get("mode") or "empty") trace["partial"] = trace["partial"] or bool(ordinary_payload.get("partial")) trace["warnings"].extend(str(item) for item in ordinary_payload.get("warnings") or []) trace["fallback_knowledge_base_ids"] = [ str(item) for item in ordinary_payload.get("fallback_knowledge_base_ids") or [] ] ordinary_diagnostics = ordinary_payload.get("diagnostics") if isinstance(ordinary_diagnostics, Mapping): trace["diagnostics"].update(dict(ordinary_diagnostics)) trace["fallback_reason"] = ordinary_diagnostics.get("fallback_reason") mapping_by_id = {str(row["id"]): row for row in ordinary_rows} for result in ordinary_payload.get("results") or []: mapping = mapping_by_id.get(str(result.get("knowledge_base_id") or "")) if mapping is None: continue source = _source_from_result(result, mapping, len(ordinary_sources) + 1) if not str(source.get("content") or "").strip(): continue ordinary_sources.append(source) if len(ordinary_sources) >= _MAX_SOURCES: break wiki_recall_sources = list(ordinary_sources) if raw_chunk_task is not None: try: raw_rows = await raw_chunk_task except Exception: # noqa: BLE001 - preserve successful Wiki results logger.exception( "Raw chunk pre-retrieval failed user=%s kb_count=%s", sanitize_log_param(user_id), len(ordinary_rows), ) trace["partial"] = True trace["warnings"].append("RAW_CHUNK_RETRIEVAL_FAILED") raw_rows = [] trace["diagnostics"]["raw_chunk_candidates"] = len(raw_rows) authorized_raw_rows: list[tuple[dict[str, Any], dict[str, Any]]] = [] rejected = 0 for raw_row in raw_rows: remote_id = str(raw_row.get("knowledge_base_id") or "") mapping = remote_mapping_by_id.get(remote_id) if mapping is None or not str(raw_row.get("content") or "").strip(): rejected += 1 continue title_score = title_match_strength( query, raw_row.get("title") or raw_row.get("knowledge_title") or raw_row.get("filename"), ) authorized_raw_rows.append( ({**raw_row, "title_match_score": title_score}, mapping) ) authorized_raw_rows.sort( key=lambda item: float(item[0].get("score") or 0) + ( title_boost_weight * float(item[0].get("title_match_score") or 0) if title_boost_enabled else 0.0 ), reverse=True, ) trace["diagnostics"]["raw_chunk_rejected"] = rejected considered_rows = authorized_raw_rows[: _MAX_SOURCES * 4] trace["diagnostics"]["raw_chunk_considered"] = len(considered_rows) if considered_rows: trace["diagnostics"]["raw_chunk_top_provider_score"] = float( considered_rows[0][0].get("score") or 0 ) for raw_row, mapping in considered_rows: raw_chunk_sources.append( _source_from_raw_chunk_result( raw_row, mapping, len(raw_chunk_sources) + 1, ) ) wiki_sources_present = bool(ordinary_sources) raw_chunk_sources_present = bool(raw_chunk_sources) if wiki_sources_present and raw_chunk_sources_present: # Preserve both evidence layers. Raw chunks stay first for an # equal-rank tie, while the quota prevents repeated chunks from # crowding every formatted Wiki page out of the final batch. ordinary_sources = _rrf_merge_source_channels( [raw_chunk_sources, ordinary_sources], limit=_MAX_SOURCES, query=query, title_boost_enabled=title_boost_enabled, title_boost_weight=title_boost_weight, min_items_per_channel=_MAX_SOURCES // 2, ) elif raw_chunk_sources_present: ordinary_sources = raw_chunk_sources[:_MAX_SOURCES] if graph_available: graph_started_at = perf_counter() try: graph_sources, graph_details = await _expand_wiki_graph( index_store=graph_store, mappings=ordinary_rows, wiki_sources=wiki_recall_sources, query=query, retrieval_diagnostics=trace["diagnostics"], ) except asyncio.CancelledError: raise except Exception: # noqa: BLE001 - graph is an optional retrieval channel logger.exception( "Wiki graph expansion failed user=%s kb_count=%s", sanitize_log_param(user_id), len(ordinary_rows), ) trace["partial"] = True trace["warnings"].append("WIKI_GRAPH_EXPANSION_FAILED") graph_sources = [] graph_details = {} finally: trace["timings_ms"]["graph_expansion"] = round( float(trace["timings_ms"].get("graph_expansion") or 0) + (perf_counter() - graph_started_at) * 1000, 2, ) if graph_details: graph_detail_batches.append(graph_details) graph_sources_present = bool(graph_sources) if graph_sources_present: ordinary_sources = _merge_graph_sources( ordinary_sources, graph_sources, deep=bool(graph_details.get("graph_hops_requested", 1) > 1), limit=_MAX_SOURCES, ) if assistant_selected_ids: assistant_store = _assistant_store(request) if assistant_store is not None: # Assistant imports materialize source-document chunks as active # Wiki pages and ``search_pages(..., include_chunks=True)`` searches # them alongside generated Wiki pages. Report the user-facing raw # chunk capability for this path too; otherwise the debug page can # show an actual chunk hit while misleadingly saying the feature is # unsupported merely because no ordinary remote library was chosen. trace["capabilities"]["raw_document_chunk_retrieval"] = True assistant_graph_available = callable( getattr(assistant_store, "get_adjacent_pages", None) ) trace["capabilities"]["wiki_graph_expansion"] = bool( trace["capabilities"].get("wiki_graph_expansion") or assistant_graph_available ) assistant_bases: list[dict[str, Any]] = [] get_assistant_base = getattr(assistant_store, "get_base", None) for base_id in assistant_selected_ids: base = await get_assistant_base(base_id) if callable(get_assistant_base) else {"id": base_id} if base is None: raise HTTPException(status_code=404, detail=f"Assistant knowledge base {base_id} not found") assistant_bases.append(base) assistant_by_id = {str(base["id"]): base for base in assistant_bases} for space in trace["selected_spaces"]: if space["kind"] == "assistant" and space["id"] in assistant_by_id: space["name"] = assistant_by_id[space["id"]].get("name") or "" trace["authorized_space_count"] += len(assistant_bases) assistant_started_at = perf_counter() try: assistant_rows = await assistant_store.search_pages( base_ids=[str(base["id"]) for base in assistant_bases], query=query, limit=_MAX_SOURCES, include_chunks=True, embedding_client=getattr(request.app.state, "llmwiki_embedding", None), ) except Exception: # noqa: BLE001 - an unavailable channel must not masquerade as an empty library logger.exception( "Assistant knowledge pre-retrieval failed user=%s kb_count=%s", sanitize_log_param(user_id), len(assistant_bases), ) trace["partial"] = True trace["warnings"].append("ASSISTANT_RETRIEVAL_FAILED") trace["zero_result_reason"] = "RETRIEVAL_PROVIDER_UNAVAILABLE" assistant_rows = [] trace["timings_ms"]["assistant"] = round((perf_counter() - assistant_started_at) * 1000, 2) trace["diagnostics"]["assistant_candidates"] = len(assistant_rows) for row in assistant_rows: assistant_sources.append(_source_from_assistant_result(row, len(assistant_sources) + 1)) if len(assistant_sources) >= _MAX_SOURCES: break assistant_recall_sources = list(assistant_sources) if assistant_graph_available: assistant_graph_started_at = perf_counter() try: assistant_graph_sources, assistant_graph_details = await _expand_wiki_graph( index_store=assistant_store, mappings=assistant_bases, wiki_sources=assistant_recall_sources, query=query, retrieval_diagnostics=trace["diagnostics"], source_kind="assistant", ) except asyncio.CancelledError: raise except Exception: # noqa: BLE001 - graph remains an optional assistant channel logger.exception( "Assistant knowledge graph expansion failed user=%s kb_count=%s", sanitize_log_param(user_id), len(assistant_bases), ) trace["partial"] = True trace["warnings"].append("ASSISTANT_GRAPH_EXPANSION_FAILED") assistant_graph_sources = [] assistant_graph_details = {} finally: trace["timings_ms"]["graph_expansion"] = round( float(trace["timings_ms"].get("graph_expansion") or 0) + (perf_counter() - assistant_graph_started_at) * 1000, 2, ) if assistant_graph_details: graph_detail_batches.append(assistant_graph_details) if assistant_graph_sources: graph_sources_present = True assistant_sources = _merge_graph_sources( assistant_sources, assistant_graph_sources, deep=bool( assistant_graph_details.get("graph_hops_requested", 1) > 1 ), limit=_MAX_SOURCES, ) else: trace["partial"] = True trace["warnings"].append("ASSISTANT_RETRIEVAL_PROVIDER_UNAVAILABLE") combined_graph_details = _combine_graph_details(graph_detail_batches) if combined_graph_details: trace["diagnostics"].update(combined_graph_details) # Search both selected channels deeply enough to avoid losing ordinary # ranks 5-8 when the assistant channel has few or zero hits. Reserve four # slots per populated channel, then backfill every unused slot from the # other channel; the final prompt remains capped at eight sources. if ordinary_sources and assistant_sources: ordinary_head = ordinary_sources[: _MAX_SOURCES // 2] assistant_head = assistant_sources[: _MAX_SOURCES // 2] sources = [*ordinary_head, *assistant_head] remaining = [ *ordinary_sources[len(ordinary_head) :], *assistant_sources[len(assistant_head) :], ] sources.extend(remaining[: _MAX_SOURCES - len(sources)]) else: sources = (ordinary_sources or assistant_sources)[:_MAX_SOURCES] sources = [{**source, "index": index} for index, source in enumerate(sources, start=1)] trace["diagnostics"]["graph_sources_used"] = sum( 1 for source in sources if source.get("retrieval_mode") == "graph_expansion" ) assistant_modes = { str(source.get("retrieval_mode") or "keyword") for source in assistant_sources } if ordinary_sources and assistant_sources: trace["mode"] = "mixed" elif assistant_sources: if assistant_graph_sources: trace["mode"] = "assistant_graph_hybrid" else: trace["mode"] = ( next(iter(assistant_modes)) if len(assistant_modes) == 1 else "hybrid" ) elif graph_sources_present and raw_chunk_sources_present: trace["mode"] = "wiki_graph_raw_hybrid" elif graph_sources_present: trace["mode"] = "wiki_graph_hybrid" elif wiki_sources_present and raw_chunk_sources_present: trace["mode"] = "wiki_raw_hybrid" elif raw_chunk_sources_present: trace["mode"] = "raw_chunk" elif ordinary_payload: trace["mode"] = str(ordinary_payload.get("mode") or "empty") if trace["fallback_reason"] and not sources: trace["zero_result_reason"] = trace["fallback_reason"] finished_trace = _finish_trace(trace, sources, started_at) return LlmWikiRagContext( selected_ids, assistant_selected_ids, len(sources), _build_prompt(query, sources, finished_trace), sources, query, finished_trace, ) def attach_llmwiki_rag_to_graph_input( graph_input: dict[str, Any], rag: LlmWikiRagContext, ) -> dict[str, Any]: if not rag.prompt and not rag.sources and not rag.trace: return graph_input messages = graph_input.get("messages") if not isinstance(messages, list) or not messages: return graph_input new_messages = list(messages) for index in range(len(new_messages) - 1, -1, -1): message = new_messages[index] if not isinstance(message, HumanMessage): continue extra = dict(getattr(message, "additional_kwargs", None) or {}) if rag.prompt: existing = str(extra.get("prompt_prefix") or "").strip() extra["prompt_prefix"] = f"{existing}\n\n{rag.prompt}".strip() if existing else rag.prompt extra["llmwiki_knowledge_base_ids"] = rag.knowledge_base_ids extra["assistant_knowledge_base_ids"] = rag.assistant_knowledge_base_ids extra["llmwiki_reference_query"] = rag.query if rag.trace: extra["knowledge_retrieval_trace"] = rag.trace if rag.sources: extra["llmwiki_reference_sources"] = rag.sources new_messages[index] = message.model_copy(update={"additional_kwargs": extra}) break return {**graph_input, "messages": new_messages}