diff --git a/api/apps/services/dataset_api_service.py b/api/apps/services/dataset_api_service.py index b92a6180ab..52965bbaf2 100644 --- a/api/apps/services/dataset_api_service.py +++ b/api/apps/services/dataset_api_service.py @@ -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. diff --git a/api/db/db_models.py b/api/db/db_models.py index 68ed6e42f1..ed977c314c 100644 --- a/api/db/db_models.py +++ b/api/db/db_models.py @@ -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"): diff --git a/api/db/services/document_service.py b/api/db/services/document_service.py index 660e90c9c0..48167aff6c 100644 --- a/api/db/services/document_service.py +++ b/api/db/services/document_service.py @@ -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 diff --git a/api/db/services/file_commit_service.py b/api/db/services/file_commit_service.py index 67cbbb4b59..86117596f0 100644 --- a/api/db/services/file_commit_service.py +++ b/api/db/services/file_commit_service.py @@ -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( diff --git a/api/db/services/knowledgebase_service.py b/api/db/services/knowledgebase_service.py index 491786baad..0cf2500360 100644 --- a/api/db/services/knowledgebase_service.py +++ b/api/db/services/knowledgebase_service.py @@ -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, diff --git a/api/db/services/pipeline_operation_log_service.py b/api/db/services/pipeline_operation_log_service.py index b4e102797a..4b700827b9 100644 --- a/api/db/services/pipeline_operation_log_service.py +++ b/api/db/services/pipeline_operation_log_service.py @@ -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", diff --git a/conf/infinity_mapping.json b/conf/infinity_mapping.json index c8a0b1bfa3..de40f58df7 100644 --- a/conf/infinity_mapping.json +++ b/conf/infinity_mapping.json @@ -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": ""}, diff --git a/rag/advanced_rag/knowlege_compile/runner.py b/rag/advanced_rag/knowlege_compile/runner.py index d649d50b18..d87dca810d 100644 --- a/rag/advanced_rag/knowlege_compile/runner.py +++ b/rag/advanced_rag/knowlege_compile/runner.py @@ -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: diff --git a/rag/advanced_rag/knowlege_compile/wiki.py b/rag/advanced_rag/knowlege_compile/wiki.py index f0f44ec87b..07684db355 100644 --- a/rag/advanced_rag/knowlege_compile/wiki.py +++ b/rag/advanced_rag/knowlege_compile/wiki.py @@ -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. diff --git a/rag/svr/task_executor_refactor/dataset_wiki_generator.py b/rag/svr/task_executor_refactor/dataset_wiki_generator.py index cce39ae7af..8b6e8c3ed3 100644 --- a/rag/svr/task_executor_refactor/dataset_wiki_generator.py +++ b/rag/svr/task_executor_refactor/dataset_wiki_generator.py @@ -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: