From 6e1e540f987f89fd487dd9b90ec73a36c1535594 Mon Sep 17 00:00:00 2001 From: buua436 Date: Mon, 27 Jul 2026 16:40:29 +0800 Subject: [PATCH] fix: persist pipeline tree graph rows (#17400) --- .../compilation_templates/page_index.yaml | 88 +++++++++------ rag/advanced_rag/knowlege_compile/runner.py | 66 ++++++++++-- .../knowlege_compile/structure.py | 102 ++++++++++++++++-- rag/flow/compiler/compiler.py | 22 +++- web/src/components/ui/tree-view.tsx | 35 +++++- .../chunk/representation/utils/adapters.ts | 10 +- 6 files changed, 268 insertions(+), 55 deletions(-) diff --git a/api/db/init_data/compilation_templates/page_index.yaml b/api/db/init_data/compilation_templates/page_index.yaml index 17338f5da5..f17519daf1 100644 --- a/api/db/init_data/compilation_templates/page_index.yaml +++ b/api/db/init_data/compilation_templates/page_index.yaml @@ -4,50 +4,74 @@ config: kind: page_index entity: description: >- - You are a robust table-of-contents extractor with source-grounded - compressed descriptions. Extract each heading as an entity. The - entity's `name` must be the exact clean heading text. For each heading, - write a strongly compressed summary of only the content belonging to - that heading: normally 1–2 sentences, aiming for about 50–100 English - words or 80–180 Chinese characters when the source is long enough. Keep - only the 2–4 most important facts, prioritizing key - numbers, dates, names, locations, and conclusions. Do not copy long - passages verbatim and do not reduce the description to a generic label. - For detailed lists, give the overall result and only the most important - examples. Do not write meta descriptions such as `heading for ...`, - `section heading`, or `subheading for ...`. Do not include image - captions, `Main article` links, or unrelated tables and lists unless - they are the actual content of the heading. If a parent heading has no - direct body before the next child heading, generate a short higher-level - overview from the available child-section content; do not copy the child - description verbatim. This parent overview is needed for later - cross-chunk merging. Only use an empty description when there is no - usable source content at all. Do not invent content. + You are a source-grounded document index extractor. First extract every + supported heading in source order, then extract the important information + belonging to each heading. Headings form the hierarchy; facts, dates, + numbers, names, and conclusions are detail information that should be + captured in concise `fact` or `conclusion` entities attached below the + nearest relevant heading through `include` relations. Do not omit a + heading because its section is short, and do not return only headings. + + A title entity's `name` must be the exact clean heading text. A detail + entity's `name` must be a concise, source-grounded label or statement, + and its `description` must preserve the useful compressed information. + Keep detail entities atomic: do not combine unrelated facts, dates, + numbers, names, or conclusions into one item. Preserve important values, + units, dates, qualifiers, and conclusions. Do not invent content. + + For each title, write a strongly compressed summary of the content + belonging to that heading: normally 1–2 sentences, aiming for about + 50–100 English words or 80–180 Chinese characters when the source is + long enough. For detailed lists, give the overall result and only the + most important examples. Do not copy long passages verbatim and do not + reduce descriptions to generic labels. Do not write meta descriptions + such as `heading for ...`, `section heading`, or `subheading for ...`. + Do not include image captions, `Main article` links, or unrelated tables + and lists unless they are actual content of the heading. If a parent + heading has no direct body before the next child heading, generate a + short higher-level overview from the available child-section content; + do not copy the child description verbatim. Only use an empty + description when there is no usable source content at all. fields: - type: title description: the heading text (clean, no page numbers or leader dots) rule: | - - Length restriction: - • Chinese heading: ≤25 characters - • English heading: ≤80 characters - - "title" must be non-empty (or exactly "-1"). - - If any part of a chunk has no valid heading, output that part as {"title":"-1", ...}. - - If a chunk contains multiple headings, expand them in order: - - Each heading → {"title":"...","chunk_id":"","description":"a compact summary of only that heading's section"}. - - A parent heading with no direct body must still receive a short higher-level overview based on its child-section content for later merging; do not copy a child's summary verbatim. - - Prefer 1–2 concise sentences, aiming for about 50–100 English words or 80–180 Chinese characters when the source is long enough; preserve only the most important facts, dates, numbers, names, and conclusions. - - When ambiguous, prefer `-1` unless the text strongly looks like a heading. - - Keep language of "title" the same as the input. - - Prefix like following must be titles: 第x章, 第N条, 第N节, 1, 1.1, 1.1.1 ... + - Chinese titles ≤25 characters; English titles ≤80. Use clean heading text. + - The title must be non-empty; use `-1` only when no heading is supported. + - Extract every supported heading in source order; do not skip major or short sections. + - For multiple headings, preserve source order and summarize each section separately. + - Write 1–2 compressed sentences, retaining key facts, dates, numbers, names, locations, and conclusions. + - Summarize parent content from its children when it has no direct body; do not copy a child verbatim. + - Exclude captions, navigation links, and unrelated lists. Preserve numbering, language, and source meaning. + - type: fact + description: a concise source-grounded factual statement belonging to a title + rule: | + - Preserve important dates, numbers, names, units, qualifiers, and values. + - Do not merge unrelated facts into one entity. + - type: conclusion + description: a source-supported finding, outcome, or conclusion belonging to a title + rule: | + - State the conclusion concisely and do not add unsupported interpretation. relation: description: >- - You are an expert logical reasoning assistant specializing in hierarchical titles. + You are an expert logical reasoning assistant specializing in document + hierarchy and source-grounded detail attachment. Build relations only + from the entities in the current output. Preserve the title hierarchy + and attach every fact, date, number, named entity, and conclusion to the + nearest relevant title. fields: - type: include description: Upper-level title includes lower-level title. rule: | - "-1" is an invalid title; it does not belong to or include any other titles. - Must follow the hierarchical index/numbering (e.g., "1", "2.1", "3.2.5") when present. + - A title may include a child title or a detail entity. + - Every entity with type `fact` or `conclusion` must have exactly one nearest-title parent. + - Detail entities must not be attached directly to another detail entity. + - Do not create relations between unrelated detail entities. + - Every `from` and `to` value must exactly match an entity `name` in the current output. + - Never use a source heading as an endpoint unless that heading was also emitted as a `title` entity. + - Before returning, remove any relation whose endpoint is missing and verify that every detail entity has one parent relation. - Keep language of "title" the same as the input. - 第N章 must include 第N条 or 第N节. global_rules: '' diff --git a/rag/advanced_rag/knowlege_compile/runner.py b/rag/advanced_rag/knowlege_compile/runner.py index be6bec3e3e..d649d50b18 100644 --- a/rag/advanced_rag/knowlege_compile/runner.py +++ b/rag/advanced_rag/knowlege_compile/runner.py @@ -42,6 +42,7 @@ from api.db.services.compilation_template_group_service import ( ) from api.db.services.llm_service import LLMBundle from common.exceptions import TaskCanceledException +from common.token_utils import num_tokens_from_string from rag.advanced_rag.knowlege_compile.structure import ( LLMCallPool, MERGE_SCOPE_DATASET, @@ -56,9 +57,15 @@ from rag.advanced_rag.knowlege_compile.structure import ( # ----- tunables ------------------------------------------------------ # Bound how many source chunks are handed to a single -# ``compile_structure_from_text`` invocation. +# ``compile_structure_from_text`` invocation for regular templates. DOC_STRUCTURE_COMPILE_BATCH_CHUNKS = 4 +# Structure compilation packs chunks up to half of the model context. +# ``compile_structure_from_text`` applies the exact prompt-aware packing again +# before the LLM call. +STRUCTURE_CONTEXT_FRACTION = 0.5 +STRUCTURE_DEFAULT_CONTEXT = 100_000 + # Bound the number of batch/template extraction calls in flight. Results are # committed in submission order so accumulator updates and merge flushes stay # deterministic while the LLM calls run concurrently. @@ -416,6 +423,12 @@ async def run_structure_compile_over_batches( completed: dict[int, tuple[int, int, str, list[dict]]] = {} submit_sequence = 0 commit_sequence = 0 + dynamic_buffers: dict[str, list[dict]] = {template_id: [] for template_id, _ in active_templates} + dynamic_buffer_tokens: dict[str, int] = {template_id: 0 for template_id in dynamic_buffers} + + def _dynamic_batch_budget(template_id: str) -> int: + max_length = getattr(chat_mdl_by_tid[template_id], "max_length", None) or STRUCTURE_DEFAULT_CONTEXT + return max(int(max_length * STRUCTURE_CONTEXT_FRACTION), 1024) async def _commit_ready() -> None: nonlocal commit_sequence @@ -454,10 +467,21 @@ async def run_structure_compile_over_batches( async def _submit_batches() -> None: nonlocal submit_sequence batch_no = 0 + + async def _submit_one(batch: list[dict], template_id: str, parser_cfg: dict) -> None: + nonlocal submit_sequence, batch_no + if not batch: + return + batch_no += 1 + task = asyncio.create_task(_compile_batch(batch_no, batch, template_id, parser_cfg)) + inflight[task] = (submit_sequence, batch_no, len(batch), template_id) + submit_sequence += 1 + if len(inflight) + len(completed) >= DOC_STRUCTURE_COMPILE_MAX_IN_FLIGHT: + await _reap_one() + try: - async for batch in chunk_batches: - batch_no += 1 - for chunk in batch: + async for incoming_batch in chunk_batches: + for chunk in incoming_batch: cid = chunk.get("id") if isinstance(cid, str) and cid not in chunks_by_id: text = chunk.get("content_with_weight") or chunk.get("text") or "" @@ -465,11 +489,35 @@ async def run_structure_compile_over_batches( for template_id, parser_cfg in active_templates: if cancel_check(): raise TaskCanceledException("Task was cancelled during document knowledge compilation") - task = asyncio.create_task(_compile_batch(batch_no, batch, template_id, parser_cfg)) - inflight[task] = (submit_sequence, batch_no, len(batch), template_id) - submit_sequence += 1 - if len(inflight) + len(completed) >= DOC_STRUCTURE_COMPILE_MAX_IN_FLIGHT: - await _reap_one() + if template_kinds[template_id] == "knowledge_graph": + await _submit_one(incoming_batch, template_id, parser_cfg) + continue + buffer = dynamic_buffers[template_id] + budget = _dynamic_batch_budget(template_id) + buffer_tokens = dynamic_buffer_tokens[template_id] + for chunk in incoming_batch: + text = chunk.get("content_with_weight") or chunk.get("text") or "" + chunk_tokens = num_tokens_from_string(text if isinstance(text, str) else "") + if buffer and buffer_tokens + chunk_tokens > budget: + await _submit_one(buffer, template_id, parser_cfg) + buffer = [] + buffer_tokens = 0 + buffer.append(chunk) + buffer_tokens += chunk_tokens + if buffer_tokens >= budget: + await _submit_one(buffer, template_id, parser_cfg) + buffer = [] + buffer_tokens = 0 + dynamic_buffers[template_id] = buffer + dynamic_buffer_tokens[template_id] = buffer_tokens + + for template_id, buffer in dynamic_buffers.items(): + if cancel_check(): + raise TaskCanceledException("Task was cancelled during document knowledge compilation") + parser_cfg = dict(active_templates)[template_id] + await _submit_one(buffer, template_id, parser_cfg) + dynamic_buffers[template_id] = [] + dynamic_buffer_tokens[template_id] = 0 except BaseException: await _cancel_pending() raise diff --git a/rag/advanced_rag/knowlege_compile/structure.py b/rag/advanced_rag/knowlege_compile/structure.py index ef59ef43b4..d95a98b0a1 100644 --- a/rag/advanced_rag/knowlege_compile/structure.py +++ b/rag/advanced_rag/knowlege_compile/structure.py @@ -266,9 +266,17 @@ def _struct_render_type_fields(fields: list, language: str, *, kind: str) -> Tup lines.append("- type: other") if kind == "relation": - skeleton = '{ "type": "", "source": "", "target": "", "description": "" }' + skeleton = ( + '{ "type": "", "source": "", "target": "", "description": "", "source_chunk_ids": ["", ...] }' + ) else: - skeleton = '{ "type": "", "name": "", "description": "" }' + skeleton = ( + '{ "type": "", "name": "", "description": "", "source_chunk_ids": ["", ...] }' + ) return "\n".join(lines), skeleton @@ -331,7 +339,7 @@ def _struct_hypergraph_prompts(parser_config: dict, language: str = "en") -> Tup if rel_desc: edge_parts.append(f"## Relation Description:\n{rel_desc}") edge_parts.append(f"## Relation Fields:\n{rel_fields_text}") - edge_parts.append("## Known Entities:\nSee the source text below.") + edge_parts.append("## Known Entities:\n{known_nodes}") edge_parts.append( "## Response Format:\n" "Reply with a single JSON object of the form: " @@ -378,7 +386,13 @@ def _struct_unwrap_items(res) -> list: async def _struct_extract_hypergraph(text: str, parser_config: dict, chat_mdl, language: str) -> Tuple[list[dict], list[dict]]: node_prompt, edge_prompt_template = _struct_hypergraph_prompts(parser_config, language) - user_prompt = f"## Source Text:\n{text}\n\n## Output (JSON only):" + user_prompt = ( + "## Source Text:\n" + "Each source chunk is enclosed by [CHUNK_ID: ...] and [END_CHUNK]. " + "For every entity and relation, return source_chunk_ids containing only " + "the IDs of chunks that support that item.\n" + f"{text}\n\n## Output (JSON only):" + ) node_res = await gen_json(node_prompt, user_prompt, chat_mdl, gen_conf=_knowledge_compile_gen_conf(chat_mdl, {"temperature": 0.1})) nodes = _struct_unwrap_items(node_res) @@ -403,6 +417,22 @@ async def _struct_extract_hypergraph(text: str, parser_config: dict, chat_mdl, l return nodes, edges +def _struct_payload_chunk_ids(payload: dict, batch_ids: list) -> list: + """Keep only model-selected chunk IDs that belong to the current batch.""" + raw_ids = payload.get("source_chunk_ids") + if isinstance(raw_ids, str): + raw_ids = [raw_ids] + if not isinstance(raw_ids, list): + raw_ids = [] + allowed = set(batch_ids) + selected = [] + for chunk_id in raw_ids: + chunk_id = str(chunk_id).strip() + if chunk_id in allowed and chunk_id not in selected: + selected.append(chunk_id) + return selected or list(batch_ids) + + _struct_embed = _encode @@ -654,8 +684,8 @@ async def _struct_process_batch( return [] batch_ids: list = [e["chunk_id"] for e in packed if e.get("chunk_id")] - batch_segments: list[str] = [e["text"] for e in packed if isinstance(e.get("text"), str)] - combined_text = "\n\n---\n\n".join(batch_segments) + batch_segments: list[str] = [f"[CHUNK_ID: {e['chunk_id']}]\n{e['text']}\n[END_CHUNK]" for e in packed if e.get("chunk_id") and isinstance(e.get("text"), str)] + combined_text = "\n\n".join(batch_segments) src_field, target_field = _struct_relation_member_fields(parser_config) @@ -694,7 +724,7 @@ async def _struct_process_batch( payload, autotype, doc_id, - batch_ids, + _struct_payload_chunk_ids(payload, batch_ids), vec, kind, src_field=src_field, @@ -786,7 +816,6 @@ async def compile_structure_from_text( chunks, chat_mdl, prompt_overhead_tokens=prompt_overhead, - batch_size_cap=1 if str(template_kind).strip().lower().replace("-", "_") in {"page_index", "pageindex"} else None, ) if not packed_batches: return [] @@ -2089,6 +2118,63 @@ async def _struct_upsert_graph_json( await thread_pool_exec(settings.docStoreConn.insert, [row], index, kb_id) +async def _struct_upsert_tree_graph_rows( + graph: dict, + tenant_id: str, + kb_id: str, + doc_id: str, + embedding_model, + compilation_template_id: str | None = None, +) -> None: + """Persist Pipeline tree entities and child relations as structure rows. + + The tree graph blob remains the compact representation and discovery row; + these raw rows provide the entity/relation representation consumed by the + structure graph API and its subgraph builder. + """ + from common import settings + from rag.nlp import search as _rag_search + + entities = [item for item in graph.get("entities") or [] if isinstance(item, dict)] + relations = [item for item in graph.get("relations") or [] if isinstance(item, dict)] + index = _rag_search.index_name(tenant_id) + payloads = [(entity, "entity") for entity in entities] + [(relation, "relation") for relation in relations] + rows = [] + if payloads: + descriptions = [_struct_payload_description(payload) for payload, _ in payloads] + vectors = await _struct_embed(embedding_model, descriptions) + if len(vectors) != len(payloads): + raise ValueError(f"Tree graph embedding count mismatch: {len(vectors)} != {len(payloads)}") + + for (payload, kind), vector in zip(payloads, vectors): + source_chunk_ids = payload.get("source_chunk_ids") or [] if kind == "entity" else [] + rows.append( + _struct_to_doc_storage_doc( + payload=payload, + compile_kwd="tree", + doc_id=doc_id, + chunk_ids=source_chunk_ids, + vec=vector, + kind=kind, + src_field="from" if kind == "relation" else None, + target_field="to" if kind == "relation" else None, + compilation_template_id=compilation_template_id, + compilation_template_kind="tree", + ) + ) + + template_filter = {"compilation_template_ids": [compilation_template_id]} if compilation_template_id else {"must_not": {"exists": "compilation_template_ids"}} + delete_condition = { + "doc_id": [doc_id], + "compile_kwd": ["tree"], + "knowledge_graph_kwd": ["entity", "relation"], + **template_filter, + } + await thread_pool_exec(settings.docStoreConn.delete, delete_condition, index, kb_id) + if rows: + await thread_pool_exec(settings.docStoreConn.insert, rows, index, kb_id) + + async def rebuild_structure_graph_json( tenant_id: str, kb_id: str, diff --git a/rag/flow/compiler/compiler.py b/rag/flow/compiler/compiler.py index 4092348a70..b0dec7a1d4 100644 --- a/rag/flow/compiler/compiler.py +++ b/rag/flow/compiler/compiler.py @@ -84,8 +84,14 @@ class Compiler(ProcessBase, LLM): chunks. Supply RAPTOR with the same ``(text, vector, chunk_id)`` shape from the current pipeline output instead. """ - from rag.advanced_rag.knowlege_compile.structure import _struct_upsert_graph_json - from rag.svr.task_executor_refactor.chunk_post_processor import raptor_tree_to_graph + from rag.advanced_rag.knowlege_compile.structure import ( + _struct_upsert_graph_json, + _struct_upsert_tree_graph_rows, + ) + from rag.svr.task_executor_refactor.chunk_post_processor import ( + raptor_tree_to_graph, + rewrite_duplicate_tree_names, + ) from rag.svr.task_executor_refactor.raptor_service import RaptorService tree_inputs = [] @@ -146,9 +152,19 @@ class Compiler(ProcessBase, LLM): if bool(raptor_cfg.get("rechunk")): self._compile_progress(msg="Compiler: tree rechunking is not supported for in-memory pipeline chunks; keeping original chunks.") + await rewrite_duplicate_tree_names(tree, chat_mdl_by_tid[template_id]) + after_graph = raptor_tree_to_graph(tree) try: + await _struct_upsert_tree_graph_rows( + after_graph, + tenant_id, + kb_id, + doc_id, + embedding_model, + compilation_template_id=template_id, + ) await _struct_upsert_graph_json( - raptor_tree_to_graph(tree), + after_graph, tenant_id, kb_id, doc_id, diff --git a/web/src/components/ui/tree-view.tsx b/web/src/components/ui/tree-view.tsx index 02d6312e99..39500d2f4c 100644 --- a/web/src/components/ui/tree-view.tsx +++ b/web/src/components/ui/tree-view.tsx @@ -19,6 +19,7 @@ const selectedTreeVariants = cva( export interface TreeDataItem { id: string; name: string; + entityType?: string; icon?: any; selectedIcon?: any; openIcon?: any; @@ -37,6 +38,36 @@ type TreeProps = React.HTMLAttributes & { defaultLeafIcon?: any; }; +const TreeItemLabel = ({ item }: { item: TreeDataItem }) => { + if (!item.entityType) { + return {item.name}; + } + + const isTitle = item.entityType === 'title'; + return ( + + + {item.name} + + + {item.entityType} + + + ); +}; + const TreeView = React.forwardRef( ( { @@ -211,7 +242,7 @@ const TreeNode = ({ isOpen={value.includes(item.id)} default={defaultNodeIcon} /> - {item.name} + {item.actions} @@ -271,7 +302,7 @@ const TreeLeaf = React.forwardRef< isSelected={selectedItemId === item.id} default={defaultLeafIcon} /> - {item.name} + {item.actions} diff --git a/web/src/pages/chunk/representation/utils/adapters.ts b/web/src/pages/chunk/representation/utils/adapters.ts index 85c18ae5c0..a566030f9c 100644 --- a/web/src/pages/chunk/representation/utils/adapters.ts +++ b/web/src/pages/chunk/representation/utils/adapters.ts @@ -30,6 +30,7 @@ function buildTreeDataItems( entities: IStructureGraphEntity[], relations: IStructureGraphRelation[], relationTypes: string[], + showEntityType = false, ): TreeDataItem[] { const normalized = entities .map(normalizeEntity) @@ -40,6 +41,7 @@ function buildTreeDataItems( { id: entity.id, name: entity.name, + entityType: showEntityType ? entity.type : undefined, source_chunk_ids: entity.source_chunk_ids, }, ]), @@ -88,6 +90,7 @@ function buildUniqueTreeDataItems( { id: entity.id, name: entity.name, + entityType: entity.type, source_chunk_ids: entity.source_chunk_ids, }, ]), @@ -131,7 +134,12 @@ function buildUniqueTreeDataItems( export function adaptPageIndexToTreeData( template: IStructureGraphTemplate, ): TreeDataItem[] { - return buildTreeDataItems(template.entities, template.relations, ['include']); + return buildTreeDataItems( + template.entities, + template.relations, + ['include'], + true, + ); } export function adaptTreeToTreeData(