在检索增强生成（RAG）系统的建设中，文档分块（Chunking）是连接原始文档与向量数据库之间的关键纽带。本文将从零开始拆解“知识库文档开始分块接口”的技术方案、RESTful 协议规范、四大切片策略的参数化抽象、异步解耦架构，并提供基于 Python FastAPI 的生产级完整代码实现。

一、 为什么“开始分块”需要独立的 API？

在早期的 Demo 级 RAG 系统中，开发者常将文档上传、文本提取、切分 Chunk 和向量化（Embedding）写在一个同步 HTTP 请求中。然而，在企业级知识库场景中，这种做法会带来严重的生产事故：

• 连接超时（HTTP Timeout）：一份 200 页的 PDF 手册，包含 OCR 识别、语义切片和向量生成，耗时可达数秒甚至数十秒，极易导致前端网关（如 Nginx）抛出 504 错误。

• 策略不可控：不同类型的文档（如 API 研发文档、财务报表、法律合同）需要完全不同的切片参数（如重叠度 Overlap、分隔符 Delimiter、父子块比例）。

• 缺乏版本与审查机制：分块完成后，业务人员通常需要在线预览切片效果，甚至进行人工二次编辑（Human-in-the-loop），才能触发最终的向量落库。

因此，“提交分块任务（Trigger Chunking API）”必须作为一个独立的、异步解耦的 RESTful 接口存在。

[前端/应用方] ─── POST /documents/{id}/chunk ───> [RAG 网关 API] │ (写入状态: PROCESSING) │ ▼ [MQ / Celery 异步队列] │ ▼ [文档切片 & 向量化 Worker]

二、 接口协议与 RESTful 参数设计

定义一个通用、扩展性强的开始分块接口，需要全面覆盖主流切片模式（通用固定切片、递归切片、语义切片、父子切片）所需的参数。

1. 接口基本信息

• 接口路径：POST /api/v1/knowledge/documents/{document_id}/chunk

• 请求头：Content-Type: application/json

• 鉴权方式：Bearer <JWT_TOKEN>

2. 请求体（Request Body）参数抽象

接口参数分为三大模块：切片模式设置、文本清洗选项和高级解析策略。

3. 请求示例（JSON）

{ "chunk_strategy": "parent_child", "max_chunk_size": 500, "overlap_size": 50, "delimiters": ["\n\n", "\n", "；", "。"], "clean_extra_spaces": true, "remove_urls_emails": false, "parent_chunk_size": 1500, "child_chunk_size": 300, "auto_vectorize": true }

4. 响应示例（Response Body）

接口采用异步响应模式，提交成功后立即返回202 Accepted以及用于追踪进展的task_id。

{ "code": 200, "message": "文档分块任务提交成功，正在后台异步处理", "data": { "task_id": "task_chunk_9527_abcd1234", "document_id": "doc_8848_xyz", "status": "PROCESSING", "created_at": "2026-08-08T19:30:00Z", "strategy_snapshot": { "chunk_strategy": "parent_child", "parent_chunk_size": 1500, "child_chunk_size": 300 } } }

三、 四种切片策略在接口背后的实现逻辑

在实现 API 核心引擎时，后端需根据chunk_strategy路由到不同的算法执行模块：

1. 通用固定切片（General / Fixed-Size Chunking）

• 原理：硬性按照固定字符数/Token数对文本进行分割。

• 适用场景：格式不规则的日志、缺乏标点符号的非结构化数据。

• 核心注意：必须使用overlap_size避免切片边界处的语义断裂。

2. 递归字符切片（Recursive Character Chunking）

• 原理：依据delimiters列表中定义的分割符优先级（如["\n\n", "\n", "。", " "]）递归尝试切分。若段落太大则下探到句号，句号太长则下探到空格。

• 适用场景： Markdown、Markdown 格式的技术文档、结构清晰的文章（RAG 首选默认策略）。

3. 语义相似度切片（Semantic Chunking）

• 原理：使用轻量级句子向量模型计算连续句子之间的语义相似度。当相邻句子的余弦相似度低于设定阈值时，自动插入断点。

• 适用场景：小说、自由对话、无明显段落结构的富文本。

4. 父子分块模式（Parent-Child / Small-to-Big Chunking）

• 原理：将文档切分为较大的父块（Parent Chunk）与较小的子块（Child Chunk）。向量数据库中仅索引子块向量（检索精度高），但在召回送给 LLM 时，自动映射并读取其所属的父块（上下文完整）。

四、 生产级异步任务架构设计

因为文档解析与分块属于 CPU 密集型/耗时任务，必须通过生产者-消费者架构解耦。

前端页面 ──(发起分块请求)──> API 网关 (FastAPI) │ (持久化任务状态为 PROCESSING) │ (推入 Redis/RabbitMQ 队列) │ ▼ Celery Worker 进程池 │ ┌──────────────────┼──────────────────┐ ▼ ▼ ▼ [文档提取模块] [文本切片引擎] [存储数据库 / Vector DB]

为了让客户端能够轮询分块进度，还需要提供一个任务状态查询接口：

• 接口路径：GET /api/v1/knowledge/chunk-tasks/{task_id}

• 响应内容：包含当前处理状态（PENDING,PROCESSING,SUCCESS,FAILED）、进度百分比、已生成的 Chunk 数量以及报错堆栈信息。

五、 手把手代码实现：基于 FastAPI + Celery 的分块 API

下面提供一份工业级 Python 代码实现，展示如何构建该接口及后台切片引擎。

1. 数据模型与 Request 定义（schemas.py）

from enum import Enum from typing import List, Optional from pydantic import BaseModel, Field class ChunkStrategyEnum(str, Enum): GENERAL = "general" RECURSIVE = "recursive" SEMANTIC = "semantic" PARENT_CHILD = "parent_child" class StartChunkingRequest(BaseModel): chunk_strategy: ChunkStrategyEnum = Field( default=ChunkStrategyEnum.RECURSIVE, description="分块策略选择" ) max_chunk_size: int = Field(default=500, ge=50, le=4000, description="单块最大字符数") overlap_size: int = Field(default=50, ge=0, le=500, description="块间重叠字符数") delimiters: Optional[List[str]] = Field( default=["\n\n", "\n", "。", "！", "？", " "], description="递归切片分隔符优先级" ) clean_extra_spaces: bool = Field(default=True, description="清洗多余空格换行") remove_urls_emails: bool = Field(default=False, description="移除 URL 与邮箱") # 父子模式专有参数 parent_chunk_size: Optional[int] = Field(default=1500, description="父块字符数") child_chunk_size: Optional[int] = Field(default=300, description="子块字符数") auto_vectorize: bool = Field(default=True, description="切片后自动向量化") class ChunkingTaskResponse(BaseModel): code: int = 200 message: str task_id: str document_id: str status: str

2. FastAPI 路由控制器（router.py）

import uuid from fastapi import APIRouter, HTTPException, BackgroundTasks, status from schemas import StartChunkingRequest, ChunkingTaskResponse router = APIRouter(prefix="/api/v1/knowledge", tags=["Knowledge Base Chunking"]) # 模拟数据库或 Redis 中的任务状态存储 TASK_DB = {} DOCUMENT_DB = { "doc_001": { "title": "大模型 RAG 架构设计规范.pdf", "content": "检索增强生成（RAG）技术正在改变企业知识库...（此处省略一万字原始文本）...", "status": "UPLOADED" } } def mock_async_chunk_worker(task_id: str, doc_id: str, params: StartChunkingRequest): """ 后台异步 Worker 执行体（生产环境建议换为 Celery Task） """ try: TASK_DB[task_id]["status"] = "PROCESSING" raw_text = DOCUMENT_DB[doc_id]["content"] # 1. 文本清洗 if params.clean_extra_spaces: raw_text = " ".join(raw_text.split()) # 2. 根据策略进行切片 chunks = [] if params.chunk_strategy == "recursive": # 简易递归切分示意 step = params.max_chunk_size - params.overlap_size for i in range(0, len(raw_text), step): chunks.append(raw_text[i : i + params.max_chunk_size]) elif params.chunk_strategy == "parent_child": # 生成父子切片逻辑... pass # 3. 保存切片结果至数据库 TASK_DB[task_id]["status"] = "SUCCESS" TASK_DB[task_id]["chunks_count"] = len(chunks) TASK_DB[task_id]["result_chunks"] = chunks DOCUMENT_DB[doc_id]["status"] = "CHUNKED" except Exception as e: TASK_DB[task_id]["status"] = "FAILED" TASK_DB[task_id]["error_msg"] = str(e) @router.post( "/documents/{document_id}/chunk", response_model=ChunkingTaskResponse, status_code=status.HTTP_202_ACCEPTED ) async def start_document_chunking( document_id: str, request_data: StartChunkingRequest, background_tasks: BackgroundTasks ): # 校验文档是否存在 if document_id not in DOCUMENT_DB: raise HTTPException(status_code=404, detail="未找到目标文档") doc = DOCUMENT_DB[document_id] if doc["status"] == "PROCESSING": raise HTTPException(status_code=400, detail="该文档正在处理中，请勿重复发起") # 生成全局唯一任务 ID task_id = f"task_chunk_{uuid.uuid4().hex[:8]}" # 记录任务初始状态 TASK_DB[task_id] = { "document_id": document_id, "status": "PENDING", "params": request_data.model_dump() } # 将长耗时任务推入后台队列 background_tasks.add_task(mock_async_chunk_worker, task_id, document_id, request_data) return ChunkingTaskResponse( code=200, message="文档分块任务已成功提交", task_id=task_id, document_id=document_id, status="PENDING" )

六、 配套的切片校对与查看 API

在完整的企业级知识库系统中，仅仅提交分块是不够的，还需要配套提供切片列表查询与人工修整接口：

1. 查询已切片结果列表 API

• 路径：GET /api/v1/knowledge/documents/{document_id}/chunks

• 用途：前端将分块结果以卡片或列表形式展示给用户，展示索引号、字符数、预览文本与所属父块编号。

2. 修改单条切片 API

• 路径：PUT /api/v1/knowledge/chunks/{chunk_id}

• 用途：若大模型切分切断了关键公式或表格，业务人员可在线修改文本内容并手动保存，再触发向量落库。

七、 生产落地避坑指南

1. 接口幂等性保障：如果对已经分块且已向量化的文档再次调用分块接口，系统必须能够自动作废旧的 Chunk 记录及向量数据库中的高维 Vector，防止重复召回历史垃圾数据。

2. Markdown / PDF 表格保护：普通的字符切分极易把 Markdown 表格（| header | header |）切成两截。在分块引擎内部，建议先利用正则提取表格并将其转换为 HTML 或 JSON 字符串整体作为一个不可分割的 Chunk 保护起来。

3. 内存 OOM 防范：遇到数百兆级别的超大文本文件（如日志文件或超长合同汇编）时，切忌将整个文件一次性read()载入内存，应采用流式读取（Stream Chunking）模式分批写入存储。

通过设计严谨的 RESTful 异步分块 API，能够将前端交互、复杂文档解析与下游向量数据库建立起清晰的架构解耦，为大模型 RAG 系统构建稳健的高质量知识基础设施。