写得越完整，切得越稀碎？我用Python做了一个GEO上下文断裂检测器
·
一篇4000字的技术文章，人类阅读时会从标题一路看到结尾。
但进入检索系统后，它很可能不再以“完整文章”的形态存在，而是被拆成多个片段：
片段1：产品定义
片段2：关键参数
片段3：适用场景
片段4：测试结果
片段5：注意事项
片段6：参考资料
问题来了。
如果片段2开头写的是：
它的最大工作压力为1.0 MPa。
“它”是谁？
如果片段4只剩下：
测试结果提升了28%。
测试对象是什么？对比基准是什么？测试条件是什么？
如果片段5写着：
因此，不建议在该场景中使用。
“该场景”又是什么？
完整文章里，这些句子都能看懂；一旦脱离上下文，就会集体失忆。
这就是GEO内容工程中一个很容易被忽略的问题：
页面可读，不代表页面切片后仍然可读。
这篇文章不讨论某个AI平台究竟采用多大的切片窗口，也不假设所有生成式搜索系统使用相同的检索流程。
我们只做一个可重复运行的工程模拟：
将Markdown文章分别进行“固定长度切片”和“结构感知切片”，然后检测代词悬空、问题答案分离、参数失去对象、缩写缺少定义等上下文断裂问题。
最终会得到一个完整的Python工具：
GEO上下文断裂检测器
它可以输出：
-
固定长度切片结果；
-
结构感知切片结果；
-
每个片段的上下文风险；
-
两种切片方式的问题数量对比；
-
需要重写的高风险段落；
-
CSV和Markdown审计报告。
一、问题现场：文章没问题，片段全是谜语
先看一段正常的技术内容：
HF-200滤芯过滤器适用于中低黏度液体过滤。
该设备的额定流量为40 m³/h，最大工作压力为1.0 MPa。
在20℃清水、入口压力0.3 MPa的测试条件下，
HF-200的流量波动控制在±3%以内。
因此，它不适合直接处理含有大量长纤维杂质的介质。
完整阅读时，逻辑很清楚：
产品是谁
→ 参数是多少
→ 测试条件是什么
→ 测试结果如何
→ 哪些场景不适用
如果系统按照固定字符数切开，可能变成：
片段1
HF-200滤芯过滤器适用于中低黏度液体过滤。
该设备的额定流量为40 m³/h，
片段2
最大工作压力为1.0 MPa。在20℃清水、
入口压力0.3 MPa的测试条件下，
片段3
HF-200的流量波动控制在±3%以内。
因此，它不适合直接处理含有大量长纤维杂质的介质。
这里至少出现了三个问题。
问题一：参数脱离对象
最大工作压力为1.0 MPa
这个参数属于哪个型号？
片段中没有产品名称。
问题二：测试条件与结果分离
片段2只有测试条件，片段3才出现测试结果。
如果检索系统只召回其中一个片段，信息就不完整。
问题三：代词指向片段外部
因此，它不适合……
“它”依赖前文中的HF-200。
当片段独立出现时，代词没有明确指向。
二、什么是“切片后可独立理解”？
一个合格的内容片段，不一定要包含整篇文章的全部背景，但至少应该回答四个问题：
1. 当前讨论的对象是谁？
2. 当前表达的属性或结论是什么？
3. 结论在什么条件下成立？
4. 是否需要依赖片段外部信息才能理解？
例如：
上下文依赖版本
它的额定流量为40 m³/h。
相对独立版本
HF-200滤芯过滤器的额定流量为40 m³/h。
再比如：
条件缺失版本
测试结果显示流量波动小于3%。
条件完整版本
在20℃清水、入口压力0.3 MPa的测试条件下，
HF-200滤芯过滤器的流量波动小于3%。
第二种写法看起来有些重复，但它具备一个重要能力：
即使被单独提取，事实仍然保持对象、条件和结论完整。
三、两种切片方式有什么区别？
本文会模拟两种常见策略。
1. 固定长度切片
按照字符数量直接截断：
每500个字符切成一块。
优点：
-
实现简单；
-
切片长度稳定；
-
处理速度快。
缺点：
-
可能从句子中间截断；
-
可能把问题和答案拆开；
-
可能把表格、代码和引用拆散；
-
不理解标题层级。
2. 结构感知切片
按照Markdown结构进行处理：
标题
→ 段落
→ 句子
→ 代码块
→ 表格
同时保留章节路径：
工业过滤器
/ HF-200技术参数
/ 工作压力
优点：
-
尽量不切断完整句子；
-
代码块和表格保持完整；
-
片段保留所属章节；
-
可以加入少量上下文重叠。
缺点：
-
实现更复杂；
-
长表格或长代码块可能超过限制；
-
仍然无法自动修复原文本中的指代问题。
四、系统架构：先切片，再检查上下文
整体流程如下：
graph TD
A[输入Markdown文章] --> B[解析标题和内容块]
B --> C1[固定长度切片]
B --> C2[结构感知切片]
C1 --> D[片段审计器]
C2 --> D
D --> E1[代词悬空检测]
D --> E2[参数脱离对象检测]
D --> E3[问题答案分离检测]
D --> E4[缩写缺少定义检测]
D --> E5[来源与结论分离检测]
D --> E6[片段过短或过长检测]
E1 --> F[生成风险清单]
E2 --> F
E3 --> F
E4 --> F
E5 --> F
E6 --> F
F --> G1[JSONL切片文件]
F --> G2[CSV审计明细]
F --> G3[Markdown报告]
需要强调：
这个工具不是任何AI平台切片算法的复刻。
它只是用可解释的规则模拟内容被拆分后的风险，帮助编辑在发布前发现明显问题。
五、项目目录
创建项目：
geo-chunk-audit/
├── geo_chunk_audit.py
├── article.md
└── output/
本文代码只使用Python标准库，不需要安装第三方依赖。
建议使用：
Python 3.10+
检查版本：
python --version
六、准备测试文章：article.md
创建article.md：
# HF-200滤芯过滤器技术说明
## 产品定义
HF-200滤芯过滤器是一种用于中低黏度液体精密过滤的设备。
它主要由壳体、滤芯、密封组件、压力表接口和排污口组成。
## 技术参数
该设备的额定流量为40 m³/h，最大工作压力为1.0 MPa。
推荐工作温度为5℃至80℃。
## 流量测试
测试条件如下：
- 测试介质：20℃清水
- 入口压力：0.3 MPa
- 过滤精度：20 μm
- 连续运行时间：8小时
测试结果显示，流量波动控制在±3%以内。
因此，它可以用于流量稳定的循环过滤系统。
## 哪些场景不适合使用HF-200？
不建议直接处理含有大量长纤维杂质的介质。
这是因为长纤维可能缠绕在滤芯表面，导致局部堵塞和压差快速升高。
## 缩写说明
CIP是Clean in Place的缩写，中文通常称为原位清洗。
该方法可以在不拆卸主要设备的情况下完成清洗。
## 数据来源
根据内部测试记录，HF-200在上述测试条件下连续运行8小时，
未出现密封泄漏。
这篇测试文章故意包含：
-
“它”；
-
“该设备”；
-
“上述测试条件”；
-
参数与产品分离风险；
-
问题型标题；
-
英文缩写；
-
数据来源表达。
七、完整代码：GEO上下文断裂检测器
创建geo_chunk_audit.py：
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
GEO上下文断裂检测器
功能：
1. 解析Markdown标题、段落、列表、表格和代码块；
2. 生成固定长度切片；
3. 生成结构感知切片；
4. 检测代词悬空、参数脱离对象、问题答案分离、
缩写未定义、来源缺失和片段长度异常；
5. 输出JSONL、CSV和Markdown报告。
注意：
该工具是内容工程模拟器，不代表任何特定AI平台
真实使用的切片大小、召回机制或排序规则。
"""
from __future__ import annotations
import argparse
import csv
import json
import re
import sys
from collections import Counter
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Iterable
HEADING_PATTERN = re.compile(
r"^(#{1,6})\s+(.+?)\s*$"
)
UNIT_PATTERN = re.compile(
r"""
(?:
\d+(?:\.\d+)?
\s*
(?:
%|℃|°C|MPa|kPa|Pa|bar|
m³/h|m3/h|L/min|kg|g|
mm|cm|m|μm|um|小时|分钟|秒
)
)
""",
flags=re.IGNORECASE | re.VERBOSE,
)
ACRONYM_PATTERN = re.compile(
r"\b[A-Z][A-Z0-9-]{1,9}\b"
)
QUESTION_PATTERN = re.compile(
r"[？?]"
)
SOURCE_CUE_PATTERN = re.compile(
r"""
根据|
数据显示|
测试结果|
研究表明|
报告显示|
according\sto|
data\sshows?|
test\sresults?|
study
""",
flags=re.IGNORECASE | re.VERBOSE,
)
CITATION_PATTERN = re.compile(
r"""
https?://|
doi:|
\[\d+\]|
来源[:：]|
source[:：]|
参考资料|
测试记录|
报告编号
""",
flags=re.IGNORECASE | re.VERBOSE,
)
DANGLING_PREFIXES = (
"它",
"它们",
"其",
"该",
"这些",
"这种",
"上述",
"前者",
"后者",
"其中",
"对此",
"因此",
"但是",
"然而",
"同时",
"此外",
"这意味着",
"the device",
"it ",
"they ",
"this ",
"these ",
"therefore",
"however",
)
ACRONYM_WHITELIST = {
"MPA",
"KPA",
"PA",
"C",
"MM",
"CM",
"KG",
"URL",
"HTML",
"JSON",
"CSV",
"API",
"HTTP",
"HTTPS",
"UTF",
}
SEVERITY_POINTS = {
"high": 3,
"medium": 2,
"low": 1,
}
@dataclass
class Block:
kind: str
text: str
heading_path: list[str] = field(
default_factory=list
)
@dataclass
class Chunk:
chunk_id: str
mode: str
heading: str
text: str
char_count: int
@dataclass
class Finding:
chunk_id: str
mode: str
severity: str
issue_type: str
detail: str
suggestion: str
def clean_text(text: str) -> str:
"""合并多余空格，但保留必要换行。"""
lines = []
for line in text.splitlines():
normalized = re.sub(
r"[ \t]+",
" ",
line,
).strip()
if normalized:
lines.append(normalized)
return "\n".join(lines).strip()
def flush_buffer(
blocks: list[Block],
buffer: list[str],
kind: str,
heading_path: list[str],
) -> None:
"""将临时缓冲区转换为内容块。"""
if not buffer:
return
text = clean_text("\n".join(buffer))
if text:
blocks.append(
Block(
kind=kind,
text=text,
heading_path=list(
heading_path
),
)
)
buffer.clear()
def parse_markdown(
source: str,
) -> list[Block]:
"""解析Markdown为结构化内容块。"""
lines = source.splitlines()
blocks: list[Block] = []
heading_stack: list[str] = []
paragraph_buffer: list[str] = []
list_buffer: list[str] = []
table_buffer: list[str] = []
code_buffer: list[str] = []
in_code = False
code_fence = ""
for line in lines:
stripped = line.strip()
if in_code:
code_buffer.append(line)
if stripped.startswith(
code_fence
):
blocks.append(
Block(
kind="code",
text="\n".join(
code_buffer
),
heading_path=list(
heading_stack
),
)
)
code_buffer.clear()
in_code = False
code_fence = ""
continue
if stripped.startswith(
("```", "~~~")
):
flush_buffer(
blocks,
paragraph_buffer,
"paragraph",
heading_stack,
)
flush_buffer(
blocks,
list_buffer,
"list",
heading_stack,
)
flush_buffer(
blocks,
table_buffer,
"table",
heading_stack,
)
in_code = True
code_fence = stripped[:3]
code_buffer.append(line)
continue
heading_match = (
HEADING_PATTERN.match(line)
)
if heading_match:
flush_buffer(
blocks,
paragraph_buffer,
"paragraph",
heading_stack,
)
flush_buffer(
blocks,
list_buffer,
"list",
heading_stack,
)
flush_buffer(
blocks,
table_buffer,
"table",
heading_stack,
)
level = len(
heading_match.group(1)
)
heading_text = (
heading_match.group(2)
.strip()
)
heading_stack = (
heading_stack[: level - 1]
)
while len(
heading_stack
) < level - 1:
heading_stack.append(
"未命名章节"
)
heading_stack.append(
heading_text
)
blocks.append(
Block(
kind="heading",
text=heading_text,
heading_path=list(
heading_stack
),
)
)
continue
is_list = bool(
re.match(
r"^(?:[-*+]|\d+[.)])\s+",
stripped,
)
)
is_table = (
stripped.startswith("|")
and stripped.endswith("|")
)
if not stripped:
flush_buffer(
blocks,
paragraph_buffer,
"paragraph",
heading_stack,
)
flush_buffer(
blocks,
list_buffer,
"list",
heading_stack,
)
flush_buffer(
blocks,
table_buffer,
"table",
heading_stack,
)
continue
if is_list:
flush_buffer(
blocks,
paragraph_buffer,
"paragraph",
heading_stack,
)
flush_buffer(
blocks,
table_buffer,
"table",
heading_stack,
)
list_buffer.append(line)
continue
if is_table:
flush_buffer(
blocks,
paragraph_buffer,
"paragraph",
heading_stack,
)
flush_buffer(
blocks,
list_buffer,
"list",
heading_stack,
)
table_buffer.append(line)
continue
flush_buffer(
blocks,
list_buffer,
"list",
heading_stack,
)
flush_buffer(
blocks,
table_buffer,
"table",
heading_stack,
)
paragraph_buffer.append(line)
flush_buffer(
blocks,
paragraph_buffer,
"paragraph",
heading_stack,
)
flush_buffer(
blocks,
list_buffer,
"list",
heading_stack,
)
flush_buffer(
blocks,
table_buffer,
"table",
heading_stack,
)
if code_buffer:
blocks.append(
Block(
kind="code",
text="\n".join(
code_buffer
),
heading_path=list(
heading_stack
),
)
)
return blocks
def split_sentences(
text: str,
) -> list[str]:
"""按中英文标点进行启发式分句。"""
normalized = clean_text(text)
if not normalized:
return []
parts = re.split(
r"""
(?<=[。！？!?；;])
\s*
|
(?<=\.)
\s+
(?=[A-Z0-9\u4e00-\u9fff])
""",
normalized,
flags=re.VERBOSE,
)
return [
part.strip()
for part in parts
if part.strip()
]
def build_plain_text(
blocks: list[Block],
) -> str:
"""将Markdown块还原为连续文本。"""
parts = []
for block in blocks:
if block.kind == "heading":
parts.append(
f"\n{block.text}\n"
)
else:
parts.append(block.text)
return clean_text(
"\n\n".join(parts)
)
def fixed_chunks(
text: str,
max_chars: int,
) -> list[Chunk]:
"""固定字符数切片，用作对照组。"""
chunks = []
for index, start in enumerate(
range(0, len(text), max_chars),
start=1,
):
content = text[
start : start + max_chars
].strip()
if not content:
continue
chunks.append(
Chunk(
chunk_id=(
f"fixed-{index:03d}"
),
mode="fixed",
heading="",
text=content,
char_count=len(content),
)
)
return chunks
def heading_label(
heading_path: list[str],
) -> str:
"""把标题层级转换为面包屑。"""
return " / ".join(
item
for item in heading_path
if item
)
def section_blocks(
blocks: list[Block],
) -> list[
tuple[list[str], list[Block]]
]:
"""按标题路径将正文分组。"""
sections: list[
tuple[list[str], list[Block]]
] = []
current_path: list[str] = []
current_blocks: list[Block] = []
for block in blocks:
if block.kind == "heading":
if current_blocks:
sections.append(
(
list(current_path),
current_blocks,
)
)
current_blocks = []
current_path = list(
block.heading_path
)
continue
block_path = (
block.heading_path
or current_path
)
if (
current_blocks
and block_path
!= current_path
):
sections.append(
(
list(current_path),
current_blocks,
)
)
current_blocks = []
current_path = list(
block_path
)
current_blocks.append(block)
if current_blocks:
sections.append(
(
list(current_path),
current_blocks,
)
)
return sections
def smart_chunks(
blocks: list[Block],
max_chars: int,
overlap_sentences: int,
) -> list[Chunk]:
"""
结构感知切片。
规则：
1. 不跨越标题章节；
2. 表格和代码块尽量保持完整；
3. 普通段落按句子切分；
4. 新片段携带章节路径；
5. 可保留少量句子重叠。
"""
chunks: list[Chunk] = []
chunk_counter = 0
for path, section in section_blocks(
blocks
):
heading = heading_label(path)
prefix = (
f"【章节：{heading}】\n"
if heading
else ""
)
current_sentences: list[str] = []
def flush_current() -> None:
nonlocal chunk_counter
nonlocal current_sentences
if not current_sentences:
return
body = "\n".join(
current_sentences
).strip()
content = (
prefix + body
).strip()
chunk_counter += 1
chunks.append(
Chunk(
chunk_id=(
f"smart-"
f"{chunk_counter:03d}"
),
mode="smart",
heading=heading,
text=content,
char_count=len(content),
)
)
if overlap_sentences > 0:
current_sentences = (
current_sentences[
-overlap_sentences:
]
)
else:
current_sentences = []
for block in section:
if block.kind in {
"code",
"table",
}:
flush_current()
atomic_text = (
prefix + block.text
).strip()
chunk_counter += 1
chunks.append(
Chunk(
chunk_id=(
f"smart-"
f"{chunk_counter:03d}"
),
mode="smart",
heading=heading,
text=atomic_text,
char_count=len(
atomic_text
),
)
)
current_sentences = []
continue
sentences = split_sentences(
block.text
)
for sentence in sentences:
candidate_sentences = [
*current_sentences,
sentence,
]
candidate = (
prefix
+ "\n".join(
candidate_sentences
)
).strip()
if (
current_sentences
and len(candidate)
> max_chars
):
previous_tail = (
current_sentences[
-overlap_sentences:
]
if overlap_sentences
else []
)
flush_current()
current_sentences = list(
previous_tail
)
retry_candidate = (
prefix
+ "\n".join(
[
*current_sentences,
sentence,
]
)
).strip()
if (
current_sentences
and len(retry_candidate)
> max_chars
):
current_sentences = []
current_sentences.append(
sentence
)
single_candidate = (
prefix
+ "\n".join(
current_sentences
)
).strip()
if len(
single_candidate
) > max_chars:
flush_current()
current_sentences = []
flush_current()
return chunks
def first_body_line(
chunk: Chunk,
) -> str:
"""忽略章节前缀，提取第一行正文。"""
lines = [
line.strip()
for line in chunk.text.splitlines()
if line.strip()
]
for line in lines:
if line.startswith(
"【章节："
):
continue
return line
return ""
def contains_entity(
text: str,
entities: list[str],
) -> bool:
lower_text = text.lower()
return any(
entity.lower() in lower_text
for entity in entities
if entity
)
def acronym_is_defined(
acronym: str,
text: str,
) -> bool:
escaped = re.escape(acronym)
patterns = (
rf"\b{escaped}\b"
r".{0,20}(?:全称|是|即|缩写)",
rf"(?:全称|缩写为|简称)"
rf".{{0,20}}\b{escaped}\b",
rf"[A-Za-z][A-Za-z ]{{3,}}"
rf"\(\s*{escaped}\s*\)",
rf"\b{escaped}\b"
rf".{0,10}\("
rf"[A-Za-z][A-Za-z ]{{3,}}\)",
)
return any(
re.search(
pattern,
text,
flags=re.IGNORECASE,
)
for pattern in patterns
)
def detect_findings(
chunk: Chunk,
entities: list[str],
min_chars: int,
max_chars: int,
) -> list[Finding]:
"""检测单个片段的上下文风险。"""
findings: list[Finding] = []
body_start = (
first_body_line(chunk)
.lower()
)
if any(
body_start.startswith(
prefix.lower()
)
for prefix in DANGLING_PREFIXES
):
findings.append(
Finding(
chunk_id=chunk.chunk_id,
mode=chunk.mode,
severity="high",
issue_type="代词或连接词悬空",
detail=(
f"片段开头为："
f"{first_body_line(chunk)[:80]}"
),
suggestion=(
"在片段首句重复产品名、"
"概念名或明确对象，"
"避免使用“它、该、上述、因此”"
"直接开头。"
),
)
)
units = UNIT_PATTERN.findall(
chunk.text
)
if (
units
and entities
and not contains_entity(
chunk.text,
entities,
)
):
findings.append(
Finding(
chunk_id=chunk.chunk_id,
mode=chunk.mode,
severity="high",
issue_type="参数脱离对象",
detail=(
"检测到参数或单位："
+ "、".join(units[:5])
+ "，但未检测到配置的核心实体。"
),
suggestion=(
"在参数句中明确写出产品、"
"型号或测试对象，不要只写"
"“额定流量为……”"
),
)
)
acronyms = {
item.upper()
for item in ACRONYM_PATTERN.findall(
chunk.text
)
if item.upper()
not in ACRONYM_WHITELIST
}
undefined = [
acronym
for acronym in sorted(acronyms)
if not acronym_is_defined(
acronym,
chunk.text,
)
]
if undefined:
findings.append(
Finding(
chunk_id=chunk.chunk_id,
mode=chunk.mode,
severity="medium",
issue_type="缩写缺少定义",
detail=(
"可能未在当前片段定义的缩写："
+ "、".join(undefined[:8])
),
suggestion=(
"缩写首次出现时同时写出全称；"
"重要片段中可重复一次定义。"
),
)
)
question_positions = [
match.start()
for match in QUESTION_PATTERN.finditer(
chunk.text
)
]
if question_positions:
last_question = max(
question_positions
)
remaining = chunk.text[
last_question + 1 :
].strip()
if len(remaining) < 35:
findings.append(
Finding(
chunk_id=chunk.chunk_id,
mode=chunk.mode,
severity="high",
issue_type="问题与答案可能分离",
detail=(
"问题出现在片段末尾，"
"其后正文不足35个字符。"
),
suggestion=(
"确保问题标题和直接答案"
"进入同一片段；答案第一句"
"应明确回应问题。"
),
)
)
if SOURCE_CUE_PATTERN.search(
chunk.text
) and not CITATION_PATTERN.search(
chunk.text
):
findings.append(
Finding(
chunk_id=chunk.chunk_id,
mode=chunk.mode,
severity="medium",
issue_type="结论与来源分离",
detail=(
"片段包含“根据、数据显示、"
"测试结果”等表达，"
"但没有发现来源标识。"
),
suggestion=(
"让结论与测试编号、来源名称、"
"链接或参考文献位于同一片段。"
),
)
)
if chunk.char_count < min_chars:
findings.append(
Finding(
chunk_id=chunk.chunk_id,
mode=chunk.mode,
severity="low",
issue_type="片段过短",
detail=(
f"片段长度仅"
f"{chunk.char_count}个字符。"
),
suggestion=(
"检查是否可以与相邻定义、"
"答案或说明合并。"
),
)
)
if chunk.char_count > (
max_chars * 1.35
):
findings.append(
Finding(
chunk_id=chunk.chunk_id,
mode=chunk.mode,
severity="medium",
issue_type="原子块过长",
detail=(
f"片段长度为"
f"{chunk.char_count}个字符，"
"明显超过目标长度。"
),
suggestion=(
"长代码块或表格可以保留完整，"
"但应增加独立摘要；普通正文"
"应继续按句子或小节拆分。"
),
)
)
if (
chunk.mode == "fixed"
and not chunk.heading
):
findings.append(
Finding(
chunk_id=chunk.chunk_id,
mode=chunk.mode,
severity="low",
issue_type="缺少章节上下文",
detail=(
"固定长度片段没有携带"
"所属标题路径。"
),
suggestion=(
"切片时附带H1/H2/H3"
"标题路径，帮助片段保留主题。"
),
)
)
return findings
def audit_chunks(
chunks: list[Chunk],
entities: list[str],
min_chars: int,
max_chars: int,
) -> list[Finding]:
findings = []
for chunk in chunks:
findings.extend(
detect_findings(
chunk=chunk,
entities=entities,
min_chars=min_chars,
max_chars=max_chars,
)
)
return findings
def write_jsonl(
path: Path,
chunks: Iterable[Chunk],
) -> None:
with path.open(
"w",
encoding="utf-8",
) as file:
for chunk in chunks:
file.write(
json.dumps(
asdict(chunk),
ensure_ascii=False,
)
+ "\n"
)
def write_findings_csv(
path: Path,
findings: list[Finding],
) -> None:
fieldnames = [
"chunk_id",
"mode",
"severity",
"issue_type",
"detail",
"suggestion",
]
with path.open(
"w",
encoding="utf-8-sig",
newline="",
) as file:
writer = csv.DictWriter(
file,
fieldnames=fieldnames,
)
writer.writeheader()
for finding in findings:
writer.writerow(
asdict(finding)
)
def risk_score(
findings: list[Finding],
) -> int:
return sum(
SEVERITY_POINTS.get(
finding.severity,
0,
)
for finding in findings
)
def issue_counter(
findings: list[Finding],
mode: str,
) -> Counter:
return Counter(
finding.issue_type
for finding in findings
if finding.mode == mode
)
def escape_markdown(
value: object,
) -> str:
return (
str(value)
.replace("|", "\\|")
.replace("\n", " ")
)
def build_report(
fixed: list[Chunk],
smart: list[Chunk],
findings: list[Finding],
max_chars: int,
) -> str:
fixed_findings = [
item
for item in findings
if item.mode == "fixed"
]
smart_findings = [
item
for item in findings
if item.mode == "smart"
]
fixed_score = risk_score(
fixed_findings
)
smart_score = risk_score(
smart_findings
)
fixed_counter = issue_counter(
findings,
"fixed",
)
smart_counter = issue_counter(
findings,
"smart",
)
all_issue_types = sorted(
set(fixed_counter)
| set(smart_counter)
)
high_risk = [
item
for item in findings
if item.severity == "high"
]
lines = [
"# GEO上下文断裂审计报告",
"",
"## 一、切片概况",
"",
f"- 目标片段长度：{max_chars}字符",
f"- 固定长度片段：{len(fixed)}个",
f"- 结构感知片段：{len(smart)}个",
f"- 固定切片风险分：{fixed_score}",
f"- 结构感知切片风险分：{smart_score}",
"",
"> 风险分仅用于比较两种切片方式，"
"不是AI平台的引用概率或排名分数。",
"",
"## 二、问题数量对比",
"",
"| 问题类型 | 固定切片 | 结构感知切片 |",
"|---|---:|---:|",
]
for issue_type in all_issue_types:
lines.append(
f"| {escape_markdown(issue_type)} "
f"| {fixed_counter.get(issue_type, 0)} "
f"| {smart_counter.get(issue_type, 0)} |"
)
lines.extend(
[
"",
"## 三、高风险问题",
"",
]
)
if not high_risk:
lines.append(
"未发现高风险问题。"
)
else:
lines.extend(
[
"| 片段 | 模式 | 问题 | 说明 | 建议 |",
"|---|---|---|---|---|",
]
)
for item in high_risk:
lines.append(
"| {chunk} | {mode} | {issue} | "
"{detail} | {suggestion} |".format(
chunk=escape_markdown(
item.chunk_id
),
mode=escape_markdown(
item.mode
),
issue=escape_markdown(
item.issue_type
),
detail=escape_markdown(
item.detail
),
suggestion=escape_markdown(
item.suggestion
),
)
)
lines.extend(
[
"",
"## 四、结构感知片段预览",
"",
]
)
for chunk in smart[:5]:
lines.extend(
[
f"### {chunk.chunk_id}",
"",
f"- 章节：{chunk.heading or '无'}",
f"- 长度：{chunk.char_count}",
"",
"```text",
chunk.text[:800],
"```",
"",
]
)
lines.extend(
[
"## 五、建议处理顺序",
"",
"1. 先修复以“它、该、上述、因此”等词开头的高风险片段。",
"2. 在参数句中重复产品名、型号或测试对象。",
"3. 确保问题标题与直接答案位于同一片段。",
"4. 将测试条件、结果和来源放在相邻段落中。",
"5. 为缩写补充全称，不依赖页面其他章节解释。",
"6. 对表格和代码块增加一段可独立理解的文字摘要。",
"7. 修改文章后重新运行，比较两种切片的风险变化。",
"",
]
)
return "\n".join(lines)
def parse_arguments() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description=(
"模拟Markdown内容切片，"
"检测GEO上下文断裂问题。"
)
)
parser.add_argument(
"input",
help="输入Markdown文件。",
)
parser.add_argument(
"--output",
default="output",
help="输出目录，默认output。",
)
parser.add_argument(
"--max-chars",
type=int,
default=500,
help="目标片段最大字符数。",
)
parser.add_argument(
"--min-chars",
type=int,
default=80,
help="片段过短判定字符数。",
)
parser.add_argument(
"--overlap",
type=int,
default=1,
help="结构感知切片保留的重叠句数。",
)
parser.add_argument(
"--entities",
default="",
help=(
"核心实体，使用英文逗号分隔，"
"例如HF-200,滤芯过滤器。"
),
)
return parser.parse_args()
def validate_arguments(
max_chars: int,
min_chars: int,
overlap: int,
) -> None:
if max_chars < 100:
raise ValueError(
"max-chars不能小于100。"
)
if min_chars < 0:
raise ValueError(
"min-chars不能小于0。"
)
if min_chars >= max_chars:
raise ValueError(
"min-chars必须小于max-chars。"
)
if overlap < 0:
raise ValueError(
"overlap不能小于0。"
)
def main() -> int:
arguments = parse_arguments()
try:
validate_arguments(
max_chars=arguments.max_chars,
min_chars=arguments.min_chars,
overlap=arguments.overlap,
)
input_path = Path(
arguments.input
)
output_path = Path(
arguments.output
)
output_path.mkdir(
parents=True,
exist_ok=True,
)
source = input_path.read_text(
encoding="utf-8",
)
blocks = parse_markdown(
source
)
if not blocks:
raise ValueError(
"输入文件没有可分析内容。"
)
entities = [
item.strip()
for item in arguments.entities.split(
","
)
if item.strip()
]
plain_text = build_plain_text(
blocks
)
fixed = fixed_chunks(
text=plain_text,
max_chars=arguments.max_chars,
)
smart = smart_chunks(
blocks=blocks,
max_chars=arguments.max_chars,
overlap_sentences=(
arguments.overlap
),
)
findings = audit_chunks(
chunks=[*fixed, *smart],
entities=entities,
min_chars=arguments.min_chars,
max_chars=arguments.max_chars,
)
fixed_file = (
output_path
/ "fixed_chunks.jsonl"
)
smart_file = (
output_path
/ "smart_chunks.jsonl"
)
findings_file = (
output_path
/ "chunk_findings.csv"
)
report_file = (
output_path
/ "chunk_audit_report.md"
)
write_jsonl(
fixed_file,
fixed,
)
write_jsonl(
smart_file,
smart,
)
write_findings_csv(
findings_file,
findings,
)
report = build_report(
fixed=fixed,
smart=smart,
findings=findings,
max_chars=arguments.max_chars,
)
report_file.write_text(
report,
encoding="utf-8",
)
fixed_findings = [
item
for item in findings
if item.mode == "fixed"
]
smart_findings = [
item
for item in findings
if item.mode == "smart"
]
print("")
print("分析完成。")
print(
f"Markdown块数量：{len(blocks)}"
)
print(
f"固定长度片段：{len(fixed)}"
)
print(
f"结构感知片段：{len(smart)}"
)
print(
"固定切片风险分："
f"{risk_score(fixed_findings)}"
)
print(
"结构感知切片风险分："
f"{risk_score(smart_findings)}"
)
print(
f"审计报告：{report_file}"
)
return 0
except (
FileNotFoundError,
UnicodeDecodeError,
ValueError,
OSError,
) as error:
print(
f"执行失败：{error}",
file=sys.stderr,
)
return 1
if __name__ == "__main__":
raise SystemExit(main())
八、运行检测器
执行：
python geo_chunk_audit.py article.md \
--max-chars 300 \
--min-chars 60 \
--overlap 1 \
--entities "HF-200,HF-200滤芯过滤器,滤芯过滤器"
Windows命令行可以写成一行：
python geo_chunk_audit.py article.md --max-chars 300 --min-chars 60 --overlap 1 --entities "HF-200,HF-200滤芯过滤器,滤芯过滤器"
执行完成后会生成：
output/
├── fixed_chunks.jsonl
├── smart_chunks.jsonl
├── chunk_findings.csv
└── chunk_audit_report.md
终端输出类似：
分析完成。
Markdown块数量：18
固定长度片段：5
结构感知片段：7
固定切片风险分：18
结构感知切片风险分：9
审计报告：output/chunk_audit_report.md
具体分数取决于文章内容、切片长度和实体配置。
不要把示例数字当成固定结果。
九、四个输出文件分别怎么看？
1. fixed_chunks.jsonl
记录固定长度切片。
每行是一个JSON对象：
{
"chunk_id": "fixed-001",
"mode": "fixed",
"heading": "",
"text": "HF-200滤芯过滤器技术说明……",
"char_count": 300
}
它主要用于观察：
-
是否从句子中间切断；
-
问题标题是否落在片段结尾；
-
参数是否与产品名称分开；
-
测试条件是否与结果分开。
2. smart_chunks.jsonl
记录结构感知切片：
{
"chunk_id": "smart-003",
"mode": "smart",
"heading": "HF-200滤芯过滤器技术说明 / 技术参数",
"text": "【章节：HF-200滤芯过滤器技术说明 / 技术参数】\n该设备的额定流量为40 m³/h……",
"char_count": 96
}
它会保留章节路径，但仍可能出现：
该设备
它
上述条件
这种方法
所以结构感知切片只能减少机械截断，不能自动修复原文。
3. chunk_findings.csv
记录所有风险：
| 片段 | 模式 | 严重度 | 问题 |
|---|---|---|---|
| fixed-002 | fixed | high | 参数脱离对象 |
| fixed-003 | fixed | high | 问题与答案可能分离 |
| smart-002 | smart | high | 代词或连接词悬空 |
| smart-005 | smart | medium | 结论与来源分离 |
可以在Excel中筛选：
severity = high
优先处理高风险片段。
4. chunk_audit_report.md
报告会比较两种切片方式：
| 问题类型 | 固定切片 | 结构感知切片 |
|---|---:|---:|
| 缺少章节上下文 | 5 | 0 |
| 参数脱离对象 | 2 | 1 |
| 问题与答案可能分离 | 1 | 0 |
| 代词或连接词悬空 | 1 | 2 |
这里有一个很重要的现象：
结构感知切片可能解决标题丢失，却暴露更多代词问题。
原因不是算法变差，而是它把原来隐藏在长段落中的指代句独立了出来。
十、踩坑实录：把切片调大，问题就消失了吗？
看到上下文断裂，最直接的想法是：
把每个片段从300字符改成2000字符。
这样确实能减少部分切断问题，但会带来新麻烦。
1. 一个片段包含太多主题
例如同一片段中同时出现：
产品定义
技术参数
安装步骤
故障处理
维护周期
检索到这个片段时，相关信息密度可能下降。
2. 问题答案被其他内容淹没
用户问：
HF-200的最大工作压力是多少？
召回的片段却有2000字，答案只占其中一句。
3. 更新成本增加
产品参数只修改了一行，却需要重新处理一个超大片段。
所以切片长度并不是越大越好。
更合理的原则是：
一个片段尽量完成一个明确的信息任务。
十一、踩坑实录：重叠越多，上下文越安全吗？
重叠切片的逻辑是：
片段1结尾的句子
复制到
片段2开头
例如：
片段1：
HF-200的额定流量为40 m³/h。
最大工作压力为1.0 MPa。
片段2：
最大工作压力为1.0 MPa。
推荐工作温度为5℃至80℃。
适量重叠可以降低边界损失。
但重叠过多也会产生问题。
1. 重复内容大量增加
如果每个片段都重复前一片段的30%，整个索引会出现大量近似文本。
2. 同一句话出现在多个片段
统计引用、召回和覆盖时更难判断到底是哪一块内容发挥了作用。
3. 错误也会被复制
原句如果含有模糊代词：
它的最大压力为1.0 MPa。
增加重叠只会让这句谜语出现更多次。
因此，本文默认：
--overlap 1
即保留一句重叠。
这只是测试起点，不是固定标准。
十二、解决方案：不要依赖代词维持信息密度
技术文章为了避免重复，经常使用：
它
该设备
该产品
这种方案
上述方法
前者
后者
在人类连续阅读场景中，这种表达很自然。
在内容切片场景中，过度使用代词会增加上下文依赖。
改写前
HF-200是一种滤芯过滤器。
它的额定流量为40 m³/h。
该设备适用于中低黏度液体。
此外，它不适合长纤维杂质较多的介质。
改写后
HF-200是一种用于中低黏度液体的滤芯过滤器。
HF-200的额定流量为40 m³/h。
当介质含有大量长纤维杂质时，
不建议使用HF-200滤芯过滤器，
因为长纤维可能缠绕滤芯并导致压差快速升高。
后者稍微重复了产品名称，却让每个事实都更容易独立成立。
十三、解决方案：让“对象、属性、值”出现在同一句中
技术参数最常见的写法是表格：
| 参数 | 数值 |
|---|---|
| 额定流量 | 40 m³/h |
| 最大压力 | 1.0 MPa |
| 工作温度 | 5℃—80℃ |
人类知道表格属于当前产品。
如果表格被独立处理，可能只剩下：
额定流量：40 m³/h
最大压力：1.0 MPa
此时对象消失。
可以在表格前增加摘要：
HF-200滤芯过滤器的额定流量为40 m³/h，
最大工作压力为1.0 MPa，
推荐工作温度为5℃至80℃。
详细参数见下表。
这样即使表格解析效果不理想，正文中仍然存在完整事实。
参数表达可以遵循：
实体 + 属性 + 值 + 单位 + 条件
例如：
在20℃清水条件下，
HF-200滤芯过滤器的额定流量为40 m³/h。
十四、解决方案：问题标题后第一句必须直接回答
下面这种写法很常见：
## HF-200适合哪些介质？
在工业生产过程中，液体性质差异较大。
不同工况对过滤设备提出了不同要求。
选择设备时需要综合考虑多方面因素。
HF-200适合中低黏度液体。
如果切片边界出现在第三句之后，问题和答案就会分离。
更好的写法是：
## HF-200适合哪些介质？
HF-200适合水、低黏度油品及其他中低黏度液体。
选型时还需要检查介质温度、颗粒类型、
目标过滤精度和设计流量。
原则非常简单：
问题标题后第一句先给答案，再解释背景。
不要让读者和检索程序一起参加“答案寻宝游戏”。
十五、解决方案：测试条件、结果和来源不要分居三地
一个技术事实经常被拆成三部分：
页面前半部分
测试介质为20℃清水。
页面中间
流量波动低于3%。
页面末尾
数据来源：内部测试报告。
完整页面中勉强能建立关系，切片后很容易变成三个独立事实。
推荐将其合并：
根据编号为HF200-FT-2026-03的内部测试记录，
在20℃清水、入口压力0.3 MPa、
连续运行8小时的条件下，
HF-200滤芯过滤器的流量波动控制在±3%以内。
这一句同时包含：
来源
对象
条件
结果
即使单独出现，也具有基本可验证性。
十六、代码块和表格为什么不能直接暴力切开？
假设文章中有一段JSON-LD：
{
"@type": "Product",
"name": "HF-200",
"model": "HF-200",
"material": "Stainless Steel"
}
如果从中间切开：
片段1
{
"@type": "Product",
"name": "HF-200",
片段2
"model": "HF-200",
"material": "Stainless Steel"
}
两个片段都不是完整JSON。
因此，本文代码将：
代码块
表格
视为原子块，优先保持完整。
但这会产生另一个问题：
一个超长代码块可能远远超过目标片段长度。
解决方法不是强行切代码，而是增加文字摘要：
下面的JSON-LD定义了HF-200产品实体，
其中name和model均为HF-200，
材质属性为Stainless Steel。
这样即使代码块没有被完整召回，核心信息仍可通过自然语言获取。
十七、缩写为什么要在重要片段中重复定义？
假设文章开头写：
CIP是Clean in Place的缩写，即原位清洗。
三千字后又出现：
该设备支持CIP。
如果第二句话被独立切片，读者可能知道CIP是什么，也可能不知道。
对于通用缩写，可以不必每次解释。
但对于以下内容，建议提高定义密度：
-
行业专有缩写；
-
企业内部术语；
-
产品系列名称；
-
测试方法简称；
-
标准流程简称。
例如：
HF-200支持CIP原位清洗，
即在不拆卸主要设备的情况下完成清洗。
不是要求每句话都写百科全书，而是让关键能力描述脱离上下文后仍可理解。
十八、进一步升级：检测跨片段指代关系
本文代码只检查片段是否以模糊代词开头。
进一步可以建立更复杂的规则：
片段中出现“它”
→ 向前寻找最近实体
→ 判断最近实体是否在当前片段
→ 如果不在，则标记为跨片段指代
伪代码如下：
def detect_cross_chunk_reference(
chunk_text: str,
entities: list[str],
) -> bool:
pronouns = [
"它",
"该设备",
"该产品",
"上述方法",
]
has_pronoun = any(
item in chunk_text
for item in pronouns
)
has_entity = any(
entity in chunk_text
for entity in entities
)
return (
has_pronoun
and not has_entity
)
还可以进一步接入：
-
中文指代消解模型；
-
命名实体识别；
-
句向量相似度；
-
大语言模型审计。
但在加入复杂模型前，简单规则通常已经能发现大量明显问题。
十九、进一步升级：为每种内容类型设置切片策略
不同内容不应该使用同一种切片方式。
FAQ
推荐：
一个问题
+
一个直接答案
+
必要解释
问题和答案必须位于同一块。
产品参数
推荐：
产品名称
+
参数摘要
+
参数表
+
测试条件
不要只保留孤立数值。
操作教程
推荐按步骤组切分：
准备条件
步骤1—3
步骤4—6
故障处理
不要从某个命令中间截断。
对比文章
推荐保证每个片段至少包含：
对比对象A
对比对象B
对比维度
对比结论
否则片段可能只剩下单方面描述。
案例内容
推荐保证：
背景
问题
方法
结果
限制
至少有三个要素位于同一片段。
二十、避坑指南：切片优化最容易犯的7个错误
1. 把片段长度当成万能参数
内容结构有问题时，调整字符数只能掩盖问题。
2. 为了独立理解，机械重复整段背景
必要实体可以重复，但不需要每个片段都复制公司介绍。
3. 只优化标题，不优化正文首句
标题能提供主题，首句负责给出事实和答案。
4. 参数表没有自然语言摘要
表格解析失败时，页面将失去核心参数。
5. 问题和答案分属不同小节
问题型标题后应立即给出直接答案。
6. 测试结果没有条件和来源
孤立的百分比很容易失去意义。
7. 认为结构感知切片可以修复所有问题
切片器只能保护已有结构，无法替作者补充缺失对象、条件和证据。
二十一、下一步行动：先测试一篇核心文章
第一次使用时，不需要扫描整个网站。
可以按以下步骤执行。
第一步：选择一篇参数较多的文章
优先选择：
-
产品技术说明；
-
采购指南；
-
对比文章；
-
故障排查教程；
-
标准解读；
-
测试报告。
第二步：配置核心实体
例如：
--entities "HF-200,HF-200滤芯过滤器,工业过滤器"
实体越准确，参数脱离对象的检测越有效。
第三步：尝试三个切片长度
分别运行：
--max-chars 300
--max-chars 500
--max-chars 800
比较：
-
片段数量；
-
高风险问题；
-
问题答案分离；
-
参数脱离对象；
-
过短片段。
第四步：先修改正文，再调整切片参数
优先处理：
它
该设备
上述条件
测试结果显示
因此
最大压力为
额定流量为
这些高度依赖前文的表达。
第五步：重新运行
确认修改后：
-
高风险片段是否减少；
-
参数句是否包含实体；
-
问题和答案是否在同一片段；
-
测试条件和结果是否保持相邻。
二十二、总结：GEO内容真正的考验，是离开全文后还能不能讲清楚
写文章时，我们习惯依赖连续阅读：
上一段已经介绍过产品；
前面已经解释过缩写；
表格标题已经说明对象；
来源放在文末就够了。
但内容进入检索、切片和重新组合流程后，这些默认前提可能全部失效。
一个片段被单独取出时，需要重新面对几个问题：
“它”到底是谁？
“该参数”属于哪个产品？
“提升28%”和谁比较？
“上述条件”具体是什么？
“测试结果”来自哪份记录？
“因此”之前发生了什么？
所以，GEO内容优化不能只看完整页面读起来是否流畅。
还要做一次更残酷的测试：
随机截取其中一段，不看上文，它还能不能准确表达一个完整事实？
真正稳定的知识片段，通常包含：
明确实体
+
明确属性
+
明确结论
+
必要条件
+
适用边界
+
可验证来源
文章是给人连续阅读的。
知识片段还要经得住拆分、检索和重新组合。
当一段内容离开全文后依然知道自己在说谁、说什么、为什么成立时，它才不只是文章中的一句话，而是一个真正可复用的知识单元。
更多推荐
15
0- 0
已为社区贡献3条内容
所有评论(0)