Agent-Skills-for-Context-Engineering:Book SFT 流水线的文本分块策略详解——两层切分、场景感知分段与校验管线【免费下载链接】Agent-Skills-for-Context-EngineeringA comprehensive collection of Agent Skills for context engineering, multi-agent architectures, and production agent systems. Use when building, optimizing, or debugging agent systems that require effective context management.项目地址: https://gitcode.com/GitHub_Trending/ag/Agent-Skills-for-Context-Engineering本篇技术文章以 segmentation-strategies.md 为核心,系统讲解图书转 SFT 训练数据流水线中最关键的分段(Segmentation)环节:从段落累积的两层切分策略、LLM 辅助切分的零删除校验,到场景感知、对话感知、结果校验与管线集成。结合 book-sft-pipeline 示例技能中的概念实现源码与 Gertrude Stein 真实案例数据,你将掌握如何在 150–400 词的语义完整片段上构建风格迁移数据集,并理解分块参数选择对训练样本数量与风格还原效果的实际影响。一、分块问题:为什么书籍文本难以直接切分长篇小说是训练数据构造中一类特殊的长文本。参考文档开篇即指出,书籍文本存在四个使按字数均分失效的特征:段落长度极不均匀:部分作者会写出超过 1000 词的连续单段;对话密集段落:单个对话回合往往过短,独立成块后语义残缺;场景边界与字数不对齐:自然断点(章节切换、场景转换)不一定落在目标字数附近;文风在叙述、对话、铺陈之间切换:不同文体段落对完整语义单元的定义不同。分块质量会直接映射到模型行为上——糟糕的分段等于在训练数据里示范了坏的生成模式,模型将学会输出:未完成的思路(Incomplete thoughts)戛然而止的结尾(Abrupt endings)不连贯的转场(Incoherent transitions)碎片化的文风(Fragmented style)这也是 SKILL.md 中Intelligent Segmentation被列为 Book SFT 三大支柱之首的原因:Breaking mid-sentence teaches the model to produce fragmented output.二、两层切分策略(Two-Tier Strategy)参考文档给出的核心方案是快慢结合的两层架构:第一层用确定性算法处理绝大多数段落,第二层用 LLM 兜底处理超大段落。SKILL.md 在Integration with Context Engineering Skills一节也明确印证了这一映射:Tier 1 对应快速、确定性的压缩,Tier 2 对应LLM 辅助处理边缘情况。2.1 Tier 1:基于段落累积的默认策略Tier 1 是面向结构良好文本的默认方法。参考文档给出的实现如下(类定义与核心循环完整保留自 segmentation-strategies.md):class Tier1Segmenter: def __init__(self, min_words: int 250, max_words: int 650): self.min_words min_words self.max_words max_words def segment(self, text: str) - list[Chunk]: paragraphs self._split_paragraphs(text) chunks [] current ChunkBuilder() for para in paragraphs: word_count len(para.split()) # Check if single paragraph exceeds max if word_count self.max_words: # Finalize current chunk if exists if current.word_count 0: chunks.append(current.build()) current ChunkBuilder() # Mark for Tier 2 processing chunks.append(Chunk( textpara, requires_tier2True, word_countword_count )) continue # Would this paragraph overflow current chunk? if current.word_count word_count self.max_words: if current.word_count self.min_words: chunks.append(current.build()) current ChunkBuilder() current.add(para) # Dont forget the last chunk if current.word_count 0: chunks.append(current.build()) return chunks def _split_paragraphs(self, text: str) - list[str]: # Split on double newlines, preserve single newlines within paragraphs text.split(\n\n) return [p.strip() for p in paragraphs if p.strip()]关键参数与逻辑要点:参数默认值作用min_words250块的最小词数下限;累积不足此值时允许继续追加,避免产生过碎片max_words650块的最大词数上限;超过即触发封块算法行为可以拆成三条路径:单段超长路径:若某段本身超过max_words,先封掉当前累积块,再将该段整体放入结果并打上requires_tier2True标记,交由第二层处理——注意此处不破坏段落,而是整体让渡给 LLM;溢出封块路径:新段落加入会超过max_words时,仅当当前块已满足min_words才封块重置;若当前块还很小,则继续追加(意味着块可能短暂越过max_words,直到满足最小词数后才允许切出,这是下限优先的设计权衡);收尾路径:循环结束后,缓冲区非空则作为最后一个块输出。段落切分本身采用最朴素但稳健的规则:按双换行\n\n断段,保留段内单换行,丢弃空白段。这与 pipeline_example.py 概念实现中的做法一致(text.split(\n\n)后strip()过滤)。2.2 Tier 2:LLM 辅助切分对第一层整体让渡的超大段落,使用 LLM 在语法自然处二次切分。参考文档的实现完整如下:class Tier2Segmenter: def __init__(self, model: str gpt-4o): self.model model self.prompt_template self._load_prompt() async def segment(self, oversized_chunk: Chunk) - list[Chunk]: Split an oversized paragraph using LLM. response await self._call_llm( self.prompt_template.format(textoversized_chunk.text) ) segments self._parse_segments(response) # Validate zero-deletion original_words len(oversized_chunk.text.split()) segmented_words sum(len(s.split()) for s in segments) if abs(original_words - segmented_words) 5: # Allow tiny variance raise SegmentationError( fWord count mismatch: {original_words} - {segmented_words} ) return [ Chunk(texts, requires_tier2False, word_countlen(s.split())) for s in segments ] def _load_prompt(self) - str: return Segment this text into excerpts of minimum 300-350 words. Requirements: - Each excerpt must be grammatically complete from start - Each excerpt must not feel abruptly cut off - Zero deletion - maintain original word count exactly - Break at grammatically natural places: * After complete dialogue exchanges * At scene transitions * After complete thoughts or descriptions * Where a paragraph break would naturally occur - Avoid breaking into too many small excerpts - Start directly with the excerpts - Separate excerpts with SEGMENT Text to segment: {text} def _parse_segments(self, response: str) - list[str]: segments response.split(SEGMENT) return [s.strip() for s in segments if s.strip()]这段设计里有三个值得注意的工程决策:零删除校验(Zero-deletion check):切分前后总词数偏差超过 5 词即抛SegmentationError。这是防 LLM 改写/删字的硬约束——SFT 数据集要求 assistant 侧必须是原文逐字片段,任何删改都会污染风格样本。切分提示词约束:每段目标 300–350 词,且只允许在完整对话之后、场景转换处、完整思想之后、自然段落断点处切分,并要求避免切出过多小段、直接以片段开始输出。机器可解析的分隔符:用SEGMENT作为片段边界,解析时split后剔除空串,避免 LLM 输出前言污染片段内容。三、场景感知分段(Scene-Aware Segmentation)对追求更高切分质量的场景,参考文档建议优先利用书中已有的场景分隔标记,而不是任意段落边界:class SceneAwareSegmenter: Prefer breaking at scene boundaries when within word limits. SCENE_MARKERS [ r\n\n\* \* \*\n\n, # Asterisk dividers r\n\n---\n\n, # Dash dividers r\n\n###\n\n, # Hash dividers r\n\nCHAPTER \d, # Chapter headings r\n\n[A-Z]{3,}\n\n, # All-caps scene breaks ] def find_scene_breaks(self, text: str) - list[int]: Find character positions of scene breaks. breaks [] for pattern in self.SCENE_MARKERS: for match in re.finditer(pattern, text): breaks.append(match.start()) return sorted(set(breaks)) def segment_with_scenes(self, text: str) - list[Chunk]: scene_breaks self.find_scene_breaks(text) # If scene breaks exist, prefer them over arbitrary paragraph breaks if scene_breaks: return self._segment_at_scenes(text, scene_breaks) else: return Tier1Segmenter().segment(text)五类分隔标记覆盖了英语小说最常见的排版约定:星号分隔行、短横线分隔、井号分隔、CHAPTER n章节标题、全大写场景行。策略是有场景标记就用场景标记,没有则回退 Tier 1,属于典型的启发式优先、确定性兜底。需要说明:代码片段中使用了re.finditer但未展示import re,且_segment_at_scenes的具体实现留白,从源码结构看该类是设计参考(design reference)而非可直接运行的完整模块。四、对话感知分段(Dialogue-Aware Segmentation)对话密集段落单独处理,核心原则是绝不从一轮完整对话中间切开:class DialogueAwareSegmenter: Group dialogue exchanges to maintain conversation coherence. def is_dialogue_paragraph(self, para: str) - bool: Check if paragraph is primarily dialogue. # Count dialogue markers quote_count para.count() para.count() word_count len(para.split()) # If more than 20% of words are in quotes, its dialogue-heavy return quote_count word_count * 0.2 def segment(self, text: str) - list[Chunk]: paragraphs text.split(\n\n) chunks [] current ChunkBuilder() in_dialogue_block False for para in paragraphs: is_dialogue self.is_dialogue_paragraph(para) # Dont break in the middle of a dialogue exchange if is_dialogue: in_dialogue_block True current.add(para) else: if in_dialogue_block: # End of dialogue block - good break point in_dialogue_block False if current.word_count 250: chunks.append(current.build()) current ChunkBuilder() current.add(para) # Check if weve exceeded max if current.word_count 650: chunks.append(current.build()) current ChunkBuilder() if current.word_count 0: chunks.append(current.build()) return chunks从实现逻辑看有两个关键设计:对话块状态机:进入对话段落后置in_dialogue_block True并只累积不切分;当出现第一个非对话段落时,视其为对话块结束、理想断点,此时若当前块已达 250 词下限则封块;引号密度启发式:以引号字符数 词数 × 0.2判定对话段落。严格来说这是引号字符数与词数之比的粗略代理指标(一个引号字符对应两个词才成立 20% 的占比),从源码结构看它属于低成本启发式,胜在无需句法分析,适合批量预处理。五、校验管线:分段结果必须过检参考文档要求每一批分段结果都要通过校验,校验器同时产出 errors 与 warnings 两级信号:class SegmentationValidator: def validate(self, chunks: list[Chunk]) - ValidationResult: errors [] warnings [] for i, chunk in enumerate(chunks): # Check word count bounds if chunk.word_count 200: warnings.append(fChunk {i}: Only {chunk.word_count} words) if chunk.word_count 700: errors.append(fChunk {i}: {chunk.word_count} words exceeds max) # Check sentence completeness if not self._ends_with_terminal(chunk.text): errors.append(fChunk {i}: Ends mid-sentence) if not self._starts_grammatically(chunk.text): errors.append(fChunk {i}: Starts mid-sentence) # Check for orphaned dialogue if chunk.text.count() % 2 ! 0: warnings.append(fChunk {i}: Unbalanced quotes) return ValidationResult( validlen(errors) 0, errorserrors, warningswarnings ) def _ends_with_terminal(self, text: str) - bool: text text.strip() return text[-1] in .!?\— def _starts_grammatically(self, text: str) - bool: text text.strip() # Should start with capital or quote return text[0].isupper() or text[0] in \—五项校验规则及其严重级别:检查项规则级别词数下限 200词warning词数上限 700词error结尾完整末字符须为. ! ? —之一error开头完整首字符须为大写字母或 —之一error引号配平双引号数量为奇数warning注意校验边界(200/700)比分段边界(250/650)更宽,相当于给分段器留出一层容错缓冲:轻微越界只报警不阻断,严重越界(超出硬上限或句子不完整)才判定失败。这与 tinker-format.md 中序列长度应保持 1000 token 以内,更长的序列会稀释局部风格模式的约束互为补充——校验器的 700 词硬上限正是 token 预算的前置防线。六、策略选型与边界情况参考文档给出的策略对比表(完整继承):策略速度质量适用场景仅 Tier 1快中结构良好的散文Tier 1 Tier 2中高段落长短混合的文本场景感知快高场景分隔清晰的小说对话感知中高对话密集的小说四类典型边界情况及文档给出的处理方案:意识流写作:单段可跨数页——强制启用 Tier 2,并显式做句子边界检测;诗歌/韵文:换行是语义而非排版——将每个诗节(stanza)作为原子单元处理;含列表的纪实写作:项目符号会打断段落检测——预处理阶段先把 bullets 转为散文;多叙述者:章内文风随叙述者切换——检测叙述者标记,并优先在该处切分。七、管线集成:SegmentationAgent 的三阶段流程参考文档最后给出分段模块在整条 Book SFT 管线中的装配方式:class SegmentationAgent: def __init__(self, config: SegmentationConfig): self.tier1 Tier1Segmenter( min_wordsconfig.min_words, max_wordsconfig.max_words ) self.tier2 Tier2Segmenter(modelconfig.tier2_model) self.validator SegmentationValidator() async def segment(self, text: str) - list[Chunk]: # Phase 1: Tier 1 segmentation chunks self.tier1.segment(text) # Phase 2: Process oversized chunks with Tier 2 final_chunks [] for chunk in chunks: if chunk.requires_tier2: sub_chunks await self.tier2.segment(chunk) final_chunks.extend(sub_chunks) else: final_chunks.append(chunk) # Phase 3: Validate result self.validator.validate(final_chunks) if not result.valid: raise SegmentationError(result.errors) if result.warnings: logger.warning(fSegmentation warnings: {result.warnings}) return final_chunks三个阶段与 SKILL.md 中管线架构图里的 SEGMENTATION AGENT 一一对应:Tier 1 确定性地产出主体块 → 对requires_tier2True的块异步调用 LLM 二次切分 → 校验器拦截,errors 直接抛错终止,warnings 记录日志。这符合该示例技能所遵循的每个阶段可恢复、产出中间产物便于调试的 staged 架构原则(SKILL.md 的 project-development 映射一节)。八、结合仓库源码:实际落地参数与真实案例佐证8.1 仓库中的概念实现:150–400 词 段落重叠参考文档中的 Tier 1 默认参数是 250/650,而本仓库实际采用的训练管线参数更小,并加入了段落重叠(overlap)。pipeline_example.py 中的segment_text是这段策略的概念实现:def segment_text(text: str, min_words: int 150, max_words: int 400) - List[Chunk]: Segment text into training-sized chunks with overlap. Key insight: Smaller chunks (150-400) produce more examples and better style transfer than larger chunks (250-650). paragraphs [p.strip() for p in text.split(\n\n) if p.strip()] chunks [] buffer [] buffer_words 0 for para in paragraphs: para_words len(para.split()) if buffer_words para_words max_words and buffer_words min_words: chunks.append(Chunk( text\n\n.join(buffer), word_countbuffer_words, idlen(chunks) )) # Keep last paragraph for overlap buffer [buffer[-1], para] if buffer else [para] buffer_words len(buffer[-2].split()) para_words if len(buffer) 1 else para_words else: buffer.append(para) buffer_words para_words if buffer and buffer_words min_words // 2: chunks.append(Chunk(text\n\n.join(buffer), word_countbuffer_words, idlen(chunks))) return chunks与参考文档 Tier 1 的差异点:参数收紧为 150/400 词:docstring 明确写明Small chunks (150-400) produce more examples and better style transfer than larger chunks (250-650),这是案例实验后沉淀出的结论;段落重叠:封块后把上一块的最后一段带入新缓冲区(buffer [buffer[-1], para]),让相邻块共享上下文,缓解块间语义断裂;尾块放宽:最后一个块只要达到min_words // 2(75 词)即可输出,不再强求 150 词下限;Chunk 数据结构精简:概念实现里的Chunk只有text/word_count/id三个字段(pipeline_example.py),不含参考文档中的requires_tier2标记位——可以推断仓库脚本是简化版,Tier 2 路径在实际运行中按需引入。SKILL.md 的 Phase 2: Intelligent Segmentation 一节收录了同一算法的更简版本,并在 Expected Results 中给出了量化预期:对一本 86,000 词的书,旧方法(250–650 词)约得 150 块,新方法(150–400 词 重叠)约得 300 块,再按每块 2 个变体展开即 600 训练样本。8.2 Gertrude Stein 案例:分段策略的真实产出examples/book-sft-pipeline/examples/gertrude-stein/下保存了完整的案例数据,可用作分段参数的实证参照:training_config.json 记录了精确的生成配置:min_words150、max_words400、overlaptrue、variants_per_chunk2、指令生成模型gemini-2.0-flash-lite,最终total_chunks296、total_examples592、test_set_size50;dataset_sample.jsonl 展示了分段结果如何进入数据集:每行是标准messages三元组,assistant 侧即书籍原文分块(如 Mrs. Haydon was a good woman. She was a very good woman...),user 侧是不引用原文的场景化合成指令;gertrude-stein/README.md 给出端到端结果:86,000 词原著 → 592 训练样本,Qwen3-8B-Base LoRA(rank 32,学习率 5e-4,3 epochs,batch 4)训练约 15 分钟,test loss 从 7584.85 降至 213.36(约 97% 降幅),适配器 352 MB,总成本约 $2。8.3 下游衔接:分块如何进入训练分段产出的块经由多样指令生成扩展为训练样本后,进入 Tinker 的 Datum 结构。pipeline_example.py 的build_tinker_datum展示了 system/user 权重置 0、assistant 权重置 1 的 next-token 对齐;token 权重分配规则与 JSONL→Datum 的完整转换在 tinker-format.md 中有逐 token 的示例。分块阶段的 200/700 词校验边界、150–400 词目标区间,与序列保持 1000 token 以内的训练侧约束共同构成了从切分到训练的完整预算链。九、实操要点与适用前提技能加载方式:该示例不是独立发布的插件,而是目录式 Agent Skill。若要在 Claude Code / Cursor / Codex 中使用,按 book-sft-pipeline/README.md 的做法把整个目录拷贝到技能根目录,例如cp -R examples/book-sft-pipeline .claude/skills/book-sft-pipeline(仓库为只读,拷贝动作在你的目标工作区执行)。参数起点建议:从仓库实际案例出发,以min_words150 / max_words400 段落重叠作为起点;若你的语料段落极长(意识流等),叠加 Tier 2 的零删除 LLM 切分;若语体对话密集,优先叠加 Dialogue-Aware 逻辑;若书稿分隔符规范,场景感知几乎零成本即可启用。校验不可省:无论选哪条策略,SegmentationValidator的句首/句尾完整性 词数硬上限应作为入库前的最后闸门,errors 阻断、warnings 留痕。适用前提与边界:参考文档中的Tier1Segmenter/Tier2Segmenter/SceneAwareSegmenter等类是设计参考,ChunkBuilder、SegmentationError、_segment_at_scenes等支撑定义未随文档给出;仓库内可直接对照运行的是 pipeline_example.py(概念实现,需自行替换 LLM 调用与训练平台接口)。参数取值(250/650 vs 150/400)分别对应通用设计与单书风格迁移两类目标,移植到非小说语料(技术文档、对话日志)时应重新校准,并以校验器输出和训练 loss 曲线作为调节依据。十、小结Book SFT 流水线的分段环节本质是训练数据的信息密度工程:segmentation-strategies.md 给出的两层切分、场景感知、对话感知与校验管线构成了一套完整的策略工具箱;而 pipeline_example.py 与 gertrude-stein 案例 则用 296 块 / 592 样本 / 97% loss 降幅的真实数据证明了更小、语义完整、带重叠的分块配合多样指令,能以约 $2 的成本让 8B 基座模型习得可迁移到现代场景的作者文风。理解这套分段逻辑,也就理解了风格迁移数据质量的上限在哪里。【免费下载链接】Agent-Skills-for-Context-EngineeringA comprehensive collection of Agent Skills for context engineering, multi-agent architectures, and production agent systems. Use when building, optimizing, or debugging agent systems that require effective context management.项目地址: https://gitcode.com/GitHub_Trending/ag/Agent-Skills-for-Context-Engineering创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Agent-Skills-for-Context-Engineering:Book SFT 流水线的文本分块策略详解——两层切分、场景感知分段与校验管线
Agent-Skills-for-Context-Engineering:Book SFT 流水线的文本分块策略详解——两层切分、场景感知分段与校验管线【免费下载链接】Agent-Skills-for-Context-EngineeringA comprehensive collection of Agent Skills for context engineering, multi-agent architectures, and production agent systems. Use when building, optimizing, or debugging agent systems that require effective context management.项目地址: https://gitcode.com/GitHub_Trending/ag/Agent-Skills-for-Context-Engineering本篇技术文章以 segmentation-strategies.md 为核心,系统讲解图书转 SFT 训练数据流水线中最关键的分段(Segmentation)环节:从段落累积的两层切分策略、LLM 辅助切分的零删除校验,到场景感知、对话感知、结果校验与管线集成。结合 book-sft-pipeline 示例技能中的概念实现源码与 Gertrude Stein 真实案例数据,你将掌握如何在 150–400 词的语义完整片段上构建风格迁移数据集,并理解分块参数选择对训练样本数量与风格还原效果的实际影响。一、分块问题:为什么书籍文本难以直接切分长篇小说是训练数据构造中一类特殊的长文本。参考文档开篇即指出,书籍文本存在四个使按字数均分失效的特征:段落长度极不均匀:部分作者会写出超过 1000 词的连续单段;对话密集段落:单个对话回合往往过短,独立成块后语义残缺;场景边界与字数不对齐:自然断点(章节切换、场景转换)不一定落在目标字数附近;文风在叙述、对话、铺陈之间切换:不同文体段落对完整语义单元的定义不同。分块质量会直接映射到模型行为上——糟糕的分段等于在训练数据里示范了坏的生成模式,模型将学会输出:未完成的思路(Incomplete thoughts)戛然而止的结尾(Abrupt endings)不连贯的转场(Incoherent transitions)碎片化的文风(Fragmented style)这也是 SKILL.md 中Intelligent Segmentation被列为 Book SFT 三大支柱之首的原因:Breaking mid-sentence teaches the model to produce fragmented output.二、两层切分策略(Two-Tier Strategy)参考文档给出的核心方案是快慢结合的两层架构:第一层用确定性算法处理绝大多数段落,第二层用 LLM 兜底处理超大段落。SKILL.md 在Integration with Context Engineering Skills一节也明确印证了这一映射:Tier 1 对应快速、确定性的压缩,Tier 2 对应LLM 辅助处理边缘情况。2.1 Tier 1:基于段落累积的默认策略Tier 1 是面向结构良好文本的默认方法。参考文档给出的实现如下(类定义与核心循环完整保留自 segmentation-strategies.md):class Tier1Segmenter: def __init__(self, min_words: int 250, max_words: int 650): self.min_words min_words self.max_words max_words def segment(self, text: str) - list[Chunk]: paragraphs self._split_paragraphs(text) chunks [] current ChunkBuilder() for para in paragraphs: word_count len(para.split()) # Check if single paragraph exceeds max if word_count self.max_words: # Finalize current chunk if exists if current.word_count 0: chunks.append(current.build()) current ChunkBuilder() # Mark for Tier 2 processing chunks.append(Chunk( textpara, requires_tier2True, word_countword_count )) continue # Would this paragraph overflow current chunk? if current.word_count word_count self.max_words: if current.word_count self.min_words: chunks.append(current.build()) current ChunkBuilder() current.add(para) # Dont forget the last chunk if current.word_count 0: chunks.append(current.build()) return chunks def _split_paragraphs(self, text: str) - list[str]: # Split on double newlines, preserve single newlines within paragraphs text.split(\n\n) return [p.strip() for p in paragraphs if p.strip()]关键参数与逻辑要点:参数默认值作用min_words250块的最小词数下限;累积不足此值时允许继续追加,避免产生过碎片max_words650块的最大词数上限;超过即触发封块算法行为可以拆成三条路径:单段超长路径:若某段本身超过max_words,先封掉当前累积块,再将该段整体放入结果并打上requires_tier2True标记,交由第二层处理——注意此处不破坏段落,而是整体让渡给 LLM;溢出封块路径:新段落加入会超过max_words时,仅当当前块已满足min_words才封块重置;若当前块还很小,则继续追加(意味着块可能短暂越过max_words,直到满足最小词数后才允许切出,这是下限优先的设计权衡);收尾路径:循环结束后,缓冲区非空则作为最后一个块输出。段落切分本身采用最朴素但稳健的规则:按双换行\n\n断段,保留段内单换行,丢弃空白段。这与 pipeline_example.py 概念实现中的做法一致(text.split(\n\n)后strip()过滤)。2.2 Tier 2:LLM 辅助切分对第一层整体让渡的超大段落,使用 LLM 在语法自然处二次切分。参考文档的实现完整如下:class Tier2Segmenter: def __init__(self, model: str gpt-4o): self.model model self.prompt_template self._load_prompt() async def segment(self, oversized_chunk: Chunk) - list[Chunk]: Split an oversized paragraph using LLM. response await self._call_llm( self.prompt_template.format(textoversized_chunk.text) ) segments self._parse_segments(response) # Validate zero-deletion original_words len(oversized_chunk.text.split()) segmented_words sum(len(s.split()) for s in segments) if abs(original_words - segmented_words) 5: # Allow tiny variance raise SegmentationError( fWord count mismatch: {original_words} - {segmented_words} ) return [ Chunk(texts, requires_tier2False, word_countlen(s.split())) for s in segments ] def _load_prompt(self) - str: return Segment this text into excerpts of minimum 300-350 words. Requirements: - Each excerpt must be grammatically complete from start - Each excerpt must not feel abruptly cut off - Zero deletion - maintain original word count exactly - Break at grammatically natural places: * After complete dialogue exchanges * At scene transitions * After complete thoughts or descriptions * Where a paragraph break would naturally occur - Avoid breaking into too many small excerpts - Start directly with the excerpts - Separate excerpts with SEGMENT Text to segment: {text} def _parse_segments(self, response: str) - list[str]: segments response.split(SEGMENT) return [s.strip() for s in segments if s.strip()]这段设计里有三个值得注意的工程决策:零删除校验(Zero-deletion check):切分前后总词数偏差超过 5 词即抛SegmentationError。这是防 LLM 改写/删字的硬约束——SFT 数据集要求 assistant 侧必须是原文逐字片段,任何删改都会污染风格样本。切分提示词约束:每段目标 300–350 词,且只允许在完整对话之后、场景转换处、完整思想之后、自然段落断点处切分,并要求避免切出过多小段、直接以片段开始输出。机器可解析的分隔符:用SEGMENT作为片段边界,解析时split后剔除空串,避免 LLM 输出前言污染片段内容。三、场景感知分段(Scene-Aware Segmentation)对追求更高切分质量的场景,参考文档建议优先利用书中已有的场景分隔标记,而不是任意段落边界:class SceneAwareSegmenter: Prefer breaking at scene boundaries when within word limits. SCENE_MARKERS [ r\n\n\* \* \*\n\n, # Asterisk dividers r\n\n---\n\n, # Dash dividers r\n\n###\n\n, # Hash dividers r\n\nCHAPTER \d, # Chapter headings r\n\n[A-Z]{3,}\n\n, # All-caps scene breaks ] def find_scene_breaks(self, text: str) - list[int]: Find character positions of scene breaks. breaks [] for pattern in self.SCENE_MARKERS: for match in re.finditer(pattern, text): breaks.append(match.start()) return sorted(set(breaks)) def segment_with_scenes(self, text: str) - list[Chunk]: scene_breaks self.find_scene_breaks(text) # If scene breaks exist, prefer them over arbitrary paragraph breaks if scene_breaks: return self._segment_at_scenes(text, scene_breaks) else: return Tier1Segmenter().segment(text)五类分隔标记覆盖了英语小说最常见的排版约定:星号分隔行、短横线分隔、井号分隔、CHAPTER n章节标题、全大写场景行。策略是有场景标记就用场景标记,没有则回退 Tier 1,属于典型的启发式优先、确定性兜底。需要说明:代码片段中使用了re.finditer但未展示import re,且_segment_at_scenes的具体实现留白,从源码结构看该类是设计参考(design reference)而非可直接运行的完整模块。四、对话感知分段(Dialogue-Aware Segmentation)对话密集段落单独处理,核心原则是绝不从一轮完整对话中间切开:class DialogueAwareSegmenter: Group dialogue exchanges to maintain conversation coherence. def is_dialogue_paragraph(self, para: str) - bool: Check if paragraph is primarily dialogue. # Count dialogue markers quote_count para.count() para.count() word_count len(para.split()) # If more than 20% of words are in quotes, its dialogue-heavy return quote_count word_count * 0.2 def segment(self, text: str) - list[Chunk]: paragraphs text.split(\n\n) chunks [] current ChunkBuilder() in_dialogue_block False for para in paragraphs: is_dialogue self.is_dialogue_paragraph(para) # Dont break in the middle of a dialogue exchange if is_dialogue: in_dialogue_block True current.add(para) else: if in_dialogue_block: # End of dialogue block - good break point in_dialogue_block False if current.word_count 250: chunks.append(current.build()) current ChunkBuilder() current.add(para) # Check if weve exceeded max if current.word_count 650: chunks.append(current.build()) current ChunkBuilder() if current.word_count 0: chunks.append(current.build()) return chunks从实现逻辑看有两个关键设计:对话块状态机:进入对话段落后置in_dialogue_block True并只累积不切分;当出现第一个非对话段落时,视其为对话块结束、理想断点,此时若当前块已达 250 词下限则封块;引号密度启发式:以引号字符数 词数 × 0.2判定对话段落。严格来说这是引号字符数与词数之比的粗略代理指标(一个引号字符对应两个词才成立 20% 的占比),从源码结构看它属于低成本启发式,胜在无需句法分析,适合批量预处理。五、校验管线:分段结果必须过检参考文档要求每一批分段结果都要通过校验,校验器同时产出 errors 与 warnings 两级信号:class SegmentationValidator: def validate(self, chunks: list[Chunk]) - ValidationResult: errors [] warnings [] for i, chunk in enumerate(chunks): # Check word count bounds if chunk.word_count 200: warnings.append(fChunk {i}: Only {chunk.word_count} words) if chunk.word_count 700: errors.append(fChunk {i}: {chunk.word_count} words exceeds max) # Check sentence completeness if not self._ends_with_terminal(chunk.text): errors.append(fChunk {i}: Ends mid-sentence) if not self._starts_grammatically(chunk.text): errors.append(fChunk {i}: Starts mid-sentence) # Check for orphaned dialogue if chunk.text.count() % 2 ! 0: warnings.append(fChunk {i}: Unbalanced quotes) return ValidationResult( validlen(errors) 0, errorserrors, warningswarnings ) def _ends_with_terminal(self, text: str) - bool: text text.strip() return text[-1] in .!?\— def _starts_grammatically(self, text: str) - bool: text text.strip() # Should start with capital or quote return text[0].isupper() or text[0] in \—五项校验规则及其严重级别:检查项规则级别词数下限 200词warning词数上限 700词error结尾完整末字符须为. ! ? —之一error开头完整首字符须为大写字母或 —之一error引号配平双引号数量为奇数warning注意校验边界(200/700)比分段边界(250/650)更宽,相当于给分段器留出一层容错缓冲:轻微越界只报警不阻断,严重越界(超出硬上限或句子不完整)才判定失败。这与 tinker-format.md 中序列长度应保持 1000 token 以内,更长的序列会稀释局部风格模式的约束互为补充——校验器的 700 词硬上限正是 token 预算的前置防线。六、策略选型与边界情况参考文档给出的策略对比表(完整继承):策略速度质量适用场景仅 Tier 1快中结构良好的散文Tier 1 Tier 2中高段落长短混合的文本场景感知快高场景分隔清晰的小说对话感知中高对话密集的小说四类典型边界情况及文档给出的处理方案:意识流写作:单段可跨数页——强制启用 Tier 2,并显式做句子边界检测;诗歌/韵文:换行是语义而非排版——将每个诗节(stanza)作为原子单元处理;含列表的纪实写作:项目符号会打断段落检测——预处理阶段先把 bullets 转为散文;多叙述者:章内文风随叙述者切换——检测叙述者标记,并优先在该处切分。七、管线集成:SegmentationAgent 的三阶段流程参考文档最后给出分段模块在整条 Book SFT 管线中的装配方式:class SegmentationAgent: def __init__(self, config: SegmentationConfig): self.tier1 Tier1Segmenter( min_wordsconfig.min_words, max_wordsconfig.max_words ) self.tier2 Tier2Segmenter(modelconfig.tier2_model) self.validator SegmentationValidator() async def segment(self, text: str) - list[Chunk]: # Phase 1: Tier 1 segmentation chunks self.tier1.segment(text) # Phase 2: Process oversized chunks with Tier 2 final_chunks [] for chunk in chunks: if chunk.requires_tier2: sub_chunks await self.tier2.segment(chunk) final_chunks.extend(sub_chunks) else: final_chunks.append(chunk) # Phase 3: Validate result self.validator.validate(final_chunks) if not result.valid: raise SegmentationError(result.errors) if result.warnings: logger.warning(fSegmentation warnings: {result.warnings}) return final_chunks三个阶段与 SKILL.md 中管线架构图里的 SEGMENTATION AGENT 一一对应:Tier 1 确定性地产出主体块 → 对requires_tier2True的块异步调用 LLM 二次切分 → 校验器拦截,errors 直接抛错终止,warnings 记录日志。这符合该示例技能所遵循的每个阶段可恢复、产出中间产物便于调试的 staged 架构原则(SKILL.md 的 project-development 映射一节)。八、结合仓库源码:实际落地参数与真实案例佐证8.1 仓库中的概念实现:150–400 词 段落重叠参考文档中的 Tier 1 默认参数是 250/650,而本仓库实际采用的训练管线参数更小,并加入了段落重叠(overlap)。pipeline_example.py 中的segment_text是这段策略的概念实现:def segment_text(text: str, min_words: int 150, max_words: int 400) - List[Chunk]: Segment text into training-sized chunks with overlap. Key insight: Smaller chunks (150-400) produce more examples and better style transfer than larger chunks (250-650). paragraphs [p.strip() for p in text.split(\n\n) if p.strip()] chunks [] buffer [] buffer_words 0 for para in paragraphs: para_words len(para.split()) if buffer_words para_words max_words and buffer_words min_words: chunks.append(Chunk( text\n\n.join(buffer), word_countbuffer_words, idlen(chunks) )) # Keep last paragraph for overlap buffer [buffer[-1], para] if buffer else [para] buffer_words len(buffer[-2].split()) para_words if len(buffer) 1 else para_words else: buffer.append(para) buffer_words para_words if buffer and buffer_words min_words // 2: chunks.append(Chunk(text\n\n.join(buffer), word_countbuffer_words, idlen(chunks))) return chunks与参考文档 Tier 1 的差异点:参数收紧为 150/400 词:docstring 明确写明Small chunks (150-400) produce more examples and better style transfer than larger chunks (250-650),这是案例实验后沉淀出的结论;段落重叠:封块后把上一块的最后一段带入新缓冲区(buffer [buffer[-1], para]),让相邻块共享上下文,缓解块间语义断裂;尾块放宽:最后一个块只要达到min_words // 2(75 词)即可输出,不再强求 150 词下限;Chunk 数据结构精简:概念实现里的Chunk只有text/word_count/id三个字段(pipeline_example.py),不含参考文档中的requires_tier2标记位——可以推断仓库脚本是简化版,Tier 2 路径在实际运行中按需引入。SKILL.md 的 Phase 2: Intelligent Segmentation 一节收录了同一算法的更简版本,并在 Expected Results 中给出了量化预期:对一本 86,000 词的书,旧方法(250–650 词)约得 150 块,新方法(150–400 词 重叠)约得 300 块,再按每块 2 个变体展开即 600 训练样本。8.2 Gertrude Stein 案例:分段策略的真实产出examples/book-sft-pipeline/examples/gertrude-stein/下保存了完整的案例数据,可用作分段参数的实证参照:training_config.json 记录了精确的生成配置:min_words150、max_words400、overlaptrue、variants_per_chunk2、指令生成模型gemini-2.0-flash-lite,最终total_chunks296、total_examples592、test_set_size50;dataset_sample.jsonl 展示了分段结果如何进入数据集:每行是标准messages三元组,assistant 侧即书籍原文分块(如 Mrs. Haydon was a good woman. She was a very good woman...),user 侧是不引用原文的场景化合成指令;gertrude-stein/README.md 给出端到端结果:86,000 词原著 → 592 训练样本,Qwen3-8B-Base LoRA(rank 32,学习率 5e-4,3 epochs,batch 4)训练约 15 分钟,test loss 从 7584.85 降至 213.36(约 97% 降幅),适配器 352 MB,总成本约 $2。8.3 下游衔接:分块如何进入训练分段产出的块经由多样指令生成扩展为训练样本后,进入 Tinker 的 Datum 结构。pipeline_example.py 的build_tinker_datum展示了 system/user 权重置 0、assistant 权重置 1 的 next-token 对齐;token 权重分配规则与 JSONL→Datum 的完整转换在 tinker-format.md 中有逐 token 的示例。分块阶段的 200/700 词校验边界、150–400 词目标区间,与序列保持 1000 token 以内的训练侧约束共同构成了从切分到训练的完整预算链。九、实操要点与适用前提技能加载方式:该示例不是独立发布的插件,而是目录式 Agent Skill。若要在 Claude Code / Cursor / Codex 中使用,按 book-sft-pipeline/README.md 的做法把整个目录拷贝到技能根目录,例如cp -R examples/book-sft-pipeline .claude/skills/book-sft-pipeline(仓库为只读,拷贝动作在你的目标工作区执行)。参数起点建议:从仓库实际案例出发,以min_words150 / max_words400 段落重叠作为起点;若你的语料段落极长(意识流等),叠加 Tier 2 的零删除 LLM 切分;若语体对话密集,优先叠加 Dialogue-Aware 逻辑;若书稿分隔符规范,场景感知几乎零成本即可启用。校验不可省:无论选哪条策略,SegmentationValidator的句首/句尾完整性 词数硬上限应作为入库前的最后闸门,errors 阻断、warnings 留痕。适用前提与边界:参考文档中的Tier1Segmenter/Tier2Segmenter/SceneAwareSegmenter等类是设计参考,ChunkBuilder、SegmentationError、_segment_at_scenes等支撑定义未随文档给出;仓库内可直接对照运行的是 pipeline_example.py(概念实现,需自行替换 LLM 调用与训练平台接口)。参数取值(250/650 vs 150/400)分别对应通用设计与单书风格迁移两类目标,移植到非小说语料(技术文档、对话日志)时应重新校准,并以校验器输出和训练 loss 曲线作为调节依据。十、小结Book SFT 流水线的分段环节本质是训练数据的信息密度工程:segmentation-strategies.md 给出的两层切分、场景感知、对话感知与校验管线构成了一套完整的策略工具箱;而 pipeline_example.py 与 gertrude-stein 案例 则用 296 块 / 592 样本 / 97% loss 降幅的真实数据证明了更小、语义完整、带重叠的分块配合多样指令,能以约 $2 的成本让 8B 基座模型习得可迁移到现代场景的作者文风。理解这套分段逻辑,也就理解了风格迁移数据质量的上限在哪里。【免费下载链接】Agent-Skills-for-Context-EngineeringA comprehensive collection of Agent Skills for context engineering, multi-agent architectures, and production agent systems. Use when building, optimizing, or debugging agent systems that require effective context management.项目地址: https://gitcode.com/GitHub_Trending/ag/Agent-Skills-for-Context-Engineering创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考