153 lines
5.3 KiB
Python
153 lines
5.3 KiB
Python
"""Normalize and compile DeerFlow-owned Wiki generation guidance.
|
|
|
|
The values accepted by the public API are raw user configuration. They are
|
|
kept separately from the effective strings sent to WeKnora so editing a focus
|
|
or directory constraint never appends a second copy of an older compiled
|
|
block.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
WEKNORA_CUSTOM_INSTRUCTION_MAX = 4000
|
|
WIKI_FOCUS_INSTRUCTION_MAX = 2000
|
|
WIKI_CONFIG_SCHEMA_VERSION = 3
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CompiledWikiInstructions:
|
|
content_instructions: str
|
|
extraction_instructions: str
|
|
directory_prompt_coverage: str
|
|
directory_prompt_covered_nodes: int
|
|
directory_prompt_total_nodes: int
|
|
|
|
|
|
def normalize_focus_instructions(value: str | None) -> str:
|
|
"""Return the canonical raw focus text used for persistence and hashing."""
|
|
|
|
if value is None:
|
|
return ""
|
|
normalized = str(value).replace("\r\n", "\n").replace("\r", "\n").strip()
|
|
if len(normalized) > WIKI_FOCUS_INSTRUCTION_MAX:
|
|
raise ValueError(
|
|
f"重点要求不能超过 {WIKI_FOCUS_INSTRUCTION_MAX} 个字符"
|
|
)
|
|
return normalized
|
|
|
|
|
|
def _join_blocks(*blocks: str) -> str:
|
|
return "\n\n".join(block.strip() for block in blocks if block.strip())
|
|
|
|
|
|
def _focus_content_block(focus: str) -> str:
|
|
if not focus:
|
|
return ""
|
|
return _join_blocks(
|
|
f"【用户重点要求】\n{focus}",
|
|
(
|
|
"这是生成范围的硬约束：只生成与上述重点直接相关且有来源支撑的页面；"
|
|
"不要为背景材料中与重点无关的人物、机构、地点或主题单独生成页面。"
|
|
"不得为了满足重点要求而编造信息。"
|
|
),
|
|
)
|
|
|
|
|
|
def _focus_extraction_block(focus: str) -> str:
|
|
if not focus:
|
|
return ""
|
|
return _join_blocks(
|
|
f"【用户重点要求】\n{focus}",
|
|
(
|
|
"这是抽取范围的硬约束：只抽取与上述重点直接相关、可追溯到来源的"
|
|
"实体、事件、时间线、指标和关系；与重点无关的内容必须跳过。"
|
|
),
|
|
)
|
|
|
|
|
|
def _directory_lines(directory_paths: list[list[str]]) -> list[str]:
|
|
return [f"{' ' * (len(path) - 1)}- {path[-1]}" for path in directory_paths]
|
|
|
|
|
|
def compile_wiki_instructions(
|
|
*,
|
|
content_instructions: str | None,
|
|
extraction_instructions: str | None,
|
|
focus_instructions: str | None,
|
|
directory_paths: list[list[str]],
|
|
max_length: int = WEKNORA_CUSTOM_INSTRUCTION_MAX,
|
|
) -> CompiledWikiInstructions:
|
|
"""Compile raw inputs into the two instruction strings supported upstream.
|
|
|
|
Raw free text and focus guidance are never truncated. Directory nodes are
|
|
appended in deterministic parent-before-child order until the remaining
|
|
content budget is exhausted. They are examples for the generator rather
|
|
than a post-generation allowlist: WeKnora may adapt the hierarchy when the
|
|
source material needs a clearer organization.
|
|
"""
|
|
|
|
raw_content = str(content_instructions or "").strip()
|
|
raw_extraction = str(extraction_instructions or "").strip()
|
|
focus = normalize_focus_instructions(focus_instructions)
|
|
|
|
content_base = _join_blocks(raw_content, _focus_content_block(focus))
|
|
extraction = _join_blocks(raw_extraction, _focus_extraction_block(focus))
|
|
if len(content_base) > max_length:
|
|
raise ValueError(
|
|
f"内容生成要求与重点要求合并后不能超过 {max_length} 个字符"
|
|
)
|
|
if len(extraction) > max_length:
|
|
raise ValueError(
|
|
f"信息抽取要求与重点要求合并后不能超过 {max_length} 个字符"
|
|
)
|
|
|
|
total_nodes = len(directory_paths)
|
|
covered_nodes = 0
|
|
content = content_base
|
|
if total_nodes:
|
|
header = "【参考目录层级】"
|
|
footer = (
|
|
"上述目录仅用于帮助理解期望的组织方式，不是必须逐项套用的目录模板，也不是目录"
|
|
"白名单。请结合原始资料自行合并、拆分、改名或补充目录；内容结构与参考目录冲突时，"
|
|
"以资料实际内容和清晰表达为准。目录只用于组织，不得作为事实来源。"
|
|
)
|
|
accepted: list[str] = []
|
|
for line in _directory_lines(directory_paths):
|
|
candidate_block = "\n".join([header, *accepted, line, "", footer])
|
|
candidate = _join_blocks(content_base, candidate_block)
|
|
if len(candidate) > max_length:
|
|
break
|
|
accepted.append(line)
|
|
if accepted:
|
|
content = _join_blocks(
|
|
content_base,
|
|
"\n".join([header, *accepted, "", footer]),
|
|
)
|
|
covered_nodes = len(accepted)
|
|
|
|
coverage = (
|
|
"none"
|
|
if total_nodes == 0 or covered_nodes == 0
|
|
else "full"
|
|
if covered_nodes == total_nodes
|
|
else "partial"
|
|
)
|
|
return CompiledWikiInstructions(
|
|
content_instructions=content,
|
|
extraction_instructions=extraction,
|
|
directory_prompt_coverage=coverage,
|
|
directory_prompt_covered_nodes=covered_nodes,
|
|
directory_prompt_total_nodes=total_nodes,
|
|
)
|
|
|
|
|
|
__all__ = [
|
|
"CompiledWikiInstructions",
|
|
"WEKNORA_CUSTOM_INSTRUCTION_MAX",
|
|
"WIKI_CONFIG_SCHEMA_VERSION",
|
|
"WIKI_FOCUS_INSTRUCTION_MAX",
|
|
"compile_wiki_instructions",
|
|
"normalize_focus_instructions",
|
|
]
|