Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 77 additions & 6 deletions dataflow/operators/knowledge_cleaning/generate/kbc_text_cleaner.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,76 @@
from typing import Union

import re


_CLEANED_BLOCK_RE = re.compile(
r"<cleaned_start>\s*(.*?)\s*<cleaned_end>",
flags=re.IGNORECASE | re.DOTALL,
)
_ANSWER_BLOCK_RE = re.compile(
r"<answer>\s*(.*?)\s*</answer>",
flags=re.IGNORECASE | re.DOTALL,
)
_PROCESSING_STEPS_RE = re.compile(
r"\s*Processing Steps:\s*"
r"1\.\s*\[Tag Analysis\]\s*Classify markup tags\s*"
r"2\.\s*\[Reference Extraction\]\s*Isolate images/tables\s*"
r"3\.\s*\[Character Audit\]\s*Log special chars\s*"
r"4\.\s*\[Structure Check\]\s*Validate hierarchy\s*"
r"5\.\s*\[Final Output\]\s*Generate cleaned text.*$",
flags=re.IGNORECASE | re.DOTALL,
)
_PROCESSING_STEPS_WITHOUT_HEADER_RE = re.compile(
r"\s*1\.\s*\[Tag Analysis\]\s*Classify markup tags\s*"
r"2\.\s*\[Reference Extraction\]\s*Isolate images/tables\s*"
r"3\.\s*\[Character Audit\]\s*Log special chars\s*"
r"4\.\s*\[Structure Check\]\s*Validate hierarchy\s*"
r"5\.\s*\[Final Output\]\s*Generate cleaned text.*$",
flags=re.IGNORECASE | re.DOTALL,
)
_ZH_PROCESSING_STEPS_RE = re.compile(
r"\s*(?:处理步骤|处理流程|治理步骤)[::]\s*"
r"1[.、]\s*\[(?:标签分析|标记分析)\].*?"
r"2[.、]\s*\[(?:引用提取|参考提取)\].*?"
r"3[.、]\s*\[(?:字符审核|字符审计)\].*?"
r"4[.、]\s*\[(?:结构检查|结构校验)\].*?"
r"5[.、]\s*\[(?:最终输出|最终结果)\].*$",
flags=re.DOTALL,
)
_ZH_PROCESSING_STEPS_WITHOUT_HEADER_RE = re.compile(
r"\s*1[.、]\s*\[(?:标签分析|标记分析)\].*?"
r"2[.、]\s*\[(?:引用提取|参考提取)\].*?"
r"3[.、]\s*\[(?:字符审核|字符审计)\].*?"
r"4[.、]\s*\[(?:结构检查|结构校验)\].*?"
r"5[.、]\s*\[(?:最终输出|最终结果)\].*$",
flags=re.DOTALL,
)


def extract_cleaned_text(text, post_process=None) -> str:
"""Extract model output while dropping prompt-instruction leakage."""
text = "" if text is None else str(text)

cleaned_match = _CLEANED_BLOCK_RE.search(text)
if cleaned_match:
text = cleaned_match.group(1)
else:
answer_match = _ANSWER_BLOCK_RE.search(text)
if answer_match:
text = answer_match.group(1)
text = text.replace("<cleaned_start>", "").replace("<cleaned_end>", "")

text = _PROCESSING_STEPS_RE.sub("", text)
text = _PROCESSING_STEPS_WITHOUT_HEADER_RE.sub("", text)
text = _ZH_PROCESSING_STEPS_RE.sub("", text)
text = _ZH_PROCESSING_STEPS_WITHOUT_HEADER_RE.sub("", text)
text = text.strip()

if post_process:
text = post_process(text)
return text


@prompt_restrict(
KnowledgeCleanerPrompt
)
Expand All @@ -27,7 +97,7 @@ def __init__(self, llm_serving: LLMServingABC, lang="en", prompt_template : Unio
if prompt_template:
self.prompt_template = prompt_template
else:
self.prompt_template = KnowledgeCleanerPrompt()
self.prompt_template = KnowledgeCleanerPrompt(lang=lang)

@staticmethod
def get_desc(lang: str = "zh"):
Expand Down Expand Up @@ -123,15 +193,16 @@ def run(
formatted_prompts = self._reformat_prompt(dataframe)
cleaned = self.llm_serving.generate_from_input(formatted_prompts,"")

#for each in cleaned, only save the content in <cleaned_start> and <cleaned_end>
# Save only the final cleaned text, even if the model leaks prompt steps.
cleaned_extracted = [
str(text).split('<cleaned_start>')[1].split('<cleaned_end>')[0].strip()
if '<cleaned_start>' in str(text) and '<cleaned_end>' in str(text)
else str(text).strip()
extract_cleaned_text(
text,
getattr(self.prompt_template, "_post_process", None),
)
for text in cleaned
]
dataframe[self.output_key] = cleaned_extracted
output_file = storage.write(dataframe)
self.logger.info(f"Results saved to {output_file}")

return [output_key]
return [output_key]
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from dataflow.prompts.kbcleaning import KnowledgeCleanerPrompt
from dataflow.operators.knowledge_cleaning.generate.kbc_text_cleaner import extract_cleaned_text
import pandas as pd
from dataflow.utils.registry import OPERATOR_REGISTRY
from dataflow import get_logger
Expand Down Expand Up @@ -125,11 +126,12 @@ def run(
raw_chunks, formatted_prompts = self._reformat_prompt_from_path(chunk_path)
cleaned = self.llm_serving.generate_from_input(formatted_prompts, "")

# for each in cleaned, only save the content in <cleaned_start> and <cleaned_end>
# Save only the final cleaned text, even if the model leaks prompt steps.
cleaned_extracted = [
text.split('<cleaned_start>')[1].split('<cleaned_end>')[0].strip()
if '<cleaned_start>' in str(text) and '<cleaned_end>' in str(text)
else str(text).strip()
extract_cleaned_text(
text,
getattr(self.prompts, "_post_process", None),
)
for text in cleaned
]
json_items=[{
Expand Down
24 changes: 3 additions & 21 deletions dataflow/prompts/kbcleaning.py
Original file line number Diff line number Diff line change
Expand Up @@ -177,35 +177,17 @@ def build_prompt(self, raw_content: str) -> str:
"""

if self.lang == "en":
processing_steps = """
Processing Steps:
1. [Tag Analysis] Classify markup tags
2. [Reference Extraction] Isolate images/tables
3. [Character Audit] Log special chars
4. [Structure Check] Validate hierarchy
5. [Final Output] Generate cleaned text
""".strip()
output_requirement = 'Response must contain ONLY cleaned text between <cleaned_start> and <cleaned_end>.'
output_requirement = 'Response must contain ONLY the final cleaned text between <cleaned_start> and <cleaned_end>. Do not include analysis, explanations, processing steps, or bullet lists outside the cleaned text.'
else:
processing_steps = """
处理步骤:
1. [标签分析] 识别并分类所有标记标签
2. [引用提取] 分离图片/表格/签名等引用内容
3. [字符审核] 记录特殊字符变更
4. [结构检查] 验证文本层级
5. [最终输出] 生成清洗后文本
""".strip()
output_requirement = '响应必须只包含清洗后文本,以<cleaned_start>开头,<cleaned_end>结尾,无其他内容。'

return f"""
{self.prompt_header}

{output_requirement}

待清洗内容:
{raw_content}

{processing_steps}

{output_requirement}
""".strip()

def _post_process(self, cleaned_text: str) -> str:
Expand Down
Loading