Feat: Compilation benefit naive rag. (#17555)

### Summary

Compilation benifit naive rag

---------

Co-authored-by: Yingfeng Zhang <yingfeng.zhang@gmail.com>
This commit is contained in:
Kevin Hu
2026-07-30 18:39:01 +08:00
committed by GitHub
parent 1667d1495f
commit 3f8a3dfcff
16 changed files with 901 additions and 196 deletions

View File

@@ -637,7 +637,7 @@ async def get_document_structure_graph(tenant_id, dataset_id, document_id):
config = template.get("config") if isinstance(template.get("config"), dict) else {}
raw_kind = (config.get("kind") if isinstance(config, dict) else "") or template.get("kind") or ""
kind_norm = _compilation_template_kind(raw_kind)
if kind_norm == "artifacts":
if kind_norm == "wiki":
continue
seen_configured_ids.add(template_id)
configured_ids.append(template_id)

View File

@@ -792,16 +792,21 @@ def delete_dataset_structure(tenant_id, dataset_id):
@login_required
@add_tenant_id_to_kwargs
async def get_wiki_alteration(tenant_id, dataset_id):
"""Return document drift for the dataset Artifact wiki.
"""Return document drift for a compiled dataset product.
GET /api/v1/datasets/<dataset_id>/artifacts/alteration
GET /api/v1/datasets/<dataset_id>/artifacts/alteration?kind=<kind>
``kind`` (default ``wiki``) is one of: wiki | graph | mindmap | timeline |
tree (tree covers both ``tree`` and ``page_index``).
Success: {"code": 0, "data": {"removed": int, "newly_uploaded": int, ...}}
"""
kind = (request.args.get("kind") or "wiki").strip().lower()
if kind not in {"wiki", "graph", "mindmap", "timeline", "tree"}:
return get_error_data_result(message=f"Unsupported kind: {kind!r}. Expected one of: wiki, graph, mindmap, timeline, tree.")
try:
success, result = await dataset_api_service.get_wiki_alteration(
dataset_id,
tenant_id,
)
if kind == "wiki":
success, result = await dataset_api_service.get_wiki_alteration(dataset_id, tenant_id)
else:
success, result = await dataset_api_service.get_structure_alteration(dataset_id, tenant_id, kind)
if success:
return get_result(data=result)
return get_result(data=False, message=result, code=RetCode.AUTHENTICATION_ERROR)

View File

@@ -56,7 +56,7 @@ def validate_template_payload(req: dict, require_all: bool = True) -> str:
return f"{section.capitalize()} field description is too long."
if len(str((field or {}).get("rule") or "")) > 1024:
return f"{section.capitalize()} field rule is too long."
if config.get("kind") == "artifacts" or req.get("kind") == "artifacts":
if config.get("kind") == "wiki" or req.get("kind") == "wiki":
for field in (config.get("claim") or {}).get("fields") or []:
if not str((field or {}).get("statement") or "").strip():
return "Claim statement is required."

View File

@@ -50,13 +50,13 @@ _STRUCTURE_INDEX_TYPE_TO_KIND = {
}
_STRUCTURE_INDEX_TYPES = frozenset(_STRUCTURE_INDEX_TYPE_TO_KIND)
_VALID_INDEX_TYPES = {"graph", "raptor", "mindmap", "artifact", "skill"} | set(_STRUCTURE_INDEX_TYPES)
_VALID_INDEX_TYPES = {"graph", "raptor", "mindmap", "wiki", "skill"} | set(_STRUCTURE_INDEX_TYPES)
_INDEX_TYPE_TO_TASK_TYPE = {
"graph": "structure_graph",
"raptor": "raptor",
"mindmap": "structure_mindmap",
"artifact": "artifact",
"wiki": "wiki",
"skill": "skill",
# Structure merge types carry their own task_type (== index_type) so the
# executor can resolve which kind to merge from the task body.
@@ -67,7 +67,7 @@ _INDEX_TYPE_TO_TASK_ID_FIELD = {
"graph": "graphrag_task_id",
"raptor": "raptor_task_id",
"mindmap": "mindmap_task_id",
"artifact": "wiki_task_id",
"wiki": "wiki_task_id",
"skill": "skill_task_id",
**{t: f"{t}_task_id" for t in _STRUCTURE_INDEX_TYPES},
}
@@ -76,7 +76,7 @@ _INDEX_TYPE_TO_DISPLAY_NAME = {
"graph": "Graph",
"raptor": "RAPTOR",
"mindmap": "Mindmap",
"artifact": "Artifact",
"wiki": "Wiki",
"skill": "Skill",
"structure_graph": "Structure Graph",
"structure_mindmap": "Structure Mindmap",
@@ -1614,40 +1614,53 @@ def _extract_pipeline_compiler_group_ids(dsl) -> list[str]:
return group_ids
def _template_is_wiki(template: dict | None) -> bool:
def _normalize_template_kind(raw) -> str:
"""Lowercase a template's raw kind WITHOUT folding.
Unlike :func:`_compilation_template_kind` (which folds ``page_index`` /
``knowledge_graph`` into ``timeline``), this keeps kinds distinct so the
alteration API can tell ``graph`` / ``timeline`` / ``tree`` apart.
"""
if not isinstance(raw, str):
return ""
return raw.strip().lower().replace("-", "_")
def _template_matches_kind(template: dict | None, accepted: set[str]) -> bool:
if not isinstance(template, dict):
return False
config = template.get("config") if isinstance(template.get("config"), dict) else {}
raw_kind = config.get("kind") or template.get("kind") or ""
return _compilation_template_kind(raw_kind) == "artifacts"
return _normalize_template_kind(raw_kind) in accepted
def _group_has_wiki_template(group_id: str, tenant_id: str, group_cache: dict[str, bool]) -> bool:
def _group_has_template_of_kind(group_id: str, tenant_id: str, accepted: set[str], group_cache: dict[str, bool]) -> bool:
if group_id in group_cache:
return group_cache[group_id]
from api.db.services.compilation_template_group_service import CompilationTemplateGroupService
group = CompilationTemplateGroupService.get_saved(group_id, tenant_id)
has_wiki = any(_template_is_wiki(template) for template in (group or {}).get("templates") or [])
group_cache[group_id] = has_wiki
return has_wiki
matched = any(_template_matches_kind(template, accepted) for template in (group or {}).get("templates") or [])
group_cache[group_id] = matched
return matched
def _parser_config_has_wiki_template(parser_config, tenant_id: str, template_cache: dict[str, bool]) -> bool:
def _parser_config_has_template_of_kind(parser_config, tenant_id: str, accepted: set[str], template_cache: dict[str, bool]) -> bool:
from api.db.services.compilation_template_service import CompilationTemplateService
from rag.svr.task_executor_refactor.chunk_post_processor import _parser_config_compilation_template_ids
for template_id in _parser_config_compilation_template_ids(parser_config, tenant_id):
if template_id not in template_cache:
template_cache[template_id] = _template_is_wiki(CompilationTemplateService.get_saved(template_id, tenant_id))
template_cache[template_id] = _template_matches_kind(CompilationTemplateService.get_saved(template_id, tenant_id), accepted)
if template_cache[template_id]:
return True
return False
def _pipeline_has_wiki_compiler(
def _pipeline_has_compiler_of_kind(
pipeline_id: str,
tenant_id: str,
accepted: set[str],
pipeline_cache: dict[str, bool],
group_cache: dict[str, bool],
) -> bool:
@@ -1665,9 +1678,9 @@ def _pipeline_has_wiki_compiler(
return False
group_ids = _extract_pipeline_compiler_group_ids(getattr(canvas, "dsl", None))
has_wiki = any(_group_has_wiki_template(group_id, tenant_id, group_cache) for group_id in group_ids)
pipeline_cache[pipeline_id] = has_wiki
return has_wiki
matched = any(_group_has_template_of_kind(group_id, tenant_id, accepted, group_cache) for group_id in group_ids)
pipeline_cache[pipeline_id] = matched
return matched
def _skill_index_or_none(tenant_id: str, kb_id: str):
@@ -1841,48 +1854,63 @@ async def get_dataset_structure(dataset_id: str, tenant_id: str, kind: str, keyw
# fields only) to enumerate the distinct template ids whose top-level kind
# matches the request; ``build_bucket`` below then reads each template's rows.
meta_fields = ["id", "compile_kwd", "compilation_template_ids", "compilation_template_kind_kwd"]
kind_template_ids: list[str] = []
seen_tid: set[str] = set()
has_templateless = False
offset = 0
page_size = 1000
pages = 0
while True:
pages += 1
try:
res = await thread_pool_exec(
settings.docStoreConn.search,
meta_fields,
[],
{"knowledge_graph_kwd": ["entity", "relation"], "scope_kwd": ["dataset"]},
[],
OrderByExpr(),
offset,
page_size,
index_nm,
[dataset_id],
)
meta_rows = settings.docStoreConn.get_fields(res, meta_fields) or {}
except Exception:
logging.exception("get_dataset_structure: docStore discovery failed for kb=%s", dataset_id)
return True, empty
if not meta_rows:
break
for row in meta_rows.values():
tid = _row_template_id(row)
stamped_kind = (row.get("compilation_template_kind_kwd") or "").strip()
row_kind = stamped_kind or _template_meta(tid) or ""
if _resolve_dataset_structure_kind(row_kind) != resolved_kind:
continue
if tid:
if tid not in seen_tid:
seen_tid.add(tid)
kind_template_ids.append(tid)
else:
has_templateless = True
if len(meta_rows) < page_size:
break
offset += page_size
async def _discover_scope_templates(scope_kwd: str) -> tuple[list[str], bool, int]:
scope_template_ids: list[str] = []
seen_scope_tid: set[str] = set()
scope_has_templateless = False
offset = 0
pages = 0
while True:
pages += 1
try:
res = await thread_pool_exec(
settings.docStoreConn.search,
meta_fields,
[],
{"knowledge_graph_kwd": ["entity", "relation"], "scope_kwd": [scope_kwd]},
[],
OrderByExpr(),
offset,
page_size,
index_nm,
[dataset_id],
)
meta_rows = settings.docStoreConn.get_fields(res, meta_fields) or {}
except Exception:
logging.exception("get_dataset_structure: docStore discovery failed for kb=%s scope=%s", dataset_id, scope_kwd)
return [], False, pages
if not meta_rows:
break
for row in meta_rows.values():
tid = _row_template_id(row)
stamped_kind = row.get("compilation_template_kind_kwd") or ""
if isinstance(stamped_kind, list):
stamped_kind = stamped_kind[0].strip()
row_kind = stamped_kind or _template_meta(tid) or ""
if _resolve_dataset_structure_kind(row_kind) != resolved_kind:
continue
if tid:
if tid not in seen_scope_tid:
seen_scope_tid.add(tid)
scope_template_ids.append(tid)
else:
scope_has_templateless = True
if len(meta_rows) < page_size:
break
offset += page_size
return scope_template_ids, scope_has_templateless, pages
dataset_template_ids, has_templateless, dataset_pages = await _discover_scope_templates("dataset")
kind_template_ids: list[str] = list(dataset_template_ids)
template_scope_by_id: dict[str, str] = {tid: "dataset" for tid in dataset_template_ids}
doc_template_ids, _, doc_pages = await _discover_scope_templates("doc")
for tid in doc_template_ids:
if tid not in template_scope_by_id:
template_scope_by_id[tid] = "doc"
kind_template_ids.append(tid)
# Detect datasets that have ONLY the legacy dataset_graph blob (no
# entity/relation rows yet) so the fallback path below handles them.
@@ -1907,9 +1935,11 @@ async def get_dataset_structure(dataset_id: str, tenant_id: str, kind: str, keyw
pass
logging.debug(
"get_dataset_structure: discovered %d template(s) in %d page(s) for kb=%s kind=%s",
len(kind_template_ids),
pages,
"get_dataset_structure: discovered %d dataset template(s) and %d doc template(s) in %d/%d page(s) for kb=%s kind=%s",
len(dataset_template_ids),
len(doc_template_ids),
dataset_pages,
doc_pages,
dataset_id,
resolved_kind,
)
@@ -1927,17 +1957,20 @@ async def get_dataset_structure(dataset_id: str, tenant_id: str, kind: str, keyw
def _scope_for_template(row: dict):
tid = _row_template_id(row) or ""
stamped = (row.get("compilation_template_kind_kwd") or "").strip()
stamped = row.get("compilation_template_kind_kwd") or ""
if isinstance(stamped, list):
stamped = stamped[0].strip()
scope_kwd = template_scope_by_id.get(tid, "dataset") if tid else "dataset"
meta = _bucket_meta_for(tid, stamped) if tid else {"template_id": f"kind:{resolved_kind}", "template_name": f"kind:{resolved_kind}", "kind": resolved_kind}
if tid:
return meta, {"compilation_template_ids": [tid], "scope_kwd": ["dataset"]}
return meta, {"compilation_template_kind_kwd": [stamped], "scope_kwd": ["dataset"]}
return meta, {"compilation_template_ids": [tid], "scope_kwd": [scope_kwd]}
return meta, {"compilation_template_kind_kwd": [stamped], "scope_kwd": [scope_kwd]}
bucket_meta, kw_entities, kw_relations = await sgc.keyword_subgraph(
index_nm,
dataset_id,
embd_mdl,
{"compilation_template_ids": kind_template_ids, "knowledge_graph_kwd": ["entity"], "scope_kwd": ["dataset"]},
{"compilation_template_ids": kind_template_ids, "knowledge_graph_kwd": ["entity"], "scope_kwd": ["dataset", "doc"]},
keywords,
_scope_for_template,
log_ctx=f"kb={dataset_id}",
@@ -1956,11 +1989,18 @@ async def get_dataset_structure(dataset_id: str, tenant_id: str, kind: str, keyw
# ── normal mode: per-template subgraph sampling from raw KB-wide rows. ──
templates_out: list[dict] = []
for tid in kind_template_ids:
scope_kwd = template_scope_by_id.get(tid, "dataset")
try:
entities, relations = await sgc.build_bucket(index_nm, dataset_id, {"compilation_template_ids": [tid], "scope_kwd": ["dataset"]})
entities, relations = await sgc.build_bucket(index_nm, dataset_id, {"compilation_template_ids": [tid], "scope_kwd": [scope_kwd]})
except Exception:
logging.exception("get_dataset_structure: bucket build failed for kb=%s template=%s", dataset_id, tid)
continue
if scope_kwd == "dataset" and not entities and not relations:
try:
entities, relations = await sgc.build_bucket(index_nm, dataset_id, {"compilation_template_ids": [tid], "scope_kwd": ["doc"]})
except Exception:
logging.exception("get_dataset_structure: doc fallback bucket build failed for kb=%s template=%s", dataset_id, tid)
continue
if resolved_kind in {"knowledge_graph", "mind_map", "timeline"}:
entities = sgc.filter_entities_with_relations(entities, relations)
if not entities:
@@ -1993,7 +2033,10 @@ async def get_dataset_structure(dataset_id: str, tenant_id: str, kind: str, keyw
legacy_rows = {}
legacy_bucket = {"template_id": f"kind:{resolved_kind}", "template_name": f"kind:{resolved_kind}", "kind": resolved_kind, "entities": [], "relations": []}
for row in legacy_rows.values():
if _resolve_dataset_structure_kind((row.get("compilation_template_kind_kwd") or "").strip()) != resolved_kind:
stamped_kind = row.get("compilation_template_kind_kwd") or ""
if isinstance(stamped_kind, list):
stamped_kind = stamped_kind[0].strip()
if _resolve_dataset_structure_kind((stamped_kind or "").strip()) != resolved_kind:
continue
try:
graph = json.loads(row.get("content_with_weight") or "{}")
@@ -2012,14 +2055,39 @@ async def get_dataset_structure(dataset_id: str, tenant_id: str, kind: str, keyw
return True, {"kind": kind, "templates": templates_out}
async def get_wiki_alteration(dataset_id: str, tenant_id: str):
"""Return doc-level drift between current dataset docs and compiled wiki provenance."""
if not KnowledgebaseService.accessible(dataset_id, tenant_id):
return False, "no authorization"
ok, kb = KnowledgebaseService.get_by_id(dataset_id)
if not ok:
return False, "Invalid Dataset ID"
# ── artifacts/alteration: per-``kind`` provenance & eligibility mapping ──
#
# Alteration is a pure set operation shared by every kind:
# removed = involved (product provenance) current (dataset docs)
# newly_uploaded = eligible (should be compiled) involved
# Only two inputs vary by kind: how ``involved`` provenance is gathered, and
# which template kinds make a doc ``eligible``.
# Non-folded template kinds that make a doc eligible for each API kind.
_ALTERATION_ELIGIBLE_TEMPLATE_KINDS = {
"wiki": {"artifacts"},
"graph": {"knowledge_graph"},
"mindmap": {"mind_map"},
"timeline": {"timeline"},
"tree": {"tree", "page_index"},
}
# Dataset-merged structure kinds carry provenance in ``source_doc_ids`` on
# ``scope_kwd="dataset"`` rows, discovered by their stamped top-level kind.
_ALTERATION_KIND_TO_MERGED_ROW_KIND = {
"graph": "knowledge_graph",
"mindmap": "mind_map",
"timeline": "timeline",
}
# Per-document structure kinds keep one product row per document; provenance is
# the row's own ``doc_id`` (no dataset merge, no ``source_doc_ids``). ``tree`` and
# ``page_index`` write distinct ``compile_kwd`` values.
_ALTERATION_TREE_COMPILE_KWDS = ["tree", "page_index"]
async def _current_dataset_docs(dataset_id: str):
"""Return ``(docs, current_doc_id_set)`` for the dataset."""
docs, _ = await thread_pool_exec(
DocumentService.get_by_kb_id,
kb_id=dataset_id,
@@ -2032,81 +2100,147 @@ async def get_wiki_alteration(dataset_id: str, tenant_id: str):
types=[],
suffix=[],
)
current_doc_ids = {str(doc.get("id")) for doc in docs or [] if doc.get("id")}
docs = docs or []
current_doc_ids = {str(doc.get("id")) for doc in docs if doc.get("id")}
return docs, current_doc_ids
wiki_involved_doc_ids: set[str] = set()
pack = _wiki_index_or_none(kb.tenant_id, dataset_id)
if pack is not None:
index_nm, _ = pack
from common.doc_store.doc_store_base import OrderByExpr
select_fields = ["id", "source_doc_ids"]
offset = 0
page_size = 1000
while True:
try:
res = await thread_pool_exec(
settings.docStoreConn.search,
select_fields=select_fields,
highlight_fields=[],
condition={"compile_kwd": [WIKI_PAGE_COMPILE_KWD]},
match_expressions=[],
order_by=OrderByExpr(),
offset=offset,
limit=page_size,
index_names=index_nm,
knowledgebase_ids=[dataset_id],
)
rows = settings.docStoreConn.get_fields(res, select_fields) or {}
except Exception:
logging.exception("get_wiki_alteration: docStore search failed for kb=%s", dataset_id)
rows = {}
def _alteration_result(current_doc_ids: set, involved_doc_ids: set, eligible_doc_ids: set) -> dict:
"""Build the drift response dict shared by every ``kind``."""
removed_doc_ids = sorted(involved_doc_ids - current_doc_ids)
newly_uploaded_doc_ids = sorted(eligible_doc_ids - involved_doc_ids)
return {
"removed": len(removed_doc_ids),
"newly_uploaded": len(newly_uploaded_doc_ids),
"removed_doc_ids": removed_doc_ids,
"newly_uploaded_doc_ids": newly_uploaded_doc_ids,
"involved_doc_ids": sorted(involved_doc_ids),
"eligible_doc_ids": sorted(eligible_doc_ids),
}
if not rows:
break
for row in rows.values():
source_doc_ids = row.get("source_doc_ids")
if isinstance(source_doc_ids, str):
source_doc_ids = [source_doc_ids]
if not isinstance(source_doc_ids, list):
continue
wiki_involved_doc_ids.update(str(doc_id) for doc_id in source_doc_ids if doc_id)
offset += page_size
total = settings.docStoreConn.get_total(res)
if not total or offset >= int(total):
break
def _eligible_doc_ids_for_kind(docs, tenant_id: str, kind: str) -> set:
"""Doc ids whose parser_config or pipeline carries a template of ``kind``."""
accepted = _ALTERATION_ELIGIBLE_TEMPLATE_KINDS.get(kind) or set()
template_cache: dict[str, bool] = {}
group_cache: dict[str, bool] = {}
pipeline_cache: dict[str, bool] = {}
eligible_wiki_doc_ids: set[str] = set()
eligible: set[str] = set()
for doc in docs or []:
doc_id = str(doc.get("id") or "")
if not doc_id:
continue
parser_config = doc.get("parser_config") or {}
if _parser_config_has_wiki_template(parser_config, kb.tenant_id, template_cache):
eligible_wiki_doc_ids.add(doc_id)
if _parser_config_has_template_of_kind(parser_config, tenant_id, accepted, template_cache):
eligible.add(doc_id)
continue
if _pipeline_has_wiki_compiler(
doc.get("pipeline_id") or "",
kb.tenant_id,
pipeline_cache,
group_cache,
):
eligible_wiki_doc_ids.add(doc_id)
if _pipeline_has_compiler_of_kind(doc.get("pipeline_id") or "", tenant_id, accepted, pipeline_cache, group_cache):
eligible.add(doc_id)
return eligible
removed_doc_ids = sorted(wiki_involved_doc_ids - current_doc_ids)
newly_uploaded_doc_ids = sorted(eligible_wiki_doc_ids - wiki_involved_doc_ids)
return True, {
"removed": len(removed_doc_ids),
"newly_uploaded": len(newly_uploaded_doc_ids),
"removed_doc_ids": removed_doc_ids,
"newly_uploaded_doc_ids": newly_uploaded_doc_ids,
"involved_doc_ids": sorted(wiki_involved_doc_ids),
"eligible_doc_ids": sorted(eligible_wiki_doc_ids),
}
async def _involved_doc_ids_paged(index_nm, dataset_id: str, condition: dict, field: str, from_list: bool) -> set:
"""Page a docStore search, folding each row's ``field`` into a doc-id set.
``from_list`` reads ``field`` as ``source_doc_ids`` (a list of provenance doc
ids); otherwise ``field`` is the row's own ``doc_id`` scalar.
"""
from common.doc_store.doc_store_base import OrderByExpr
involved: set[str] = set()
select_fields = ["id", field]
offset = 0
page_size = 1000
while True:
try:
res = await thread_pool_exec(
settings.docStoreConn.search,
select_fields=select_fields,
highlight_fields=[],
condition=condition,
match_expressions=[],
order_by=OrderByExpr(),
offset=offset,
limit=page_size,
index_names=index_nm,
knowledgebase_ids=[dataset_id],
)
rows = settings.docStoreConn.get_fields(res, select_fields) or {}
except Exception:
logging.exception("alteration: docStore search failed for kb=%s cond=%s", dataset_id, condition)
rows = {}
if not rows:
break
for row in rows.values():
value = row.get(field)
if from_list:
if isinstance(value, str):
value = [value]
if isinstance(value, list):
involved.update(str(d) for d in value if d)
elif value:
involved.add(str(value))
offset += page_size
total = settings.docStoreConn.get_total(res)
if not total or offset >= int(total):
break
return involved
async def _involved_doc_ids_for_kind(index_nm, dataset_id: str, kind: str) -> set:
"""Gather the doc ids baked into the compiled product for ``kind``."""
if kind == "wiki":
return await _involved_doc_ids_paged(index_nm, dataset_id, {"compile_kwd": [WIKI_PAGE_COMPILE_KWD]}, "source_doc_ids", from_list=True)
if kind in _ALTERATION_KIND_TO_MERGED_ROW_KIND:
condition = {
"knowledge_graph_kwd": ["entity", "relation"],
"scope_kwd": ["dataset"],
"compilation_template_kind_kwd": [_ALTERATION_KIND_TO_MERGED_ROW_KIND[kind]],
}
return await _involved_doc_ids_paged(index_nm, dataset_id, condition, "source_doc_ids", from_list=True)
if kind == "tree":
return await _involved_doc_ids_paged(index_nm, dataset_id, {"compile_kwd": list(_ALTERATION_TREE_COMPILE_KWDS)}, "doc_id", from_list=False)
return set()
async def _get_alteration(dataset_id: str, tenant_id: str, kind: str):
"""Shared driver: doc-level drift between the ``kind`` product and the dataset."""
if not KnowledgebaseService.accessible(dataset_id, tenant_id):
return False, "no authorization"
ok, kb = KnowledgebaseService.get_by_id(dataset_id)
if not ok:
return False, "Invalid Dataset ID"
docs, current_doc_ids = await _current_dataset_docs(dataset_id)
involved_doc_ids: set[str] = set()
pack = _compiled_index_or_none(kb.tenant_id, dataset_id)
if pack is not None:
index_nm, _ = pack
involved_doc_ids = await _involved_doc_ids_for_kind(index_nm, dataset_id, kind)
eligible_doc_ids = _eligible_doc_ids_for_kind(docs, kb.tenant_id, kind)
return True, _alteration_result(current_doc_ids, involved_doc_ids, eligible_doc_ids)
async def get_wiki_alteration(dataset_id: str, tenant_id: str):
"""Return doc-level drift between current dataset docs and compiled wiki provenance."""
return await _get_alteration(dataset_id, tenant_id, "wiki")
async def get_structure_alteration(dataset_id: str, tenant_id: str, kind: str):
"""Return doc-level drift for a non-wiki structure ``kind``.
``kind`` is one of ``graph`` / ``mindmap`` / ``timeline`` (dataset-merged
products, provenance from ``source_doc_ids``) or ``tree`` (per-document
products covering ``tree`` + ``page_index``, provenance from ``doc_id``).
"""
kind = (kind or "").strip().lower()
if kind not in _ALTERATION_KIND_TO_MERGED_ROW_KIND and kind != "tree":
return False, f"Unsupported structure kind: {kind!r}. Expected one of: graph, mindmap, timeline, tree."
return await _get_alteration(dataset_id, tenant_id, kind)
async def list_wiki_pages(

View File

@@ -1,7 +1,7 @@
kind: artifacts
display_name: Artifacts — Graph-based wiki
kind: wiki
display_name: Wiki — Graph-based wiki
config:
kind: artifacts
kind: wiki
example: |
- Each page must be a proper encyclopedic article, NOT a flat bullet list:
- 1. Opening paragraph (2-4 sentences defining what this is). No heading.

View File

@@ -39,10 +39,10 @@ def _derive_scope(templates: list[dict]) -> str:
if not templates:
raise GroupValidationError("A template group must contain at least one template.")
kinds = [str((t or {}).get("kind") or "").strip() for t in templates]
artifact_count = sum(1 for k in kinds if k == "artifacts")
artifact_count = sum(1 for k in kinds if k == "wiki")
if artifact_count > 0:
if artifact_count != 1 or len(templates) != 1:
raise GroupValidationError("An artifacts template cannot be combined with other templates in the same group.")
raise GroupValidationError("A wiki template cannot be combined with other templates in the same group.")
return SCOPE_DATASET
_enforce_single_rechunk_tree(templates)

View File

@@ -1225,7 +1225,7 @@ def queue_raptor_o_graphrag_tasks(sample_doc, ty, priority, fake_doc_id="", doc_
"graphrag",
"raptor",
"mindmap",
"artifact",
"wiki",
"skill",
# KB-wide structure-graph merge task types (rebuild dataset_graph rows).
"structure_graph",

View File

@@ -185,7 +185,9 @@ class PipelineTaskType(StrEnum):
GRAPH_RAG = "GraphRAG"
MINDMAP = "Mindmap"
MEMORY = "Memory"
ARTIFACT = "Artifact"
# Member name kept as ARTIFACT for back-compat; value is "Wiki" so the
# runtime task_type (``.lower()`` == "wiki") matches the wiki index/task type.
ARTIFACT = "Wiki"
SKILL = "Skill"
# KB-wide structure-graph merge tasks (rebuild_dataset_structure_graph_json).
STRUCTURE_GRAPH = "StructureGraph"

View File

@@ -310,7 +310,7 @@ async def _add_template_group_compilations(comps: set[str], parser_config: dict,
comps.add("page_index")
elif kind in {"mindmap", "mind_map"}:
comps.add("mindmap")
elif kind == "artifacts":
elif kind == "wiki":
comps.add("wiki")

View File

@@ -11,9 +11,9 @@ async def direct_search(state: dict, tools) -> dict:
"""Single hybrid search → merge into kbinfos."""
question = state.get("question", "")
keywords = state.get("keywords", "")
_LOG.info("[Direct search] Looking up the knowledge base for: \"%s\" (keywords: %s)", question, keywords)
_LOG.info('[Direct search] Looking up the knowledge base for: "%s" (keywords: %s)', question, keywords)
result = await hybrid_search(tools, query=question, keywords=keywords)
result = await hybrid_search(tools, query=question, keywords=keywords, use_compiled=True)
_merge_kbinfos(tools, result)
if not _has_chunks(tools):

View File

@@ -543,13 +543,22 @@ async def dataset_navigation_by_tree(tools, topic: str, keywords: str = "", doc_
# its way to a small subgraph: seed entities by the question, hop out over
# relations to the 2nd-degree neighbours, then answer from that subgraph.
_KG_SEEDS = 3 # entities matched directly to the question
_KG_HOPS = 1 # relation hops out from the seeds (1 => "2nd degree")
_KG_NEIGHBORS = 10 # neighbour entity rows resolved per hop
# Scope of the compiled KG rows we search. Without a doc_scope we read the
# dataset-merged rows (one graph per dataset); with a doc_scope we read the
# per-document rows and filter by doc_id. Kept local rather than imported from
# the task-executor layer so the harness has no dependency on it.
_SCOPE_KWD_DATASET = "dataset"
_SCOPE_KWD_DOC = "doc"
_KG_SEEDS = 2 # top-N entities matched directly to the question
_KG_SEED_POOL = 64 # KNN candidate pool before the mention_count_int re-sort
_KG_SEED_SIM = 0.8 # dense-similarity floor for seed entities
_KG_HOPS = 2 # relation hops out from the seeds (1 => "2nd degree")
_KG_NEIGHBORS = 128 # cap on neighbour entity rows resolved per hop
_KG_REL_LIMIT = 32 # relations fetched per endpoint filter
async def _kg_scopes(tools, doc_scope: list[str] | None):
async def _kg_scopes(tools, doc_scope: list[str] | None = None):
"""Resolve the (kb_id, tenant_id, doc_ids|None) groups to search.
With a ``doc_scope`` the graph is limited to those docs (grouped by their
@@ -567,29 +576,59 @@ async def _kg_scopes(tools, doc_scope: list[str] | None):
return [(kb.id, kb.tenant_id, None) for kb in getattr(tools, "kbs", []) or []]
async def _kg_search(tools, kb_id: str, tenant_id: str, doc_ids, kind: str, text: str = "", top_n: int = 8, extra: dict | None = None) -> list[dict]:
"""Search the compiled KG rows of one KB and return the raw field maps."""
async def _kg_search(
tools,
kb_id: str,
tenant_id: str,
doc_ids,
kind: str,
text: str = "",
top_n: int = 8,
extra: dict | None = None,
scope_kwd: str | None = None,
order_desc: str | None = None,
pool: int | None = None,
similarity: float = 0.6,
) -> list[dict]:
"""Search the compiled KG rows of one KB and return the raw field maps.
``scope_kwd`` narrows to dataset-merged (``"dataset"``) or per-doc (``"doc"``)
rows. ``order_desc`` sorts the hits by that field descending; when combined
with a dense ``text`` match, ``pool`` sets the KNN candidate count so the
re-sort ranks a wider pool than the ``top_n`` finally kept. ``similarity`` is
the dense-match floor.
"""
from common import settings
from common.doc_store.doc_store_base import MatchTextExpr, OrderByExpr
from common.misc_utils import thread_pool_exec
from rag.nlp import search
condition: dict = {"knowledge_graph_kwd": [kind]}
if scope_kwd:
condition["scope_kwd"] = [scope_kwd]
if doc_ids:
condition["doc_id"] = list(doc_ids)
if extra:
condition.update(extra)
fields = ["content_with_weight", "source_chunk_ids", "doc_id", "docnm_kwd", "from_entity_kwd", "to_entity_kwd"]
fields = ["content_with_weight", "source_chunk_ids", "doc_id", "docnm_kwd", "name_kwd", "mention_count_int", "from_entity_kwd", "to_entity_kwd"]
exprs = []
if text:
knn_topn = pool or top_n
if getattr(tools, "embed_mdl", None):
try:
exprs.append(await settings.retriever.get_vector(text, tools.embed_mdl, top_n, 0.1))
exprs.append(await settings.retriever.get_vector(text, tools.embed_mdl, knn_topn, similarity))
except Exception:
_LOG.exception("[Graph exploration] vector build failed; using keyword match")
if not exprs:
exprs.append(MatchTextExpr(["content_ltks", "content_sm_ltks"], text, top_n))
exprs.append(MatchTextExpr(["content_ltks", "content_sm_ltks"], text, knn_topn))
order_by = OrderByExpr()
if order_desc:
try:
order_by.desc(order_desc)
except Exception:
order_by = OrderByExpr()
try:
res = await thread_pool_exec(
@@ -598,7 +637,7 @@ async def _kg_search(tools, kb_id: str, tenant_id: str, doc_ids, kind: str, text
[],
condition,
exprs,
OrderByExpr(),
order_by,
0,
top_n,
search.index_name(tenant_id),
@@ -651,6 +690,25 @@ def _kg_parse_relation(row: dict) -> dict | None:
}
def _endpoint_terms(names) -> list[str]:
"""Case variants for matching relation endpoints.
``dataset_structure_merger`` lowercases ``from_entity_kwd``/``to_entity_kwd``
on merged rows while entity names keep their original case, so hop queries
must try both forms. Accepts a single name or an iterable and returns the
sorted union of each name's original and lowercased form.
"""
if isinstance(names, str):
names = [names]
terms: set[str] = set()
for n in names or []:
n = (n or "").strip()
if n:
terms.add(n)
terms.add(n.lower())
return sorted(terms)
def _collect_evidence_ids(entities: list[dict], relations: list[dict], relevant_names: list[str]) -> dict:
"""Group the source_chunk_ids of the relevant entities AND relations by doc.
@@ -684,23 +742,31 @@ def _collect_evidence_ids(entities: list[dict], relations: list[dict], relevant_
async def graph_explore(tools, query: str, keywords: str = "", doc_scope: list[str] | None = None) -> dict:
"""Explore the compiled knowledge graph to answer ``query``.
Searches seed entities for the question, hops out over their relations to the
2nd-degree neighbours, asks the chat model to answer from that subgraph, then
pulls the source passages behind the entities/relations it found relevant and
narrows them with ``keywords``. Falls back to a plain hybrid search when the
KB has no compiled graph or no source text is behind the relevant nodes.
Seeds the top-``_KG_SEEDS`` entities for the question (dense match above
``_KG_SEED_SIM``, ranked by ``mention_count_int``), hops ``_KG_HOPS`` out over
their relations, then asks the chat model whether that subgraph answers the
question. When it does, the answer is returned directly; when it doesn't, the
source passages behind the entities/relations the model found relevant are
returned as evidence (narrowed by ``keywords``) so the caller can continue.
:returns: ``{"answer": str, "chunks": [...], "doc_aggs": [...]}``
Without ``doc_scope`` the dataset-merged graph is searched
(``scope_kwd="dataset"``); with it, the per-document rows of those docs
(``scope_kwd="doc"``, filtered by ``doc_id``).
:returns: ``{"answer": str, "chunks": [...], "doc_aggs": [...]}`` — exactly one
of ``answer`` / ``chunks`` is populated.
"""
from rag.advanced_rag.harness.tools.search import _narrow_by_keywords, hybrid_search
from rag.advanced_rag.harness.tools.search import _narrow_by_keywords
_empty = {"answer": "", "chunks": [], "doc_aggs": []}
_LOG.info(f'[Graph exploration] Exploring the knowledge graph for "{query}" (keywords: {keywords})')
scopes = await _kg_scopes(tools, doc_scope)
if not scopes:
_LOG.info("[Graph exploration] No knowledge base in scope — falling back to a normal search.")
return await hybrid_search(tools, query=query, keywords=keywords, doc_scope=doc_scope)
_LOG.info("[Graph exploration] No knowledge base in scope.")
return _empty
scope_kwd = _SCOPE_KWD_DOC if doc_scope else _SCOPE_KWD_DATASET
text = f"{query} {keywords}".strip()
entities: list[dict] = []
relations: list[dict] = []
@@ -718,52 +784,79 @@ async def graph_explore(tools, query: str, keywords: str = "", doc_scope: list[s
return added
for kb_id, tenant_id, doc_ids in scopes:
seed_rows = await _kg_search(tools, kb_id, tenant_id, doc_ids, "entity", text=text, top_n=_KG_SEEDS)
# (1) Seeds: condition C — dense match (>= _KG_SEED_SIM) over the scoped
# entity rows, ranked by mention_count_int desc, top _KG_SEEDS.
seed_rows = await _kg_search(
tools,
kb_id,
tenant_id,
doc_ids,
"entity",
text=text,
top_n=_KG_SEEDS,
scope_kwd=scope_kwd,
order_desc="mention_count_int",
pool=_KG_SEED_POOL,
similarity=_KG_SEED_SIM,
)
seeds = [e for e in (_kg_parse_entity(r) for r in seed_rows.values()) if e]
frontier = _add_entities(seeds, kb_id)
_LOG.info("[Graph exploration] Seeded %d entity(ies): %s", len(frontier), ", ".join(frontier) or "none")
# (2) Expand _KG_HOPS out, collecting relations and neighbour entities.
for _hop in range(_KG_HOPS):
if not frontier:
break
# Relations touching the current frontier (as source OR target).
terms = _endpoint_terms(frontier) # case variants — merged rows lowercase endpoints
rel_rows: dict = {}
rel_rows.update(await _kg_search(tools, kb_id, tenant_id, doc_ids, "relation", top_n=_KG_REL_LIMIT, extra={"from_entity_kwd": frontier}))
rel_rows.update(await _kg_search(tools, kb_id, tenant_id, doc_ids, "relation", top_n=_KG_REL_LIMIT, extra={"to_entity_kwd": frontier}))
rel_rows.update(await _kg_search(tools, kb_id, tenant_id, doc_ids, "relation", top_n=_KG_REL_LIMIT, scope_kwd=scope_kwd, extra={"from_entity_kwd": terms}))
rel_rows.update(await _kg_search(tools, kb_id, tenant_id, doc_ids, "relation", top_n=_KG_REL_LIMIT, scope_kwd=scope_kwd, extra={"to_entity_kwd": terms}))
hop_relations = [r for r in (_kg_parse_relation(x) for x in rel_rows.values()) if r]
relations.extend(hop_relations)
neighbour_names = {n for r in hop_relations for n in (r["from"], r["to"]) if n.lower() not in ent_names}
if not neighbour_names:
seen_lower = {k.split(":", 1)[1] for k in ent_names if k.startswith(f"{kb_id}:")}
neigh_names = {n.strip() for r in hop_relations for n in (r["from"], r["to"]) if n and n.strip()}
neigh_lower_set = {n.lower() for n in neigh_names} - seen_lower
if not neigh_lower_set:
break
neigh_rows = await _kg_search(tools, kb_id, tenant_id, doc_ids, "entity", text=" ".join(neighbour_names), top_n=_KG_NEIGHBORS)
wanted = {n.lower() for n in neighbour_names}
neighbours = [e for e in (_kg_parse_entity(r) for r in neigh_rows.values()) if e and e["name"].lower() in wanted]
neigh_filtered = {n for n in neigh_names if n.lower() in neigh_lower_set}
neigh_rows = await _kg_search(
tools,
kb_id,
tenant_id,
doc_ids,
"entity",
top_n=min(max(len(neigh_filtered), 1), _KG_NEIGHBORS),
scope_kwd=scope_kwd,
extra={"name_kwd": _endpoint_terms(neigh_filtered)},
)
neighbours = [e for e in (_kg_parse_entity(r) for r in neigh_rows.values()) if e]
frontier = _add_entities(neighbours, kb_id)
_LOG.info("[Graph exploration] Hop %d reached %d neighbour entity(ies).", _hop + 1, len(frontier))
if not entities and not relations:
_LOG.info("[Graph exploration] No compiled knowledge graph here — falling back to a normal search.")
return await hybrid_search(tools, query=query, keywords=keywords, doc_scope=doc_scope)
_LOG.info("[Graph exploration] No compiled knowledge graph in scope.")
return _empty
_LOG.info("[Graph exploration] Built a subgraph of %d entity(ies) and %d relation(s).", len(entities), len(relations))
# (3) Does the subgraph answer the question?
answer, relevant = await _ask_structure(tools, query, entities, relations, "knowledge graph", "Graph exploration")
# (4a) Sufficient — return the answer, no chunks.
if answer:
_LOG.info("[Graph exploration] The subgraph answered the question directly.")
return {"answer": answer, "chunks": [], "doc_aggs": []}
# (4b) Insufficient — return the source passages behind the relevant nodes.
evidence = _collect_evidence_ids(entities, relations, relevant)
chunks: list[dict] = []
for doc_id, ids in evidence.items():
if doc_id and ids:
chunks.extend(await _load_chunks_by_ids(tools, doc_id, ids))
if not chunks:
_LOG.info("[Graph exploration] No source text behind those nodes — falling back to a normal search.")
fallback = await hybrid_search(tools, query=query, keywords=keywords, doc_scope=doc_scope)
fallback["answer"] = answer
return fallback
before = len(chunks)
chunks = _narrow_by_keywords(chunks, keywords)
_LOG.info("[Graph exploration] Pulled %d source passage(s) behind those nodes, kept %d after keyword filtering.", before, len(chunks))
_LOG.info("[Graph exploration] Insufficient; returning %d evidence passage(s) (%d before keyword filtering).", len(chunks), before)
return {"answer": answer, "chunks": chunks, "doc_aggs": _doc_aggs(chunks)}
return {"answer": "", "chunks": chunks, "doc_aggs": _doc_aggs(chunks)}

View File

@@ -4,6 +4,7 @@ import logging
import re
import hashlib
from common import settings
from .navigation import _kg_scopes
_LOG = logging.getLogger(__name__)
@@ -179,7 +180,7 @@ def _normalize(kbinfos: dict, tenant_ids: list[str] | str | None) -> dict:
return kbinfos
async def hybrid_search(tools, query: str, kb_ids: list[str] | None = None, top_n: int = 12, doc_scope: list[str] | None = None, keywords: str = "") -> dict:
async def hybrid_search(tools, query: str, kb_ids: list[str] | None = None, top_n: int = 12, doc_scope: list[str] | None = None, keywords: str = "", use_compiled: bool = False) -> dict:
if not tools.kb_ids and not kb_ids:
return {"chunks": [], "doc_aggs": []}
target_ids = kb_ids or tools.kb_ids
@@ -220,6 +221,9 @@ async def hybrid_search(tools, query: str, kb_ids: list[str] | None = None, top_
length = len(kbinfos["chunks"])
kbinfos["chunks"] = _narrow_by_keywords(kbinfos.get("chunks", []), keywords)
_LOG.info(f"[Hybrid search] Kept {len(kbinfos['chunks'])} of {length} passage(s) that actually mention the keywords.")
if use_compiled and kbinfos.get("chunks"):
_LOG.info("[Hybrid search] Compiled expansion enabled — enriching with page_index/tree/KG navigation.")
await _expand_with_compiled(tools, query, keywords, kbinfos)
if cache is not None:
cache[cache_key] = kbinfos
return kbinfos
@@ -277,6 +281,473 @@ async def bm25_search(tools, query: str, kb_ids: list[str] | None = None, top_n:
return kbinfos
# ─── Compiled product expansion (zero-LLM, used by hybrid_search with use_compiled=True) ───
async def _expand_with_compiled(tools, query: str, keywords: str, kbinfos: dict) -> None:
"""Zero-LLM compiled-product expansion: page_index → tree → KG.
For each bound KB, searches compiled entity rows matching the query,
hops 1-hop via relations to find neighbour entities, then appends
their source passages to ``kbinfos["chunks"]``.
"""
before = len(kbinfos.get("chunks", []))
seen_ids = {c.get("chunk_id") or c.get("id") for c in kbinfos.get("chunks", [])}
scopes = await _kg_scopes(tools)
if not scopes:
return
for kb_id, tenant_id, doc_ids in scopes:
# 1-hop entity-graph expansion per template kind.
# Each template writes entity/relation rows tagged with
# ``compilation_template_kind_kwd`` — search them independently.
for label, template_kind in (
("knowledge_graph", "knowledge_graph"),
("mind_map", "mind_map"),
("timeline", "timeline"),
("page_index", "page_index"),
):
chunks = await _expand_compiled_strategy(
tools,
kb_id,
tenant_id,
doc_ids,
query,
seen_ids,
template_kind=template_kind,
max_chunks=5,
)
if chunks:
kbinfos.setdefault("chunks", []).extend(chunks)
_LOG.debug("[Compiled expand] %s: +%d chunks", label, len(chunks))
# Tree structure graph (uses ``compile_kwd``, not template kind).
chunks = await _expand_compiled_strategy(
tools,
kb_id,
tenant_id,
doc_ids,
query,
seen_ids,
compile_kwd="tree",
max_chunks=5,
)
if chunks:
kbinfos.setdefault("chunks", []).extend(chunks)
_LOG.debug("[Compiled expand] tree: +%d chunks", len(chunks))
# Synthesis pages — standalone rendered articles from wiki / session
# graph / session essence templates. Searched directly (no entity-graph nav).
for label, ckwd in (
("wiki_page", "wiki_page"),
("artifact_page", "artifact_page"),
("essence", "essence"),
):
chunks = await _expand_wiki_page_strategy(
tools,
kb_id,
tenant_id,
doc_ids,
query,
seen_ids,
compile_kwd=ckwd,
max_chunks=5,
)
if chunks:
kbinfos.setdefault("chunks", []).extend(chunks)
_LOG.debug("[Compiled expand] %s: +%d chunks", label, len(chunks))
# Re-sort so compiled-expansion chunks blend by similarity with regular ones.
chunks = kbinfos.get("chunks", [])
if chunks:
chunks.sort(key=lambda c: c.get("similarity", 0.0), reverse=True)
after = len(chunks)
_LOG.info("[Hybrid search] Compiled expansion added %d chunks.", after - before)
async def _search_compiled_rows(
tools,
kb_id: str,
tenant_id: str,
doc_ids: list[str] | None,
kind: str,
*,
text: str = "",
top_n: int = 8,
extra: dict | None = None,
compile_kwd: str | None = None,
template_kind: str | None = None,
) -> dict:
"""Search compiled KG rows in one KB, returning raw field maps.
*compile_kwd* filters by the ``compile_kwd`` field (e.g. "tree" for tree
structure nodes). *template_kind* filters by ``compilation_template_kind_kwd``
(e.g. "knowledge_graph", "mind_map"). Leave both ``None`` to scan all rows.
"""
from common import settings
from common.doc_store.doc_store_base import MatchTextExpr, OrderByExpr
from common.misc_utils import thread_pool_exec
from rag.nlp import search
condition: dict = {"knowledge_graph_kwd": [kind]}
if compile_kwd:
condition["compile_kwd"] = compile_kwd
if template_kind:
condition["compilation_template_kind_kwd"] = template_kind
if doc_ids:
condition["doc_id"] = list(doc_ids)
if extra:
condition.update(extra)
fields = [
"content_with_weight",
"source_chunk_ids",
"doc_id",
"docnm_kwd",
"from_entity_kwd",
"to_entity_kwd",
"name_kwd",
]
exprs = []
if text:
embd_mdl = getattr(tools, "embed_mdl", None)
if embd_mdl:
try:
exprs.append(await settings.retriever.get_vector(text, embd_mdl, top_n, 0.1))
except Exception:
_LOG.exception("[Compiled expand] vector build failed; using keyword match")
if not exprs:
exprs.append(
MatchTextExpr(
["content_ltks", "content_sm_ltks"],
text,
top_n,
)
)
try:
res = await thread_pool_exec(
settings.docStoreConn.search,
fields,
[],
condition,
exprs,
OrderByExpr(),
0,
top_n,
search.index_name(tenant_id),
[kb_id],
)
return settings.docStoreConn.get_fields(res, fields) or {}
except Exception:
_LOG.exception("[Compiled expand] search failed (kind=%s compile_kwd=%s)", kind, compile_kwd)
return {}
async def _load_chunks_for_doc(tools, doc_id: str, chunk_ids: list[str]) -> list[dict]:
"""Load chunks by their IDs from the doc store."""
if not chunk_ids:
return []
from common import settings
from common.doc_store.doc_store_base import OrderByExpr
from common.misc_utils import thread_pool_exec
from rag.nlp import search
resolved = await thread_pool_exec(tools._resolve_doc_tenant, doc_id)
if not resolved:
return []
kb_id, tenant_id = resolved
fields = ["content_with_weight", "docnm_kwd", "doc_id", "id"]
try:
res = await thread_pool_exec(
settings.docStoreConn.search,
fields,
[],
{"id": list(chunk_ids)},
[],
OrderByExpr(),
0,
len(chunk_ids),
search.index_name(tenant_id),
[kb_id],
)
rows = settings.docStoreConn.get_fields(res, fields)
if not rows:
return []
return [{**v, "chunk_id": k} for k, v in rows.items()]
except Exception:
_LOG.exception("[Compiled expand] failed to load chunks for doc_id=%s", doc_id)
return []
async def _expand_compiled_strategy(
tools,
kb_id: str,
tenant_id: str,
doc_ids: list[str] | None,
query: str,
seen_ids: set[str],
*,
compile_kwd: str | None = None,
template_kind: str | None = None,
max_chunks: int = 5,
) -> list[dict]:
"""Generic 1-hop compiled expansion: entity search → relation nav → chunk load.
1. Embedding-match seed entities (filtered by *compile_kwd* or *template_kind*).
2. Fetch relations adjacent to seed entities (forward + backward).
3. Collect neighbour entity names (1-hop away).
4. Look up neighbour entities to get ``source_chunk_ids``.
5. Load actual chunks, deduplicate, respect *max_chunks*.
*compile_kwd* is used for structure graphs (e.g. "tree").
*template_kind* is used for entity extraction rows (e.g. "knowledge_graph").
"""
import json
# -- 1. Seed entities --
seed_rows = await _search_compiled_rows(
tools,
kb_id,
tenant_id,
doc_ids,
"entity",
text=query,
top_n=5,
compile_kwd=compile_kwd,
template_kind=template_kind,
)
if not seed_rows:
return []
seed_names: set[str] = set()
for r in seed_rows.values():
try:
payload = json.loads(r.get("content_with_weight") or "{}")
except Exception:
continue
name = (payload.get("name") or payload.get("title") or "").strip()
if name:
seed_names.add(name)
if not seed_names:
return []
# -- 2. Adjacent relations (outgoing + incoming) --
# Provide both original and lowercased names — dataset_structure_merger
# lowercases merged-row endpoints while per-doc rows keep original case.
seed_list = sorted({n.lower() for n in seed_names} | seed_names)
fwd = await _search_compiled_rows(
tools,
kb_id,
tenant_id,
doc_ids,
"relation",
top_n=50,
compile_kwd=compile_kwd,
template_kind=template_kind,
extra={"from_entity_kwd": seed_list},
)
bwd = await _search_compiled_rows(
tools,
kb_id,
tenant_id,
doc_ids,
"relation",
top_n=50,
compile_kwd=compile_kwd,
template_kind=template_kind,
extra={"to_entity_kwd": seed_list},
)
all_rels = {**fwd, **bwd}
# -- 3. Neighbour names (1-hop, exclude seeds) --
seed_lower = {n.lower() for n in seed_names}
neighbour_names: set[str] = set()
for r in all_rels.values():
frm = (r.get("from_entity_kwd") or "").strip()
frm_lower = frm.lower()
to = (r.get("to_entity_kwd") or "").strip()
to_lower = to.lower()
if frm_lower in seed_lower and to and to_lower not in seed_lower:
neighbour_names.add(to)
if to_lower in seed_lower and frm and frm_lower not in seed_lower:
neighbour_names.add(frm)
if not neighbour_names:
return []
# -- 4. Neighbour entity source_chunk_ids --
# Provide both original and lowercased — same as seed_list above.
neigh_list = sorted({n.lower() for n in neighbour_names} | neighbour_names)
if len(neigh_list) > 100:
neigh_list = neigh_list[:100] # reasonable cap for name_kwd search
neigh_rows = await _search_compiled_rows(
tools,
kb_id,
tenant_id,
doc_ids,
"entity",
top_n=len(neigh_list),
compile_kwd=compile_kwd,
template_kind=template_kind,
extra={"name_kwd": neigh_list},
)
# Group chunk IDs by doc
by_doc: dict[str, set[str]] = {}
for r in neigh_rows.values():
doc_id = r.get("doc_id") or ""
for cid in r.get("source_chunk_ids") or []:
if cid and cid not in seen_ids:
by_doc.setdefault(doc_id, set()).add(cid)
# -- 5. Load and return --
new_chunks: list[dict] = []
for doc_id, cids in by_doc.items():
if len(new_chunks) >= max_chunks:
break
limit = max_chunks - len(new_chunks)
chunks = await _load_chunks_for_doc(tools, doc_id, list(cids)[:limit])
for c in chunks:
cid = c.get("chunk_id") or c.get("id")
if cid and cid not in seen_ids:
seen_ids.add(cid)
new_chunks.append(c)
return new_chunks
async def _search_synthesis_pages(
tools,
kb_id: str,
tenant_id: str,
doc_ids: list[str] | None,
text: str,
*,
compile_kwd: str = "wiki_page",
top_n: int = 8,
) -> dict:
"""Search synthesis-compiled page rows (no knowledge_graph_kwd filter).
Synthesis pages are standalone articles (wiki_page, artifact_page,
essence, etc.) with ``content_with_weight``, keyword index, and
vector. They do NOT carry the ``knowledge_graph_kwd`` field (unlike
entity/relation rows from extraction).
"""
from common import settings
from common.doc_store.doc_store_base import MatchTextExpr, OrderByExpr
from common.misc_utils import thread_pool_exec
from rag.nlp import search
condition: dict = {"compile_kwd": compile_kwd, "available_int": 1}
if doc_ids:
condition["source_doc_ids"] = list(doc_ids)
fields = [
"content_with_weight",
"summary_with_weight",
"source_chunk_ids",
"doc_id",
"title_kwd",
"topic_kwd",
]
exprs = []
if text:
embd_mdl = getattr(tools, "embed_mdl", None)
if embd_mdl:
try:
exprs.append(await settings.retriever.get_vector(text, embd_mdl, top_n, 0.1))
except Exception:
_LOG.exception("[Wiki expand] vector build failed; using keyword match")
if not exprs:
exprs.append(
MatchTextExpr(
["content_ltks", "content_sm_ltks"],
text,
top_n,
)
)
try:
res = await thread_pool_exec(
settings.docStoreConn.search,
fields,
[],
condition,
exprs,
OrderByExpr(),
0,
top_n,
search.index_name(tenant_id),
[kb_id],
)
return settings.docStoreConn.get_fields(res, fields) or {}
except Exception:
_LOG.exception("[Wiki expand] search failed for kb=%s", kb_id)
return {}
async def _expand_wiki_page_strategy(
tools,
kb_id: str,
tenant_id: str,
doc_ids: list[str] | None,
query: str,
seen_ids: set[str],
*,
compile_kwd: str = "wiki_page",
max_chunks: int = 5,
) -> list[dict]:
"""Expand synthesis-compiled pages: semantic search → load source chunks.
Unlike ``_expand_compiled_strategy`` (which does 1-hop entity-graph
navigation), synthesis pages are standalone rendered articles — we search
them directly and load the referenced source chunks as context.
*compile_kwd* selects which synthesis type: "wiki_page" (Wiki template),
"artifact_page" (Session Graph synthesis), "essence" (Session Essence).
"""
# -- 1. Search synthesis pages --
wiki_rows = await _search_synthesis_pages(
tools,
kb_id,
tenant_id,
doc_ids,
query,
compile_kwd=compile_kwd,
top_n=5,
)
if not wiki_rows:
return []
# -- 2. Collect source_chunk_ids from matching pages --
by_doc: dict[str, set[str]] = {}
for r in wiki_rows.values():
doc_id = r.get("doc_id") or ""
for cid in r.get("source_chunk_ids") or []:
if cid and cid not in seen_ids:
by_doc.setdefault(doc_id, set()).add(cid)
# -- 3. Load chunks, assign high similarity for priority ranking --
new_chunks: list[dict] = []
for doc_id, cids in by_doc.items():
if len(new_chunks) >= max_chunks:
break
limit = max_chunks - len(new_chunks)
chunks = await _load_chunks_for_doc(tools, doc_id, list(cids)[:limit])
for c in chunks:
cid = c.get("chunk_id") or c.get("id")
if cid and cid not in seen_ids:
seen_ids.add(cid)
c.setdefault("similarity", 0.9) # wiki pages rank high
new_chunks.append(c)
return new_chunks
async def web_search(tools, query: str, keywords: str = "") -> dict:
if not tools.has_web():
return {"chunks": [], "doc_aggs": []}

View File

@@ -135,7 +135,7 @@ def load_active_templates(template_ids, tenant_id: str) -> list[tuple[str, dict]
logging.warning("document_structure_compile: template %s config is invalid", template_id)
continue
kind = _compilation_template_kind(parser_cfg.get("kind"))
if not kind or kind == "artifacts":
if not kind or kind == "wiki":
continue
active_templates.append((template_id, parser_cfg))
return active_templates

View File

@@ -136,7 +136,7 @@ TASK_TYPE_TO_PIPELINE_TASK_TYPE = {
"graphrag": PipelineTaskType.GRAPH_RAG,
"mindmap": PipelineTaskType.MINDMAP,
"memory": PipelineTaskType.MEMORY,
"artifact": PipelineTaskType.ARTIFACT,
"wiki": PipelineTaskType.ARTIFACT,
"skill": PipelineTaskType.SKILL,
"structure_graph": PipelineTaskType.STRUCTURE_GRAPH,
"structure_mindmap": PipelineTaskType.STRUCTURE_MINDMAP,
@@ -152,7 +152,7 @@ _KB_FANOUT_TASK_TYPES = [
"graphrag",
"raptor",
"mindmap",
"artifact",
"wiki",
"skill",
"structure_graph",
"structure_mindmap",

View File

@@ -1028,7 +1028,7 @@ async def run_wiki(
template = CompilationTemplateService.get_saved(template_id, ctx.tenant_id)
config = template.get("config") if template else {}
kind = _compilation_template_kind(config.get("kind") if isinstance(config, dict) else "")
if kind == "artifacts":
if kind == "wiki":
eligible.append((d, template_id))
break
if not eligible:

View File

@@ -255,7 +255,7 @@ class TaskHandler:
await self._run_graphrag(embedding_model)
elif task_type == "mindmap":
ctx.progress_cb(1, "place holder")
elif task_type == "artifact":
elif task_type == "wiki":
from rag.svr.task_executor_refactor.dataset_wiki_generator import (
run_wiki,
)