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

fix: persist pipeline tree graph rows (#17400)

Browse Source
This commit is contained in:
buua436
2026-07-27 16:40:29 +08:00
committed by GitHub
parent 49e6181eca
commit 6e1e540f98
6 changed files with 268 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
api/db/init_data/compilation_templates/page_index.yaml
View File

@@ -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":"<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: ''

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

@@ -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

102
rag/advanced_rag/knowlege_compile/structure.py
View File

@@ -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": "<one of: ' + "|".join(type_values) + '>", "source": "<known entity name>", "target": "<known entity name>", "description": "<evidence or relation description>" }'
skeleton = (
'{ "type": "<one of: '
+ "|".join(type_values)
+ '>", "source": "<known entity name>", "target": "<known entity name>", "description": "<evidence or relation description>", "source_chunk_ids": ["<source chunk id>", ...] }'
)
else:
skeleton = '{ "type": "<one of: ' + "|".join(type_values) + '>", "name": "<exact extracted item text>", "description": "<evidence, definition, or detail from the source>" }'
skeleton = (
'{ "type": "<one of: '
+ "|".join(type_values)
+ '>", "name": "<exact extracted item text>", "description": "<evidence, definition, or detail from the source>", "source_chunk_ids": ["<source chunk id>", ...] }'
)
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,

22
rag/flow/compiler/compiler.py
View File

@@ -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,

35
web/src/components/ui/tree-view.tsx
View File

@@ -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<HTMLDivElement> & {
defaultLeafIcon?: any;
};
const TreeItemLabel = ({ item }: { item: TreeDataItem }) => {
if (!item.entityType) {
return <span className="flex-grow truncate text-sm">{item.name}</span>;
}
const isTitle = item.entityType === 'title';
return (
<span className="flex min-w-0 flex-grow items-center gap-2">
<span
className={cn(
'truncate text-sm',
isTitle && 'font-medium text-text-primary',
)}
>
{item.name}
</span>
<span
className={cn(
'shrink-0 rounded px-1.5 py-0.5 text-[10px] leading-none',
isTitle
? 'bg-accent/15 text-accent-foreground'
: 'bg-bg-card text-text-secondary',
)}
>
{item.entityType}
</span>
</span>
);
};
const TreeView = React.forwardRef<HTMLDivElement, TreeProps>(
(
{
@@ -211,7 +242,7 @@ const TreeNode = ({
isOpen={value.includes(item.id)}
default={defaultNodeIcon}
/>
<span className="text-sm truncate">{item.name}</span>
<TreeItemLabel item={item} />
<TreeActions isSelected={selectedItemId === item.id}>
{item.actions}
</TreeActions>
@@ -271,7 +302,7 @@ const TreeLeaf = React.forwardRef<
isSelected={selectedItemId === item.id}
default={defaultLeafIcon}
/>
<span className="flex-grow text-sm truncate">{item.name}</span>
<TreeItemLabel item={item} />
<TreeActions isSelected={selectedItemId === item.id}>
{item.actions}
</TreeActions>

10
web/src/pages/chunk/representation/utils/adapters.ts
View File

@@ -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(