zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-07-30 20:49:21 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refact: Rename "artifact_" to "wiki_". (#17553)

Browse Source
This commit is contained in:
Kevin Hu
2026-07-30 11:06:16 +08:00
committed by GitHub
parent be6263d522
commit 9bb037271e
10 changed files with 127 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
api/apps/services/dataset_api_service.py
View File

@@ -67,7 +67,7 @@ _INDEX_TYPE_TO_TASK_ID_FIELD = {
"graph": "graphrag_task_id",
"raptor": "raptor_task_id",
"mindmap": "mindmap_task_id",
"artifact": "artifact_task_id",
"artifact": "wiki_task_id",
"skill": "skill_task_id",
**{t: f"{t}_task_id" for t in _STRUCTURE_INDEX_TYPES},
}
@@ -1525,7 +1525,7 @@ async def search_datasets(tenant_id: str, req: dict):
# Artifact (knowledge compilation) page surface
#
# These three helpers power the dataset-level "Artifact" tab. They query rows
# with ``compile_kwd="artifact_page"`` written by TaskHandler's
# with ``compile_kwd="wiki_page"`` written by TaskHandler's
# ``_persist_wiki_pages_to_es``. The schema fields they rely on are:
# slug_kwd, title_kwd, page_type_kwd, content_with_weight,
# topic_kwd, entity_names_kwd, outlinks_kwd, related_kb_pages_kwd,
@@ -2867,11 +2867,11 @@ async def update_wiki_page(
``outlinks_kwd`` is rebuilt from the link-transform pass.
Per the v1 contract, only the page row is updated. The canvas
``artifact_page_graph`` / ``artifact_entity`` / ``artifact_relation``
``wiki_page_graph`` / ``wiki_entity`` / ``wiki_relation``
rows stay stale until the next full artifact compile.
Side effect: when the rendered post-save markdown differs from the
prior stored content, one ``artifact_commit`` row is recorded
prior stored content, one ``wiki_commit`` row is recorded
(git-style audit). No-op saves are silently skipped — empty diff,
no row.
@@ -3005,29 +3005,29 @@ async def update_wiki_page(
# All seven row types the artifact pipeline writes. Listed in dependency
# order so partial failures of earlier deletes don't leave behind state
# that downstream phases would silently reuse. ``artifact_page_graph``
# that downstream phases would silently reuse. ``wiki_page_graph``
# is the materialized canvas graph derived from the refined pages —
# the dataset Artifact tab's graph view reads exactly this row.
_WIKI_COMPILE_KWDS = (
"artifact_map_extract",
"artifact_reduce_result",
"artifact_compilation_plan",
"artifact_page_draft",
"artifact_page",
"artifact_page_topic",
"artifact_entity",
"artifact_relation",
"wiki_map_extract",
"wiki_reduce_result",
"wiki_compilation_plan",
"wiki_page_draft",
"wiki_page",
"wiki_page_topic",
"wiki_entity",
"wiki_relation",
)
# Tunables for the incremental graph loader. See ``get_wiki_graph``.
_WIKI_GRAPH_ENTITY_KWD = "artifact_entity"
_WIKI_GRAPH_RELATION_KWD = "artifact_relation"
_WIKI_GRAPH_ENTITY_KWD = "wiki_entity"
_WIKI_GRAPH_RELATION_KWD = "wiki_relation"
_WIKI_GRAPH_ENTITY_PAGE_SIZE = 32
_WIKI_GRAPH_MAX_LOADING_ENTITY = 512
def _wiki_entity_payload(row: dict) -> dict | None:
"""Project one ``artifact_entity`` ES row onto the canvas entity shape.
"""Project one ``wiki_entity`` ES row onto the canvas entity shape.
The row stores the canvas payload pre-built as JSON in
``content_with_weight``; we parse it back and overlay the columns
@@ -3085,11 +3085,11 @@ async def _wiki_search_entity_page(
limit: int,
keywords: str = "",
):
"""One page of artifact_entity rows.
"""One page of wiki_entity rows.
Without ``keywords``: ordered by ``weight_int DESC`` (heaviest nodes first).
With ``keywords``: BM25 full-text match over ``content_ltks`` (slug +
summary), ordered by relevance. ``artifact_entity`` rows are BM25-only (no
summary), ordered by relevance. ``wiki_entity`` rows are BM25-only (no
embedding vector), so this is a lexical search, not a dense KNN.
"""
from common.doc_store.doc_store_base import OrderByExpr
@@ -3214,23 +3214,23 @@ async def get_wiki_graph(
``top_n`` overrides the entity budget (default ``_WIKI_GRAPH_MAX_LOADING_ENTITY``).
``keywords`` (overview mode only; ignored when ``node`` is given) seeds the
graph from the best BM25 matches on ``artifact_entity`` rows instead of the
graph from the best BM25 matches on ``wiki_entity`` rows instead of the
heaviest-weighted ones. Only entities referenced by a relation are returned.
Two modes:
* **Overview** (``node`` is None) — paginate ``artifact_entity`` rows
* **Overview** (``node`` is None) — paginate ``wiki_entity`` rows
ordered by ``weight_int DESC`` in pages of
``_WIKI_GRAPH_ENTITY_PAGE_SIZE``. For each page, append entities
to a running set while the **cumulative** weight stays within
``_WIKI_GRAPH_MAX_LOADING_ENTITY``. Pull ``artifact_relation``
``_WIKI_GRAPH_MAX_LOADING_ENTITY``. Pull ``wiki_relation``
rows whose ``from_kwd`` is in the just-added entities; pull the
``to`` targets that we haven't seen yet (they count toward the same
cap). Stop once the cap is hit, or the page is empty, or no entry
from the page fit under the budget.
* **Click** (``node`` is a slug) — load the centre entity (always
included), pull every ``artifact_relation`` with ``from_kwd=node``,
included), pull every ``wiki_relation`` with ``from_kwd=node``,
then pull the ``to`` entities. Capped at
``_WIKI_GRAPH_MAX_LOADING_ENTITY`` for hub-node safety.

12
api/db/db_models.py
View File

@@ -862,8 +862,8 @@ class Knowledgebase(DataBaseModel):
raptor_task_finish_at = DateTimeField(null=True)
mindmap_task_id = CharField(max_length=32, null=True, help_text="Mindmap task ID", index=True)
mindmap_task_finish_at = DateTimeField(null=True)
artifact_task_id = CharField(max_length=32, null=True, help_text="Artifact compilation task ID", index=True)
artifact_task_finish_at = DateTimeField(null=True)
wiki_task_id = CharField(max_length=32, null=True, help_text="Artifact compilation task ID", index=True)
wiki_task_finish_at = DateTimeField(null=True)
skill_task_id = CharField(max_length=32, null=True, help_text="Skill generation task ID", index=True)
skill_task_finish_at = DateTimeField(null=True)
# KB-wide structure-graph merge tasks, one traceable task id per merged kind
@@ -995,7 +995,7 @@ class FileCommitItem(DataBaseModel):
# ``ArtifactCommit`` retired — artifact page history is now stored under
# ``FileCommit`` + ``FileCommitItem`` via ``FileCommitService.record_page_edit``
# (see the artifact-commit extension columns on those models above).
# Pre-existing ``artifact_commit`` rows are intentionally left in place;
# Pre-existing ``wiki_commit`` rows are intentionally left in place;
# no code path reads them.
@@ -1768,8 +1768,10 @@ def migrate_db():
alter_db_add_column(migrator, "knowledgebase", "raptor_task_finish_at", DateTimeField(null=True))
alter_db_add_column(migrator, "knowledgebase", "mindmap_task_id", CharField(max_length=32, null=True, help_text="Mindmap task ID", index=True))
alter_db_add_column(migrator, "knowledgebase", "mindmap_task_finish_at", DateTimeField(null=True))
alter_db_add_column(migrator, "knowledgebase", "artifact_task_id", CharField(max_length=32, null=True, help_text="Artifact compilation task ID", index=True))
alter_db_add_column(migrator, "knowledgebase", "artifact_task_finish_at", DateTimeField(null=True))
alter_db_rename_column(migrator, "knowledgebase", "artifact_task_id", "wiki_task_id")
alter_db_rename_column(migrator, "knowledgebase", "artifact_task_finish_at", "wiki_task_finish_at")
alter_db_add_column(migrator, "knowledgebase", "wiki_task_id", CharField(max_length=32, null=True, help_text="Artifact compilation task ID", index=True))
alter_db_add_column(migrator, "knowledgebase", "wiki_task_finish_at", DateTimeField(null=True))
alter_db_add_column(migrator, "knowledgebase", "skill_task_id", CharField(max_length=32, null=True, help_text="Skill generation task ID", index=True))
alter_db_add_column(migrator, "knowledgebase", "skill_task_finish_at", DateTimeField(null=True))
for _structure_type in ("structure_graph", "structure_mindmap", "timeline", "session_graph", "session_essence", "structure"):

6
api/db/services/document_service.py
View File

@@ -531,7 +531,7 @@ class DocumentService(CommonService):
# survives; one this doc solely owned is removed.
try:
if chunk_index_exists:
cls.remove_artifact_products(doc, tenant_id)
cls.remove_wiki_products(doc, tenant_id)
except Exception as e:
logging.warning(f"Failed to clean up artifact products for document {doc.id}: {e}")
@@ -582,7 +582,7 @@ class DocumentService(CommonService):
page += 1
@classmethod
def remove_artifact_products(cls, doc, tenant_id):
def remove_wiki_products(cls, doc, tenant_id):
"""Reference-counted cleanup of KB-scoped wiki/artifact products
in the doc store when a document is deleted.
@@ -591,7 +591,7 @@ class DocumentService(CommonService):
of the documents that contributed to it. On delete we detach
``doc.id`` from that list and drop the row only when this document
was its sole contributor — a product shared by other docs
survives. ``artifact_map_extract`` resume rows are 1:1 with a
survives. ``wiki_map_extract`` resume rows are 1:1 with a
document and are removed directly by ``doc_id``.
The compile_kwd set is pulled from the wiki generator so new

26
api/db/services/file_commit_service.py
View File

@@ -53,12 +53,12 @@ logger = logging.getLogger(__name__)
# Content storage for ``content_after`` is switched by a module-level
# constant so ops can move blobs between MinIO and the doc-store index
# without touching the schema.
ARTIFACT_CONTENT_STORAGE = "minio" # one of {"minio", "es"}
_ARTIFACT_COMMIT_BUCKET_PREFIX = ".artifact_commits"
_ARTIFACT_ES_KWD = "artifact_commit_content"
WIKI_CONTENT_STORAGE = "minio" # one of {"minio", "es"}
_WIKI_COMMIT_BUCKET_PREFIX = ".wiki_commits"
_WIKI_ES_KWD = "wiki_commit_content"
def _artifact_file_id(kb_id: str, slug: str) -> str:
def _wiki_file_id(kb_id: str, slug: str) -> str:
"""Deterministic 32-char id for the artifact-page 'file' identity.
Not a real File row — just an index key that groups all commits for
@@ -83,7 +83,7 @@ def _unified_diff(before: str, after: str, slug: str) -> str:
def _store_content_after(kb_id: str, content: str) -> tuple[str, str]:
"""Persist ``content`` per :data:`ARTIFACT_CONTENT_STORAGE`. Returns
"""Persist ``content`` per :data:`WIKI_CONTENT_STORAGE`. Returns
``(storage_kind, location)`` for the row's persistence columns.
Content-addressed by SHA-256 so re-saves with identical bodies share
@@ -92,8 +92,8 @@ def _store_content_after(kb_id: str, content: str) -> tuple[str, str]:
content_bytes = (content or "").encode("utf-8")
content_hash = hashlib.sha256(content_bytes).hexdigest()
if ARTIFACT_CONTENT_STORAGE == "minio":
location = f"{_ARTIFACT_COMMIT_BUCKET_PREFIX}/{content_hash}"
if WIKI_CONTENT_STORAGE == "minio":
location = f"{_WIKI_COMMIT_BUCKET_PREFIX}/{content_hash}"
try:
storage = settings.STORAGE_IMPL
if storage is not None:
@@ -106,7 +106,7 @@ def _store_content_after(kb_id: str, content: str) -> tuple[str, str]:
)
return "minio", location
if ARTIFACT_CONTENT_STORAGE == "es":
if WIKI_CONTENT_STORAGE == "es":
# Store as a single doc-store row so the same connector serves
# reads. The row is not retrievable (available_int=0).
from rag.nlp import search as _rag_search
@@ -116,7 +116,7 @@ def _store_content_after(kb_id: str, content: str) -> tuple[str, str]:
"id": content_hash,
"kb_id": kb_id,
"doc_id": kb_id,
"compile_kwd": _ARTIFACT_ES_KWD,
"compile_kwd": _WIKI_ES_KWD,
"content_with_weight": content or "",
"available_int": 0,
}
@@ -133,8 +133,8 @@ def _store_content_after(kb_id: str, content: str) -> tuple[str, str]:
# Unknown storage kind — fall through with empty location; the
# detail path treats missing location as "content not recoverable".
logging.warning(
"record_page_edit: unknown ARTIFACT_CONTENT_STORAGE=%r; content not persisted",
ARTIFACT_CONTENT_STORAGE,
"record_page_edit: unknown WIKI_CONTENT_STORAGE=%r; content not persisted",
WIKI_CONTENT_STORAGE,
)
return "", ""
@@ -732,7 +732,7 @@ class FileCommitService(CommonService):
final_title = f"{(title or '').strip() or f'{title_ts} {slug}'} "
commit_id = get_uuid()
item_id = get_uuid()
file_id = _artifact_file_id(kb_id, slug)
file_id = _wiki_file_id(kb_id, slug)
now_ts = current_timestamp()
now_dt = datetime_format(date_time=datetime.datetime.now())
@@ -819,7 +819,7 @@ class FileCommitService(CommonService):
"""
page = max(int(page or 1), 1)
page_size = max(min(int(page_size or 50), 200), 1)
file_id = _artifact_file_id(kb_id, slug)
file_id = _wiki_file_id(kb_id, slug)
base = (
FileCommit.select(

4
api/db/services/knowledgebase_service.py
View File

@@ -306,8 +306,8 @@ class KnowledgebaseService(CommonService):
cls.model.raptor_task_finish_at,
cls.model.mindmap_task_id,
cls.model.mindmap_task_finish_at,
cls.model.artifact_task_id,
cls.model.artifact_task_finish_at,
cls.model.wiki_task_id,
cls.model.wiki_task_finish_at,
cls.model.skill_task_id,
cls.model.skill_task_finish_at,
cls.model.structure_graph_task_id,

2
api/db/services/pipeline_operation_log_service.py
View File

@@ -40,7 +40,7 @@ _PIPELINE_TASK_TYPE_TO_FINISH_FIELD = {
PipelineTaskType.GRAPH_RAG: "graphrag_task_finish_at",
PipelineTaskType.RAPTOR: "raptor_task_finish_at",
PipelineTaskType.MINDMAP: "mindmap_task_finish_at",
PipelineTaskType.ARTIFACT: "artifact_task_finish_at",
PipelineTaskType.ARTIFACT: "wiki_task_finish_at",
PipelineTaskType.SKILL: "skill_task_finish_at",
PipelineTaskType.STRUCTURE_GRAPH: "structure_graph_task_finish_at",
PipelineTaskType.STRUCTURE_MINDMAP: "structure_mindmap_task_finish_at",

2
conf/infinity_mapping.json
View File

@@ -52,7 +52,7 @@
"compilation_template_kind_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
"chunk_hash_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
"input_hash_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
"artifact_slug_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
"wiki_slug_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
"md_with_weight": {"type": "varchar", "default": ""},
"summary_with_weight": {"type": "varchar", "default": ""},
"skill_with_weight": {"type": "varchar", "default": ""},

2
rag/advanced_rag/knowlege_compile/runner.py
View File

@@ -631,7 +631,7 @@ async def run_structure_compile_over_batches(
synthesis_cfg = (parser_cfg or {}).get("synthesis") or {}
if synthesis_cfg.get("enabled"):
example = synthesis_cfg.get("example")
compile_kwd = synthesis_cfg.get("compile_kwd", "artifact_page")
compile_kwd = synthesis_cfg.get("compile_kwd", "wiki_page")
plan_cfg = synthesis_cfg.get("plan") or {}
# Reserved for future wiki_plan_from_reduction extension:

107
rag/advanced_rag/knowlege_compile/wiki.py
View File

@@ -25,7 +25,7 @@
a byte position. The LLM is prompted to tag each extracted item with the
``[CHUNK_ID …]`` of the chunk it came from.
- Resume: per-chunk extracts are persisted to ES under
``compile_kwd="artifact_map_extract"`` with ``available_int=0`` and no vector
``compile_kwd="wiki_map_extract"`` with ``available_int=0`` and no vector
/ token-list fields, so retrievers ignore them but downstream phases can
fetch them by ``doc_id`` + ``source_chunk_ids``. Re-running MAP for the same
``doc_id`` skips chunks that already have an extract row.
@@ -56,8 +56,8 @@ from ._common import (
# Global pipeline-rev — bumping this constant invalidates every cached
# artifact_map_extract / artifact_reduce_result / artifact_compilation_plan
# / artifact_page_draft / artifact_page row on the next re-run. Use it
# wiki_map_extract / wiki_reduce_result / wiki_compilation_plan
# / wiki_page_draft / wiki_page row on the next re-run. Use it
# when a prompt or extraction schema changes in a way that should
# invalidate prior caches.
_WIKI_PIPELINE_REV = "v1"
@@ -85,7 +85,7 @@ from .structure import (
# Constants
# ---------------------------------------------------------------------------
WIKI_MAP_COMPILE_KWD = "artifact_map_extract"
WIKI_MAP_COMPILE_KWD = "wiki_map_extract"
DEFAULT_WIKI_MAP_WORKERS = 6
DEFAULT_WIKI_MAP_TIMEOUT = 600
@@ -675,7 +675,7 @@ async def _wiki_load_resume_map(
tenant_id: str,
kb_id: str,
) -> dict[str, str]:
"""Query ES for chunks that already have a artifact_map_extract row for
"""Query ES for chunks that already have a wiki_map_extract row for
this doc. Returns ``{chunk_id → chunk_hash}``.
``chunk_hash`` may be empty for legacy rows that predate the field —
@@ -732,7 +732,7 @@ async def _wiki_delete_map_rows(
tenant_id: str,
kb_id: str,
) -> int:
"""Delete ``artifact_map_extract`` rows for ``(doc_id, chunk_id)`` pairs.
"""Delete ``wiki_map_extract`` rows for ``(doc_id, chunk_id)`` pairs.
Used by the incremental MAP path:
* stale rows whose chunk content has changed → re-extracted next.
@@ -945,7 +945,7 @@ async def wiki_map_from_chunks(
Packs the provided RAGFlow chunks into batches via ``split_chunks``, runs
one ``gen_json`` extraction call per batch in parallel (bounded by
``max_workers``), then splits each batch's output back to per-chunk
extracts and persists them to ES as non-searchable ``artifact_map_extract``
extracts and persists them to ES as non-searchable ``wiki_map_extract``
rows so subsequent runs can skip chunks already processed.
Args:
@@ -1141,16 +1141,8 @@ async def wiki_map_from_chunks(
# REDUCE phase (KB-scoped)
# ---------------------------------------------------------------------------
#
# Migrated from D:/git/arkon/app/ai/mrp/reducer.py, steps 2.1-2.4.
# KB reconciliation (arkon 2.5-2.6) and the planning LLM call (arkon 2.7) are
# deferred to the PLAN phase — they belong with the planner, not the dedup.
#
# Scope difference from arkon: arkon REDUCE runs per source document. Here it
# runs per knowledge base — one set of canonical entities/concepts for the
# entire KB. Inputs come from ES (every artifact_map_extract row in this KB across
# all docs); the result lives in ES under artifact_reduce_result.
WIKI_REDUCE_COMPILE_KWD = "artifact_reduce_result"
WIKI_REDUCE_COMPILE_KWD = "wiki_reduce_result"
DEFAULT_WIKI_REDUCE_MERGE_THRESHOLD = 0.95
DEFAULT_WIKI_REDUCE_AMBIGUOUS_LOW = 0.75
DEFAULT_WIKI_REDUCE_AMBIGUOUS_BATCH = 50
@@ -1168,7 +1160,7 @@ WIKI_REDUCE_DISAMBIGUATE_SYSTEM = "You are a named-entity resolution assistant.
async def _wiki_load_all_map_extracts(tenant_id: str, kb_id: str) -> dict:
"""Aggregate every artifact_map_extract row in this KB into one merged dict.
"""Aggregate every wiki_map_extract row in this KB into one merged dict.
Pages through ES if the KB has more than the per-call cap. Returns a dict
in the same shape as wiki_map_from_chunks' return value.
@@ -1238,7 +1230,7 @@ async def _wiki_load_all_map_extracts(tenant_id: str, kb_id: str) -> dict:
async def _wiki_all_map_doc_ids(tenant_id: str, kb_id: str) -> list[str]:
"""Distinct ``doc_id`` across every ``artifact_map_extract`` row in this KB.
"""Distinct ``doc_id`` across every ``wiki_map_extract`` row in this KB.
These are the documents that fed the current compilation. Stamped onto
the KB-wide aggregate rows (REDUCE / PLAN) as ``source_doc_ids`` so a
@@ -1291,7 +1283,7 @@ async def _wiki_all_map_doc_ids(tenant_id: str, kb_id: str) -> list[str]:
async def _wiki_compute_map_input_hash(tenant_id: str, kb_id: str) -> str:
"""xxh64 fingerprint of the **current** ``artifact_map_extract`` rows for
"""xxh64 fingerprint of the **current** ``wiki_map_extract`` rows for
this KB — used by REDUCE / PLAN to cache-bust when MAP changed.
Built from ``sorted((chunk_id, chunk_hash))`` so:
@@ -1420,7 +1412,7 @@ async def _wiki_persist_reduce(
input_hash: str = "",
source_doc_ids: Optional[list[str]] = None,
) -> None:
"""Upsert the single non-searchable artifact_reduce_result row for this KB.
"""Upsert the single non-searchable wiki_reduce_result row for this KB.
``input_hash`` records the MAP-state fingerprint this reduction was
computed from; the next call compares it before re-running.
@@ -1478,7 +1470,7 @@ async def wiki_reduce_from_extracts(
) -> dict:
"""Phase 2 (REDUCE/Dedup) — KB-scoped.
Loads every ``artifact_map_extract`` row in this KB (across all documents) and
Loads every ``wiki_map_extract`` row in this KB (across all documents) and
produces a single canonical dict of entities/concepts via:
1. Exact dedup by ``(normalize(name), type)`` for entities and by
``normalize(term)`` for concepts.
@@ -1492,9 +1484,9 @@ async def wiki_reduce_from_extracts(
``chunk_ids`` per canonical entity.
The result is persisted to ES as a single non-searchable
``artifact_reduce_result`` row per KB. Subsequent calls with
``wiki_reduce_result`` row per KB. Subsequent calls with
``force_rerun=False`` (default) return the cached row immediately; pass
``force_rerun=True`` after new ``artifact_map_extract`` rows have been added.
``force_rerun=True`` after new ``wiki_map_extract`` rows have been added.
Args:
chat_mdl, embd_mdl: ragflow LLMBundle instances.
@@ -1503,7 +1495,7 @@ async def wiki_reduce_from_extracts(
ambiguous_low: cosine in [ambiguous_low, merge_threshold) goes to LLM.
ambiguous_batch_size: max pairs per LLM disambiguation call.
llm_timeout: seconds per LLM disambiguation batch.
force_rerun: bypass the cached artifact_reduce_result.
force_rerun: bypass the cached wiki_reduce_result.
callback: optional ``(progress: float, msg: str)`` callback.
Returns the canonical extract dict::
@@ -1642,27 +1634,19 @@ async def wiki_reduce_from_extracts(
# ---------------------------------------------------------------------------
# PLAN phase (KB-scoped)
# ---------------------------------------------------------------------------
#
# Migrated from D:/git/arkon/app/ai/mrp/reducer.py, steps 2.5-2.7 + 2.8 persist.
# Scope: per KB (one Compilation Plan covering the entire knowledge base),
# matching the REDUCE phase above.
#
# Flow:
# 1. Resume — return cached artifact_compilation_plan ES row when present.
# 2. Load REDUCE output from artifact_reduce_result.
# 1. Resume — return cached wiki_compilation_plan ES row when present.
# 2. Load REDUCE output from wiki_reduce_result.
# 3. KB reconciliation — batch-embed entity/concept query texts and run a
# per-item KNN against existing artifact_page rows in this KB. Classify
# per-item KNN against existing wiki_page rows in this KB. Classify
# UPDATE / MAYBE / CREATE by similarity. Batched LLM resolves MAYBE.
# 4. Planning call — one gen_json call producing the Compilation Plan JSON.
# 5. Attach raw items as side context for REFINE (no extra ES round-trips).
# 6. Persist as a single non-searchable artifact_compilation_plan row per KB.
# 6. Persist as a single non-searchable wiki_compilation_plan row per KB.
#
# Differences vs arkon: KB-scoped instead of per-source; no `source` pages
# emitted (chunk_ids attribution is enough); plan status defaults to
# "approved" so REFINE can consume immediately (review workflow deferred).
WIKI_PLAN_COMPILE_KWD = "artifact_compilation_plan"
WIKI_PAGE_COMPILE_KWD = "artifact_page"
WIKI_PLAN_COMPILE_KWD = "wiki_compilation_plan"
WIKI_PAGE_COMPILE_KWD = "wiki_page"
DEFAULT_WIKI_PLAN_UPDATE_THRESHOLD = 0.95
DEFAULT_WIKI_PLAN_MAYBE_THRESHOLD = 0.60
DEFAULT_WIKI_PLAN_TIMEOUT = 600 # ~10 min — the planning call emits one big
@@ -1821,7 +1805,7 @@ async def _wiki_reconcile_with_kb(
update_threshold: float,
maybe_threshold: float,
) -> dict[str, dict]:
"""Per-entity / per-concept KNN against compile_kwd=artifact_page rows in this KB.
"""Per-entity / per-concept KNN against compile_kwd=wiki_page rows in this KB.
Returns ``{name_or_term: {"action", "page_slug", "page_title", "page_id",
"similarity"}}``. When no artifact pages exist (first run before REFINE), every
@@ -2326,7 +2310,7 @@ async def _wiki_persist_plan(
input_hash: str = "",
source_doc_ids: Optional[list[str]] = None,
) -> None:
"""Upsert the single non-searchable artifact_compilation_plan row for this KB.
"""Upsert the single non-searchable wiki_compilation_plan row for this KB.
``input_hash`` records the REDUCE-state fingerprint this plan was
derived from; the next call compares it before re-planning.
@@ -2384,11 +2368,11 @@ async def wiki_plan_from_reduction(
) -> dict:
"""Phase 3 (PLAN) — KB-scoped.
Loads the cached ``artifact_reduce_result`` for this KB, reconciles every
canonical entity/concept against existing ``artifact_page`` rows in the same
Loads the cached ``wiki_reduce_result`` for this KB, reconciles every
canonical entity/concept against existing ``wiki_page`` rows in the same
KB (top-1 KNN, with MAYBE matches resolved by a batched LLM call), then
asks the LLM for one Compilation Plan JSON. The plan is persisted under
``compile_kwd="artifact_compilation_plan"`` with ``_status="approved"`` so
``compile_kwd="wiki_compilation_plan"`` with ``_status="approved"`` so
REFINE can consume it immediately.
Args:
@@ -2400,7 +2384,7 @@ async def wiki_plan_from_reduction(
maybe_threshold: cosine in [maybe_threshold, update_threshold) → ask LLM.
reconcile_batch_size: max pairs per LLM MAYBE-resolution call.
llm_timeout: seconds per LLM call (both MAYBE resolution and planning).
force_rerun: bypass the cached artifact_compilation_plan.
force_rerun: bypass the cached wiki_compilation_plan.
callback: optional ``(progress: float, msg: str)`` callback.
Returns the plan dict with this shape (plus underscore-prefixed side
@@ -2577,24 +2561,19 @@ async def wiki_plan_from_reduction(
# REFINE phase (KB-scoped)
# ---------------------------------------------------------------------------
#
# Migrated from D:/git/arkon/app/ai/mrp/writer.py (simple writer path) and
# merger.py (merge_page_content).
#
# Scope: per KB. Consumes the artifact_compilation_plan row written by PLAN,
# writes one artifact_page per planned page in parallel under a semaphore.
# Scope: per KB. Consumes the wiki_compilation_plan row written by PLAN,
# writes one wiki_page per planned page in parallel under a semaphore.
# UPDATE actions LLM-merge new vs existing content with a 70 % shrink-check
# fallback to the new content. Each written page is persisted to ES as a
# searchable artifact_page row (with embedding) so PLAN reconciliation finds it
# searchable wiki_page row (with embedding) so PLAN reconciliation finds it
# on the next REDUCE→PLAN cycle.
#
# Resume: per-slug artifact_page_draft rows act as a cache; a re-entry skips
# Resume: per-slug wiki_page_draft rows act as a cache; a re-entry skips
# slugs already cached unless force_rerun=True.
#
# Differences vs arkon: no full_text — source context is the union of the
# evidence chunks fetched from ES by id. Image-marker handling and the
# complex tool-using writer are deliberately deferred.
WIKI_DRAFT_COMPILE_KWD = "artifact_page_draft"
WIKI_DRAFT_COMPILE_KWD = "wiki_page_draft"
DEFAULT_WIKI_REFINE_WORKERS = 4
DEFAULT_WIKI_REFINE_TIMEOUT = 300
WIKI_REFINE_SOURCE_BUDGET_CHARS = 60_000
@@ -3018,7 +2997,7 @@ async def _wiki_build_source_context(
_WIKILINK_PIPE_RE = re.compile(r"\[\[([^\[\]\|]+?)\|([^\[\]]+?)\]\]")
_WIKILINK_SIMPLE_RE = re.compile(r"\[\[([^\[\]\|]+?)\]\]")
_ARTIFACT_MARKDOWN_LINK_RE = re.compile(r"\[([^\]]+)\]\(([^)]+)\)")
_WIKI_MARKDOWN_LINK_RE = re.compile(r"\[([^\]]+)\]\(([^)]+)\)")
def _wiki_transform_links(
@@ -3061,7 +3040,7 @@ def _wiki_transform_links(
def _is_valid(slug: str) -> bool:
return valid_slugs is None or slug in valid_slugs
def _artifact_slug(href: str) -> str | None:
def _wiki_slug(href: str) -> str | None:
parsed = urlsplit(href)
if parsed.scheme or parsed.netloc:
if parsed.netloc != "artifact":
@@ -3077,7 +3056,7 @@ def _wiki_transform_links(
return "/".join(parts[1:])
def _markdown_artifact(m: re.Match) -> str:
slug = _artifact_slug(m.group(2))
slug = _wiki_slug(m.group(2))
if not slug:
return m.group(0)
if not _is_valid(slug):
@@ -3100,7 +3079,7 @@ def _wiki_transform_links(
_track(slug)
return f"[{_display_text(slug, slug)}](artifact/{kb_id_str}/{slug})"
rewritten = _ARTIFACT_MARKDOWN_LINK_RE.sub(_markdown_artifact, content_md or "")
rewritten = _WIKI_MARKDOWN_LINK_RE.sub(_markdown_artifact, content_md or "")
rewritten = _WIKILINK_PIPE_RE.sub(_piped, rewritten)
rewritten = _WIKILINK_SIMPLE_RE.sub(_simple, rewritten)
return rewritten, outlinks
@@ -3410,7 +3389,7 @@ async def _wiki_persist_draft(
"id": _wiki_draft_row_id(kb_id, slug),
"doc_id": str(kb_id),
"compile_kwd": WIKI_DRAFT_COMPILE_KWD,
"artifact_slug_kwd": slug,
"wiki_slug_kwd": slug,
"source_id": [str(kb_id)],
"source_doc_ids": draft_doc_ids,
"input_hash_kwd": plan_input_hash,
@@ -3421,7 +3400,7 @@ async def _wiki_persist_draft(
try:
await thread_pool_exec(
settings.docStoreConn.delete,
{"compile_kwd": WIKI_DRAFT_COMPILE_KWD, "artifact_slug_kwd": slug},
{"compile_kwd": WIKI_DRAFT_COMPILE_KWD, "wiki_slug_kwd": slug},
index,
kb_id,
)
@@ -3449,7 +3428,7 @@ async def _wiki_load_refine_resume(
index = _rag_search.index_name(tenant_id)
condition = {"compile_kwd": [WIKI_DRAFT_COMPILE_KWD]}
select_fields = ["id", "artifact_slug_kwd", "content_with_weight", "input_hash_kwd"]
select_fields = ["id", "wiki_slug_kwd", "content_with_weight", "input_hash_kwd"]
PAGE_SIZE = 500
offset = 0
@@ -3475,7 +3454,7 @@ async def _wiki_load_refine_resume(
if not field_map:
break
for row in field_map.values():
slug = row.get("artifact_slug_kwd")
slug = row.get("wiki_slug_kwd")
content = row.get("content_with_weight")
if not isinstance(slug, str) or not isinstance(content, str):
continue
@@ -3754,7 +3733,7 @@ async def wiki_refine_from_plan(
logging.exception("wiki_refine: writer failed for slug=%s", slug)
return None
# Searchable artifact_page persistence has moved to the task
# Searchable wiki_page persistence has moved to the task
# handler so the doc-storage schema can be controlled in one
# place at the ingest layer.
# REFINE now just builds the page dict and resume cache.

68
rag/svr/task_executor_refactor/dataset_wiki_generator.py
View File

@@ -22,11 +22,11 @@ methods and one ``_run_wiki`` orchestrator. The public entry point is
:func:`run_wiki`.
The pipeline runs MAP per (doc, template) — each MAP call resumes from
its own ``artifact_map_extract`` ES rows — then REDUCE / PLAN / REFINE
its own ``wiki_map_extract`` ES rows — then REDUCE / PLAN / REFINE
KB-wide via ``rag.advanced_rag.knowlege_compile.wiki``. Refined pages
land in ES as searchable ``artifact_page`` rows, one
``artifact_page_topic`` row per topic, and ``artifact_entity`` /
``artifact_relation`` rows for the dataset Artifact tab's canvas graph.
land in ES as searchable ``wiki_page`` rows, one
``wiki_page_topic`` row per topic, and ``wiki_entity`` /
``wiki_relation`` rows for the dataset Artifact tab's canvas graph.
Design notes:
@@ -93,29 +93,29 @@ WIKI_REFINE_WORKERS = WIKI_MAP_LLM_POOL_SIZE
# Pages can accumulate hundreds of source chunks; the graph response is
# meant for fast canvas rendering, not full provenance audit, so we trim
# each node's list. The full per-page list is still available on the
# ``artifact_page`` row the UI deep-links into.
# ``wiki_page`` row the UI deep-links into.
WIKI_GRAPH_MAX_CHUNK_IDS_PER_NODE = 64
# Title + comments stamped on every ``artifact_commit`` row produced by
# Title + comments stamped on every ``wiki_commit`` row produced by
# :func:`run_wiki`'s regeneration path (as opposed to the dialog-edit
# path, which carries the user's own title/comments).
WIKI_REGEN_COMMIT_TITLE = "Regenerated by artifact compilation"
WIKI_REGEN_COMMIT_COMMENTS_TEMPLATE = "Auto-update via run_wiki (action={action})"
WIKI_MAP_COMPILE_KWD = "artifact_map_extract"
WIKI_REDUCE_COMPILE_KWD = "artifact_reduce_result"
WIKI_PLAN_COMPILE_KWD = "artifact_compilation_plan"
WIKI_DRAFT_COMPILE_KWD = "artifact_page_draft"
WIKI_PAGE_COMPILE_KWD = "artifact_page"
WIKI_PAGE_TOPIC_COMPILE_KWD = "artifact_page_topic"
WIKI_MAP_COMPILE_KWD = "wiki_map_extract"
WIKI_REDUCE_COMPILE_KWD = "wiki_reduce_result"
WIKI_PLAN_COMPILE_KWD = "wiki_compilation_plan"
WIKI_DRAFT_COMPILE_KWD = "wiki_page_draft"
WIKI_PAGE_COMPILE_KWD = "wiki_page"
WIKI_PAGE_TOPIC_COMPILE_KWD = "wiki_page_topic"
WIKI_DERIVED_COMPILE_KWDS = (
WIKI_REDUCE_COMPILE_KWD,
WIKI_PLAN_COMPILE_KWD,
WIKI_DRAFT_COMPILE_KWD,
WIKI_PAGE_COMPILE_KWD,
WIKI_PAGE_TOPIC_COMPILE_KWD,
"artifact_entity",
"artifact_relation",
"artifact_page_graph",
"wiki_entity",
"wiki_relation",
"wiki_page_graph",
)
@@ -292,7 +292,7 @@ async def _wiki_delete_deleted_doc_state(
return
# 2. Derived KB-scoped rows: reference-counted self-healing backstop for
# the eager delete-time cleanup (DocumentService.remove_artifact_products).
# the eager delete-time cleanup (DocumentService.remove_wiki_products).
# Read every row referencing any deleted doc, drop the ones left with no
# surviving owner, and shrink the rest to their surviving doc set. This
# replaces the former blunt "delete every derived row" wipe so products
@@ -499,7 +499,7 @@ async def persist_wiki_pages_to_es(
knowledge-compilation schema:
id xxh64(kb_id + ":" + slug)
compile_kwd "artifact_page"
compile_kwd "wiki_page"
slug_kwd page.slug
title_kwd page.title
page_type_kwd page.page_type
@@ -584,7 +584,7 @@ async def persist_wiki_pages_to_es(
n_emb = 0
if n_emb != len(pages):
logging.warning(
"artifact_persist: embedding count %d != pages %d for kb=%s; aborting",
"wiki_persist: embedding count %d != pages %d for kb=%s; aborting",
n_emb,
len(pages),
kb_id_str,
@@ -608,7 +608,7 @@ async def persist_wiki_pages_to_es(
vec_list = vec.tolist() if hasattr(vec, "tolist") else list(vec)
if not vec_list:
logging.warning(
"artifact_persist: empty embedding for slug=%s; skipping",
"wiki_persist: empty embedding for slug=%s; skipping",
slug,
)
continue
@@ -792,13 +792,13 @@ def build_wiki_page_graph(
entity_rows.append(
{
"id": xxhash.xxh64(
f"artifact_entity:{kb_id}:{slug}".encode("utf-8", "surrogatepass"),
f"wiki_entity:{kb_id}:{slug}".encode("utf-8", "surrogatepass"),
).hexdigest(),
"kb_id": kb_id,
"doc_id": kb_id, # KB-scoped sentinel
"available_int": 1,
"compile_kwd": "artifact_entity",
"type_kwd": "artifact_" + page_type,
"compile_kwd": "wiki_entity",
"type_kwd": "wiki_" + page_type,
"slug_kwd": slug,
"weight_int": int(weight),
"source_chunk_ids": capped_chunk_ids,
@@ -836,13 +836,13 @@ def build_wiki_page_graph(
relation_rows.append(
{
"id": xxhash.xxh64(
f"artifact_relation:{kb_id}:{src}:{tgt}".encode("utf-8", "surrogatepass"),
f"wiki_relation:{kb_id}:{src}:{tgt}".encode("utf-8", "surrogatepass"),
).hexdigest(),
"kb_id": kb_id,
"doc_id": kb_id,
"available_int": 1,
"compile_kwd": "artifact_relation",
"type_kwd": "artifact_relation",
"compile_kwd": "wiki_relation",
"type_kwd": "wiki_relation",
"from_kwd": src,
"to_kwd": tgt,
"source_doc_ids": edge_doc_ids,
@@ -863,12 +863,12 @@ async def persist_wiki_page_graph_to_es(
Writes two row types — both delete-then-insert for idempotent
re-runs:
1. ``compile_kwd="artifact_entity"`` — one row per page node,
1. ``compile_kwd="wiki_entity"`` — one row per page node,
BM25-only via ``content_ltks``.
2. ``compile_kwd="artifact_relation"`` — one row per surviving
2. ``compile_kwd="wiki_relation"`` — one row per surviving
edge (dangling outlinks dropped by the builder).
Also sweeps any leftover legacy ``artifact_page_graph`` blob so
Also sweeps any leftover legacy ``wiki_page_graph`` blob so
the index doesn't accumulate stale state.
"""
kb_id_str = str(ctx.kb_id)
@@ -910,19 +910,19 @@ async def persist_wiki_page_graph_to_es(
try:
await thread_pool_exec(
settings.docStoreConn.delete,
{"compile_kwd": "artifact_page_graph"},
{"compile_kwd": "wiki_page_graph"},
index,
ctx.kb_id,
)
except Exception:
logging.debug(
"artifact_page_graph: legacy blob sweep failed for kb=%s",
"wiki_page_graph: legacy blob sweep failed for kb=%s",
kb_id_str,
)
await asyncio.gather(
_replace_bucket("artifact_entity", entity_rows),
_replace_bucket("artifact_relation", relation_rows),
_replace_bucket("wiki_entity", entity_rows),
_replace_bucket("wiki_relation", relation_rows),
_sweep_legacy_blob(),
)
@@ -1085,7 +1085,7 @@ async def run_wiki(
return _cb
# 4. MAP per eligible doc. Each MAP call's own resume mechanism
# (artifact_map_extract rows keyed by chunk_id) skips chunks that
# (wiki_map_extract rows keyed by chunk_id) skips chunks that
# were already processed in a prior run — this is the incremental
# behavior the user asked for.
#
@@ -1287,7 +1287,7 @@ async def run_wiki(
progress(-1, "Wiki pipeline failed during REDUCE/PLAN/REFINE.")
return
# 6. Persist searchable artifact_page rows.
# 6. Persist searchable wiki_page rows.
try:
await persist_wiki_pages_to_es(ctx=ctx, pages=pages or [], embd_mdl=embedding_model)
except Exception: