阿里百炼多模态文档拆分实战
一、问题的来源
给湖北理工计算机学院官网助手做 RAG，教务处上传的资料很多格式都有。一最开始想开始偷懒用纯文本拆分器——就是那种按字符数或段落切的通用 splitter。
出问题的例子一抓一把：
PPT 课件拆分结果:
"计算机学院课程体系数据结构与算法CS201程序设计基础CS101..."
所有文字挤成一团，标题、正文、图表、注释之间的边界全没了
Word 实验指导书里的架构图:
文字提取只拿到 "图3-1 系统架构图"（图注），图的内容丢了
PDF 扫描版规章制度:
文字提取结果: ""（空的，一个字没有）
Markdown 笔记里的 :
拆出来是 "[图片: arch]"——只有占位符
本质原因是：文本拆分器只能看到字符流。它不知道页面上哪里有图、表格的行列关系是什么、PPT 的标题和正文在哪里。你给它一张纯图片 PDF，它连一个字都看不到。
为了解决这个问题就要用到阿里百炼：
- 百炼文档解析 API：把 PDF/Word/PPT/Markdown 上传，自动抽出文字、表格、嵌入图片。返回按页组织的结构化 JSON
- 千问 VL（qwen-vl-max）：多模态大模型，能看图说话。把一页文档的截图传给它，它输出 "这页有一个架构图，展示了 API Gateway → Service A → Database 的调用链"
用这两个 API 配合的思路：文档解析把文字骨架抽出来 → 千问 VL 把图表的语义补上 → 拼成完整表示 → 再做语义拆分。这样架构图、表格、扫描件上的文字都不会丢。
二、先把百炼接入
2.1 开账号
打开 bailian.console.aliyun.com，登录阿里云账号：
- 点「模型广场」→ 搜
qwen-vl-max→ 点开通 - 点「应用中心」→ 新建应用 → 选"文档解析"模板 → 创建完拿到一个
app_id（纯数字） - 右上角头像 →「API-KEY 管理」→ 创建一个 key（形如
sk-xxxx）
2.2 初始化代码
百炼的千问 VL 兼容 OpenAI SDK。这意味着用 openai 包就能调，只改 base_url 和 api_key：
import os, json, base64, time, re
from pathlib import Path
from typing import Optional
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from openai import OpenAI
DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY") # 百炼 API Key
BAILIAN_APP_ID = os.getenv("BAILIAN_APP_ID") # 文档解析应用 ID
# 和调 GPT-4V 几乎一样的代码，就换了个 base_url
vl_client = OpenAI(
api_key=DASHSCOPE_API_KEY,
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
为什么有两个端点？ 百炼提供了两种调千问 VL 的方式。一种是原生 DashScope API（dashscope.aliyuncs.com/api/v1），需要按阿里云自己的 SDK 方式调用。另一种是 OpenAI 兼容端点（dashscope.aliyuncs.com/compatible-mode/v1）——URL 里带 compatible-mode，完全模拟 OpenAI 的 API 格式。选兼容端点是因为现有的 ChatOpenAI、OpenAI() 代码不用改。
三、核心结构：PageUnit
四种格式结构完全不同——PDF 有页码、PPT 有幻灯片编号、Word 有章节、Markdown 有标题。如果每种格式都写一套下游处理逻辑，维护成本太高。
我搞了一个统一的数据结构，所有格式先转成这个，后续处理不用管来源：
@dataclass
class PageUnit:
page_index: int
# 页序号，从 0 开始。对 Markdown 来说，这里存的是按 ## 标题拆分后的段落块序号
raw_text: str
# 文档解析 API 抽出来的原始文字。
# 这是"骨架"——纯文字页靠它就够了，含图表的页还需要千问 VL 补视觉描述
tables_as_markdown: list[str]
# 页面内每个表格的 Markdown 字符串。
# 为什么用 Markdown 存而不是 JSON？
# embedding 模型在训练时见过大量 Markdown 表格。它能理解 | 管道符的语义是列边界，
# |---| 下面是表头，后面的是数据行。JSON 数组对 embedding 模型来说更陌生——
# 同样的数据，Markdown 表格比 JSON 做 embedding 检索命中率高不少
page_image_base64: str
# 这页渲染成图片后的 base64 编码。传给千问 VL 做视觉描述用。
# 空字符串表示这页不需要视觉描述（比如纯文字页）
has_visual_content: bool
# True 表示这页需要送千问 VL。判断条件：
# - 文字<100字 → 可能是图片页或扫描件
# - 百炼检测到了嵌入图片 → 需要 VL 描述图的内容
# - PPT 每页都标记为 True → 幻灯片排版本身就有信息量
source_metadata: dict = field(default_factory=dict)
# 存的格式相关元信息，每种格式填不同的东西：
# PDF: {"source": "考试管理办法.pdf", "page": 3, "total_pages": 20}
# PPT: {"source": "课件.pptx", "slide": 5}
# Word: {"source": "实验指导.docx", "section": "第3章", "outline_level": 2}
# Markdown: {"source": "笔记.md", "heading": "## 数据结构", "images": [...]}
这个结构的核心价值是让下游不用关心来源。无论原始文件是什么格式，拿到 PageUnit 列表后，遍历 → 判断 has_visual_content → 决定要不要调 VL → 调统一的拆分函数。四套格式共用一套下游逻辑。
四、百炼文档解析 API 的封装
百炼的文档解析是异步的——上传文件、提交任务、然后轮询等结果。三个步骤对应三个函数。我把它们封装好：
import requests
BAILIAN_API_BASE = "https://bailian.aliyuncs.com"
def _upload_to_bailian(file_path: str) -> str:
"""上传文件到百炼，返回 file_id。
file_id 是一串随机字符（如 "file_abc123"），在百炼服务端保存 24 小时。
上传方式就是 HTTP multipart/form-data——post 一个文件过去就行。"""
with open(file_path, "rb") as f:
resp = requests.post(
f"{BAILIAN_API_BASE}/api/v1/files",
headers={"Authorization": f"Bearer {DASHSCOPE_API_KEY}"},
# 下面这个 files 参数，requests 库会自动设 Content-Type 为 multipart/form-data
files={"file": (Path(file_path).name, f)},
timeout=60,
)
resp.raise_for_status()
return resp.json()["id"]
def _submit_parse_job(file_id: str, file_type: str) -> str:
"""提交解析任务，返回 task_id。
file_type 告诉百炼用什么策略解析：
- "pptx": 每页幻灯片渲染成图片 + 提取文字
- "docx": 识别标题层级 + 提取文字 + 定位嵌入图片
- "pdf": 分页 + 判断是否扫描件 + 提取文字
- "md": 按标题拆段落 + 提取表格
result_type="structured" 表示要结构化输出——按页组织、表格单独分离。
如果设 "text" 只返回纯文本，不推荐：页码边界会丢、表格变成乱序文字。"""
resp = requests.post(
f"{BAILIAN_API_BASE}/api/v1/apps/{BAILIAN_APP_ID}/documents",
headers={
"Authorization": f"Bearer {DASHSCOPE_API_KEY}",
"Content-Type": "application/json",
},
json={
"file_id": file_id,
"file_type": file_type,
"result_type": "structured",
},
timeout=30,
)
resp.raise_for_status()
return resp.json()["task_id"]
def _wait_for_result(task_id: str, poll_interval: int = 2, max_wait: int = 120) -> dict:
"""轮询直到解析完成，返回解析结果。
poll_interval=2: 每 2 秒查一次。太频繁可能触发 API 限流
max_wait=120: 超时保护。一份 30 页 PPTX 通常 5-10 秒完成，2 分钟够用"""
deadline = time.time() + max_wait
while time.time() < deadline:
resp = requests.get(
f"{BAILIAN_API_BASE}/api/v1/tasks/{task_id}",
headers={"Authorization": f"Bearer {DASHSCOPE_API_KEY}"},
timeout=10,
)
resp.raise_for_status()
data = resp.json()
if data["status"] == "completed":
return data["result"]
if data["status"] == "failed":
raise RuntimeError(f"解析失败: {data.get('error')}")
time.sleep(poll_interval)
raise TimeoutError(f"解析超时 ({max_wait}s)")
三个函数串联起来的调用链很直观：
本地文件 (PPTX/PDF/DOCX/MD)
│
▼ _upload_to_bailian(file_path)
file_id（"file_abc123"）
│
▼ _submit_parse_job(file_id, "pptx")
task_id（"task_xyz789"）
│
▼ _wait_for_result(task_id) ← 每 2 秒查一次，直到完成
解析结果 JSON:
{
"pages": [
{
"index": 0,
"text": "这页抽出来的文字...",
"tables": [{"headers": ["列1","列2"], "rows": [["a","b"]]}],
"images": [{"url": "https://..."}],
"image_url": "https://.../page_0.png" // PPT/PDF 才有——百炼渲染的截图
},
...
]
}
两个工具函数——下载页面截图和表格转 Markdown：
def _download_page_image(image_url: str) -> str:
"""下载百炼渲染好的页面截图，转 base64。整个过程不落盘。"""
resp = requests.get(image_url, timeout=30)
resp.raise_for_status()
return base64.b64encode(resp.content).decode()
def _table_to_markdown(table_data: dict) -> str:
"""百炼返回的表格 JSON → Markdown 格式。
百炼返回: {"headers": ["课程", "学分"], "rows": [["CS201", "4"]]}
输出: | 课程 | 学分 |
|------|------|
| CS201 | 4 |
为什么是 Markdown？embedding 模型（text-embedding-v3 等）预训练数据里
有海量 Markdown 表格——模型知道 | 是列边界、---|---| 是表头分隔线、
下面是数据行。同样的内容，Markdown 表格的检索命中率比 JSON 高。"""
headers = table_data.get("headers", [])
rows = table_data.get("rows", [])
if not headers:
return ""
lines = ["| " + " | ".join(str(h) for h in headers) + " |"]
lines.append("| " + " | ".join("---" for _ in headers) + " |")
for row in rows:
lines.append("| " + " | ".join(str(c) for c in row) + " |")
return "\n".join(lines)
这一节是整条管线的基础——后面四种格式的处理全部依赖这三个函数的组合。
五、PPT 拆分
PPT 是四种格式里最依赖视觉的一个。一页幻灯片里的标题在顶部、正文在左侧、图表在右侧——这些空间关系在 PPTX 源文件的 XML 里不直接存储，文字提取只能按 XML 元素的出现顺序把字捞出来，结果是一锅粥。
架构图和流程图更惨——它们根本不走文字层，是矢量图形对象。文字提取直接跳过。
PPT 的策略：每页幻灯片都渲染成图 + 送千问 VL + VL 输出转成 chunk。
5.1 pptx → PageUnit 列表
def pptx_to_page_units(pptx_path: str) -> list[PageUnit]:
file_id = _upload_to_bailian(pptx_path)
task_id = _submit_parse_job(file_id, file_type="pptx")
parse_result = _wait_for_result(task_id)
units = []
for page_data in parse_result["pages"]:
text = page_data.get("text", "")
# 这页的所有表格转 Markdown
tables_md = []
for t in page_data.get("tables", []):
tables_md.append(_table_to_markdown(t))
# 百炼渲染的幻灯片截图——每页 PPT 都有
page_img = _download_page_image(page_data["image_url"])
units.append(PageUnit(
page_index=page_data["index"],
raw_text=text,
tables_as_markdown=tables_md,
page_image_base64=page_img,
has_visual_content=True, # PPT 每页都走 VL——排版也是信息
source_metadata={
"source": Path(pptx_path).name,
"slide": page_data["index"] + 1,
},
))
return units
5.2 千问 VL 描述一页幻灯片
这一步是整个 PPT 拆分的核心。prompt 要仔细设计——不能太宽泛（VL 会自由发挥），也不能太窄（漏掉重要信息）：
PPT_DESCRIBE_PROMPT = """这是 PPT 一页幻灯片的截图。请你分析并输出严格 JSON：
{
"slide_title": "这页的标题（如果有）",
"slide_type": "标题页 / 内容页 / 图表页 / 总结页",
"text_content": "页面上所有文字，按阅读顺序。表格必须用 Markdown 格式写在里面",
"visual_elements": [
{
"type": "image / chart / diagram / icon",
"description": "用自己的话描述。架构图要写清楚节点和箭头——比如'API Gateway → Service A → Database，Gateway 收到请求后路由到对应的 Service'",
"position": "大概位置"
}
],
"layout_hint": "这一页的排版结构——左右两栏/标题在上内容在下/三列卡片 等",
"key_takeaway": "这一页最核心的信息，一句话"
}
要点：
- visual_elements 不要放过任何图片、图表、流程图
- 架构图的箭头关系是核心价值，描述时写清楚谁指向谁
- 表格用 Markdown 格式写进 text_content 字段
- 只输出 JSON，不要前后加说明文字"""
def describe_ppt_slide(unit: PageUnit) -> dict:
response = vl_client.chat.completions.create(
model="qwen-vl-max",
temperature=0.1, # 设低，保证同一页跑两次 JSON 结构一致
max_tokens=3000, # 一页幻灯片的完整描述通常在 500-1500 字
messages=[{
"role": "user",
"content": [
{"type": "text", "text": PPT_DESCRIBE_PROMPT},
{
"type": "image_url",
"image_url": {
# data:image/png;base64,... 直接在 HTTP body 里传图
# 不需要先把图片上传到某个存储
"url": f"data:image/png;base64,{unit.page_image_base64}"
},
},
],
}],
)
raw = response.choices[0].message.content
return _safe_json_parse(raw)
三个必设参数的原因：
temperature=0.1是为了 JSON 结构稳定。默认值是 0.7-1.0，同一张图跑两次输出的字段名可能不一样——第一次是slide_title，第二次是title。下游代码按slide_title读字段，第二次就读空了max_tokens=3000是防止模型"自由发挥"写一堆废话撑爆输出。一页幻灯片的文字通常 500-1500 字，3000 token 完全够model="qwen-vl-max"是最强的 VL 版本。还有一个qwen-vl-plus便宜 30% 但 OCR 中文小字时偶尔翻车。对文档处理来说准确率比省那几毛钱重要
5.3 一个例子的效果
一页标题为"计算机学院课程体系"的幻灯片。页面上有：顶部标题、左侧课程关系流程图（7 个框+箭头）、右侧 5 行课程表格、底部一行注释。
千问 VL 输出：
{
"slide_title": "计算机学院课程体系",
"slide_type": "图表页",
"text_content": "计算机学院课程分为三大方向：软件理论、人工智能、网络与安全。\n| 方向 | 必修课 | 选修课 |\n|------|--------|--------|\n| 软件理论 | 6 | 4 |\n| 人工智能 | 5 | 5 |\n| 网络与安全 | 5 | 3 |",
"visual_elements": [
{
"type": "diagram",
"description": "课程先修关系流程图：CS101程序设计基础(大一上) → CS201数据结构与算法(大一下) → 分为两条支线：CS301算法设计与分析(大二上) 和 CS310数据库原理(大二上) → 最终汇聚到 CS401编译原理(大二下)。每门课标注了学期和学分",
"position": "左侧60%区域"
}
],
"layout_hint": "左右两栏：左侧流程图展示先修关系，右侧表格列课程数据",
"key_takeaway": "数据结构与算法(CS201)是整个课程体系的核心枢纽，几乎所有高级课程都以它为先修课"
}
如果只用文字提取能拿到什么？PPTX 的 XML 里只有文字碎片："计算机学院课程体系三大方向软件理论人工智能网络与安全核心课程..."——连"数据结构"和"CS201"之间的对应关系都丢了。
VL 补上了：流程图里 7 个节点的先修依赖网、表格的结构、核心结论。全转成了可检索的文字。
5.4 从 VL 输出拆 chunk
def ppt_slide_to_chunks(unit: PageUnit, vl_desc: dict) -> list[dict]:
# 拼接所有信息源
composite = f"PPT {unit.source_metadata['source']} 第{unit.source_metadata['slide']}页"
if vl_desc.get("slide_title"):
composite += f" - {vl_desc['slide_title']}"
composite += "\n"
# VL 输出的正文 + Markdown 表格
composite += vl_desc.get("text_content", "") + "\n"
# 每个视觉元素转成一句描述：[diagram] 先修关系流程图：CS101 → CS201 → ...
for ve in vl_desc.get("visual_elements", []):
composite += f"[{ve.get('type', '元素')}] {ve.get('description', '')}\n"
# 核心要点——一页最浓缩的东西
if vl_desc.get("key_takeaway"):
composite += f"[要点] {vl_desc['key_takeaway']}\n"
# 按句子边界切，chunk_size=500 适合大多数一页=一个 chunk 的情况
return _semantic_chunk_split(composite, chunk_size=500, overlap=50)
六、Word 拆分
Word 文档的结构比 PPT 层次丰富——标题 H1-H6、正文段落、嵌入图片、表格、页眉页脚。百炼文档解析 API 能识别标题层级，但对嵌入在段落中间的图片，只能告诉你"这里有一张图"（返回一个 image URL），不知道图里是什么。
Word 的策略：检测到嵌入图片的页 → 把整页渲染成图 → 送 VL 补充图片描述。纯文字页直接拆，不走 VL。
6.1 docx → PageUnit 列表
def docx_to_page_units(docx_path: str) -> list[PageUnit]:
file_id = _upload_to_bailian(docx_path)
task_id = _submit_parse_job(file_id, file_type="docx")
parse_result = _wait_for_result(task_id)
units = []
for page_data in parse_result["pages"]:
text = page_data.get("text", "")
tables_md = [_table_to_markdown(t) for t in page_data.get("tables", [])]
# 百炼返回的 images 字段：这页里检测到的嵌入图片的 URL 列表
embedded_images = page_data.get("images", [])
has_visual = len(embedded_images) > 0
page_img = ""
if has_visual:
# 有图 → 整页渲染成图，后面送 VL。
# 注意 Word 文档解析 API 不会自动渲染页面截图（和 PPT 不同），
# 需要自己用 LibreOffice 等工具渲染
page_img = _render_docx_page(docx_path, page_data["index"])
units.append(PageUnit(
page_index=page_data["index"],
raw_text=text,
tables_as_markdown=tables_md,
page_image_base64=page_img,
has_visual_content=has_visual,
source_metadata={
"source": Path(docx_path).name,
"outline_level": page_data.get("outline_level", 0),
"section": page_data.get("section", ""),
},
))
return units
outline_level 是百炼解析 Word 时自动识别的标题层级。1 表示一级标题（H1），2 表示二级标题（H2），0 表示正文段落。这个信息保留到 metadata 里——后续检索时如果想只在标题里搜，可以用这个字段过滤。
6.2 Word 嵌入图片的描述
Word 里嵌的图片不是独立的——它周围有文字在解释它。把周围文字和图片一起送 VL，模型能结合上下文判断图片类型：
def describe_docx_page_with_image(unit: PageUnit) -> Optional[dict]:
if not unit.has_visual_content or not unit.page_image_base64:
return None
# 拿周围文字作为上下文——帮 VL 理解图片在说什么
context_text = unit.raw_text[:600]
prompt = f"""这是 Word 文档一页的截图。页面文字内容如下：
---
{context_text}
---
请只描述页面上嵌入的图片/图表/架构图（不要重复已有文字内容），输出 JSON：
{{
"images_described": [
{{
"type": "diagram / chart / photo / screenshot",
"content_summary": "图的具体内容。架构图写调用关系。图表解读数据",
"relation_to_text": "这张图和页面文字的关系——解释某个概念/提供数据/独立展示",
"caption": "图注（如果有）"
}}
]
}}"""
response = vl_client.chat.completions.create(
model="qwen-vl-max",
temperature=0.1,
max_tokens=2000,
messages=[{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {
"url": f"data:image/png;base64,{unit.page_image_base64}"
}},
],
}],
)
return _safe_json_parse(response.choices[0].message.content)
6.3 Word 页面→chunk
def docx_page_to_chunks(unit: PageUnit, image_desc: dict | None) -> list[dict]:
composite = ""
section = unit.source_metadata.get("section", "")
if section:
composite += f"[章节: {section}]\n"
composite += unit.raw_text + "\n"
for tmd in unit.tables_as_markdown:
composite += tmd + "\n"
if image_desc:
for img in image_desc.get("images_described", []):
composite += (
f"[嵌入{img.get('type', '图片')}]: {img.get('content_summary', '')}"
)
rel = img.get("relation_to_text", "")
if rel:
composite += f" (与上下文关系: {rel})"
cap = img.get("caption", "")
if cap:
composite += f" [图注: {cap}]"
composite += "\n"
return _semantic_chunk_split(composite, chunk_size=800, overlap=100)
七、PDF 拆分
PDF 是最复杂的格式——可能是纯文字（打印的规章制度）、图文混排（实验指导书）、甚至整本扫描件。
三种形态需要用不同策略处理。我写了一个分类函数先判断类型，再决定哪些页需要 VL。
7.1 分类
def classify_pdf(text_per_page: list[str]) -> str:
"""
判断 PDF 类型。
三个指标：
- avg_text: 每页平均字数。扫描件 0-10 字，文字 PDF 500-2000 字
- blank_ratio: 低文字页占比。扫描件超过一半的页几乎没有文字
返回:
- "scanned": 整本扫描件，每页都需要 VL OCR
- "mixed": 图文混排，有图/表的页才需要 VL
- "text_heavy": 纯文字，完全不需要 VL
"""
total = "".join(text_per_page)
avg = len(total) / max(len(text_per_page), 1)
blanks = sum(1 for t in text_per_page if len(t.strip()) < 50)
if avg < 100 or blanks > len(text_per_page) * 0.5:
return "scanned"
elif avg < 500:
return "mixed"
else:
return "text_heavy"
三种类型的处理策略：
| 类型 | 文档解析结果 | VL 策略 | 每页额外成本 |
|---|---|---|---|
| text_heavy | 文字完整 | 0页需要VL | ¥0 |
| mixed | 文字可靠，图/表缺失 | 仅对有图表的页调VL | ~¥0.01/页 |
| scanned | 文字约等于空 | 每页调VL做OCR | ~¥0.015/页 |
7.2 PDF 渲染 + PageUnit 构建
import fitz # PyMuPDF
def _render_pdf_to_base64_images(pdf_path: str, dpi: int = 200) -> list[str]:
"""把 PDF 每一页渲染成 base64 PNG。
DPI=200 的取舍：
- 300 DPI：一页 A4 渲染出来约 15MB base64，传给 VL 的 token 飙到 1.5 万+。
一份 30 页 PDF 光图片 token 就 50 万，费用太高
- 150 DPI：小号中文字可能出现模糊，表格数字识别不准
- 200 DPI：一页约 5-8MB base64，中文字清晰可辨，token 适中。
实测 200 和 300 的 OCR 准确率几乎一样"""
doc = fitz.open(pdf_path)
images = []
for page in doc:
pix = page.get_pixmap(dpi=dpi)
images.append(base64.b64encode(pix.tobytes("png")).decode())
doc.close()
return images
def pdf_to_page_units(pdf_path: str) -> list[PageUnit]:
# 文档解析
file_id = _upload_to_bailian(pdf_path)
task_id = _submit_parse_job(file_id, file_type="pdf")
parse_result = _wait_for_result(task_id)
pages = parse_result["pages"]
# 分类 + 按需渲染
text_per_page = [p.get("text", "") for p in pages]
pdf_type = classify_pdf(text_per_page)
needs_images = pdf_type in ("scanned", "mixed")
page_imgs = _render_pdf_to_base64_images(pdf_path, dpi=200) if needs_images else []
# 构建 PageUnit
units = []
for page_data in pages:
text = page_data.get("text", "")
tables_md = [_table_to_markdown(t) for t in page_data.get("tables", [])]
has_imgs = len(page_data.get("images", [])) > 0
text_len = len(text.strip())
needs_vl = (
pdf_type == "scanned"
or text_len < 100
or has_imgs
or len(tables_md) > 0
)
units.append(PageUnit(
page_index=page_data["index"],
raw_text=text,
tables_as_markdown=tables_md,
page_image_base64=page_imgs[page_data["index"]] if needs_vl else "",
has_visual_content=needs_vl,
source_metadata={
"source": Path(pdf_path).name,
"page": page_data["index"] + 1,
"total_pages": len(pages),
"pdf_type": pdf_type,
},
))
return units
7.3 PDF 每页的 VL 描述
有文字和没文字的页面用不同的 prompt 策略：
def describe_pdf_page(unit: PageUnit) -> dict:
has_text = len(unit.raw_text.strip()) > 100
if has_text:
# 有基础文字 → VL 只补充图表，不重复 OCR
prompt = f"""这是 PDF 一页的截图。已有文字提取：
---
{unit.raw_text[:800]}
---
只描述文字没覆盖的视觉内容（图片、图表、架构图）。输出 JSON：
{{
"figures": [
{{"type": "diagram/chart/screenshot/photo",
"position": "大概位置",
"summary": "图的内容。架构图写清节点和箭头"}}
],
"tables_markdown": ["文档解析漏掉的表格，用 Markdown 补"],
"extra_text": "文字提取遗漏的内容"
}}"""
else:
# 扫描件：整页靠 VL 做 OCR
prompt = """这是 PDF 扫描件，请做完整 OCR。输出 JSON：
{{
"full_text": "页面上所有文字。表格用 Markdown",
"figures": [
{{"type": "diagram/chart/photo",
"summary": "图的内容"}}
],
"language": "zh / en / mixed"
}}
注意多栏排版按阅读顺序输出。"""
response = vl_client.chat.completions.create(
model="qwen-vl-max",
temperature=0.1,
max_tokens=4000,
messages=[{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {
"url": f"data:image/png;base64,{unit.page_image_base64}"
}},
],
}],
)
return _safe_json_parse(response.choices[0].message.content)
为什么要分两种 prompt？ 如果某页已经用文档解析拿到了 500 字文字，还让 VL 重新 OCR——浪费钱（输出 token 也收费），而且可能出现两版文字不一致的尴尬。让 VL 只补文档解析漏掉的内容——输出更短、更省钱、信息不重复。
7.4 PDF 页面→chunk
PDF 有天然的页边界，不需要跨页做语义拆分。一页内的内容按句子边界切：
def pdf_page_to_chunks(unit: PageUnit, vl_desc: dict | None) -> list[dict]:
chunks = []
# === 文字 chunk ===
text_content = unit.raw_text
if vl_desc:
# 扫描件：VL 输出的 full_text 替代空文字
if vl_desc.get("full_text"):
text_content = vl_desc["full_text"]
# 补充遗漏的文字
if vl_desc.get("extra_text"):
text_content += "\n" + vl_desc["extra_text"]
# 拼接表格
for tmd in unit.tables_as_markdown:
text_content += "\n" + tmd
if vl_desc:
for tmd in vl_desc.get("tables_markdown", []):
text_content += "\n" + tmd
for c in _semantic_chunk_split(text_content, chunk_size=600, overlap=80):
chunks.append({
"content": c,
"page": unit.source_metadata["page"],
"chunk_type": "text",
})
# === 图片 chunk — 单独建，和文字分开 ===
# 好处：向量检索搜"微服务架构图"能命中 diagram chunk；
# 前端可以根据 chunk_type 决定渲染文字还是原图
if vl_desc:
for fig in vl_desc.get("figures", []):
chunks.append({
"content": f"[PDF图片 {fig.get('type', '')}] {fig.get('summary', '')}",
"page": unit.source_metadata["page"],
"chunk_type": "figure",
"source_image_b64": unit.page_image_base64, # 存原图
})
return chunks
八、Markdown 拆分
Markdown 是四种格式里最简单的——它本来就是纯文本。唯一的问题是  这种本地图片引用。文本拆分器只能看到一个相对路径，不知道图片的内容。
处理策略：解析图片引用 → 读到 base64 → 送 VL 描述 → 把描述插回原文。
8.1 Markdown → 段落块
按 ## 标题拆成块——Markdown 的标题是天然的语义边界，每个 ## 下面的内容围绕同一个主题。这样拆出来的块自带主题一致性：
def _split_markdown_by_heading(text: str, level: int = 2) -> list[dict]:
"""按 ## 标题拆分 Markdown。"""
pattern = rf'(^|\n)({{{level}}})\s+(.+?)\n'
splits = list(re.finditer(pattern, text))
if not splits:
return [{"text": text, "heading": ""}]
blocks = []
for i, match in enumerate(splits):
start = match.end()
end = splits[i + 1].start() if i + 1 < len(splits) else len(text)
blocks.append({
"text": text[start:end].strip(),
"heading": match.group(3).strip(),
})
return blocks
def markdown_to_page_units(md_path: str) -> list[PageUnit]:
with open(md_path, "r", encoding="utf-8") as f:
content = f.read()
blocks = _split_markdown_by_heading(content, level=2)
# Markdown 文件所在目录——解析相对路径图片引用时需要
base_dir = Path(md_path).parent
units = []
for i, block in enumerate(blocks):
# 找到所有图片引用：
img_refs = re.findall(r'!\[([^\]]*)\]\(([^)]+)\)', block["text"])
images_b64 = []
for alt_text, img_path in img_refs:
# 只处理本地图片，远程 URL 保留原样
if not img_path.startswith(("http://", "https://")):
resolved = base_dir / img_path
if resolved.exists():
with open(resolved, "rb") as f:
img_b64 = base64.b64encode(f.read()).decode()
images_b64.append({
"alt": alt_text,
"path": str(resolved),
"base64": img_b64,
})
else:
print(f"[WARN] 图片不存在: {resolved}")
# 把  替换成 [图片: alt] 占位符
clean_text = re.sub(r'!\[([^\]]*)\]\([^)]+\)', r'[图片: \1]', block["text"])
units.append(PageUnit(
page_index=i,
raw_text=clean_text,
tables_as_markdown=_extract_md_tables(block["text"]),
page_image_base64="",
has_visual_content=len(images_b64) > 0,
source_metadata={
"source": Path(md_path).name,
"heading": block.get("heading", ""),
"images": images_b64,
},
))
return units
_extract_md_tables 从 Markdown 中提取已有的表格——它们本身就是 Markdown 格式了，不需要转换：
def _extract_md_tables(text: str) -> list[str]:
"""从 Markdown 中提取已有的表格。匹配连续 | 开头的行。"""
lines = text.split("\n")
tables, current = [], []
in_table = False
for line in lines:
s = line.strip()
if s.startswith("|") and s.endswith("|"):
current.append(s)
in_table = True
else:
if in_table and len(current) >= 2:
tables.append("\n".join(current))
current = []
in_table = False
if in_table and len(current) >= 2:
tables.append("\n".join(current))
return tables
8.2 Markdown 图片的 VL 描述
Markdown 里的图片是嵌入在上下文中的——一张架构图下面写着"图3-1 系统架构图"。把周围文字和图片一起传 VL，效果比单独传图片好得多：
def describe_markdown_images(unit: PageUnit) -> list[dict]:
images = unit.source_metadata.get("images", [])
if not images:
return []
context = unit.raw_text[:300]
results = []
for img in images:
prompt = f"""这张图片来自 Markdown 文档。
前后文：{context}
alt 文本：{img['alt']}
分析图片，输出 JSON：
{{
"figure_type": "diagram / chart / screenshot / photo / illustration",
"summary": "图片内容——架构图写清节点和箭头，截图 OCR 界面上关键文字",
"relation_to_text": "图与文档文字的关系——解释上文/提供示例/独立展示/补充数据"
}}"""
response = vl_client.chat.completions.create(
model="qwen-vl-max",
temperature=0.1,
max_tokens=1500,
messages=[{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {
"url": f"data:image/png;base64,{img['base64']}"
}},
],
}],
)
results.append(_safe_json_parse(response.choices[0].message.content))
return results
8.3 Markdown 块→chunk
def markdown_block_to_chunks(unit: PageUnit, img_descs: list[dict]) -> list[dict]:
composite = ""
heading = unit.source_metadata.get("heading", "")
if heading:
composite += f"## {heading}\n"
composite += unit.raw_text + "\n"
for tmd in unit.tables_as_markdown:
composite += tmd + "\n"
for img in img_descs:
composite += (
f"[{img.get('figure_type', '图片')}]: {img.get('summary', '')}"
)
if img.get("relation_to_text"):
composite += f" (与上下文关系: {img['relation_to_text']})"
composite += "\n"
return _semantic_chunk_split(composite, chunk_size=600, overlap=60)
九、共用工具函数
语义拆分器——按句子边界切，不断在句子中间：
def _semantic_chunk_split(text: str, chunk_size: int, overlap: int) -> list[str]:
"""按句号、问号、感叹号、换行切句子。不在句子中间切断。
为什么不能按固定字符数硬切？
在一个句子中间切断——比如"数据结构是计算"——embedding 模型
看到半句话，输出的向量和完整句子的向量差异很大。
检索质量因为切分方式不好而降一大截。"""
if len(text) <= chunk_size:
return [text] if text.strip() else []
# 在句号、问号、感叹号、换行符处切
sentences = re.split(r'(?<=[。！？\n])', text)
sentences = [s for s in sentences if s]
chunks = []
current = ""
for sent in sentences:
if len(current) + len(sent) <= chunk_size:
current += sent
else:
if current.strip():
chunks.append(current.strip())
if overlap > 0 and len(current) > overlap:
current = current[-overlap:] + sent
else:
current = sent
if current.strip():
chunks.append(current.strip())
return chunks
overlap 参数让相邻 chunk 之间有一段重叠——如果用户的问题刚好跨两个 chunk 的边界，两边都能命中。
安全 JSON 解析——LLM 的输出不是完美的 JSON：
def _safe_json_parse(text: str) -> dict:
"""LLM 输出的 JSON 可能前后有废话、被 markdown 包裹。处理掉。"""
text = re.sub(r'```\w*\n?', '', text) # 去掉 ```json 包裹
start = text.find('{')
end = text.rfind('}')
if start != -1 and end != -1 and end > start:
text = text[start:end + 1]
try:
return json.loads(text)
except json.JSONDecodeError as e:
print(f"[WARN] JSON 解析失败: {e}")
return {"raw": text, "parse_error": True}
Word 页面渲染——用 LibreOffice headless 把 DOCX 转 PDF，再取对应页：
def _render_docx_page(docx_path: str, page_index: int) -> str:
import subprocess, tempfile, fitz
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
pdf_path = tmp.name
# LibreOffice headless 转 PDF
subprocess.run([
"libreoffice", "--headless", "--convert-to", "pdf",
"--outdir", str(Path(pdf_path).parent), docx_path,
], check=True, timeout=30)
# 取对应页渲染
doc = fitz.open(pdf_path)
pix = doc[page_index].get_pixmap(dpi=200)
img = base64.b64encode(pix.tobytes("png")).decode()
doc.close()
Path(pdf_path).unlink(missing_ok=True)
return img
十、统一入口——传文件路径，出 chunk 列表
四种格式的单页处理逻辑全部就绪了。接下来需要一个统一入口——识别文件格式、分发给对应的处理函数、并发调 VL：
def process_document(file_path: str, max_workers: int = 4) -> list[dict]:
"""
统一入口。给一个文件路径，返回所有 chunk。
参数:
- file_path: 支持 .pptx / .docx / .pdf / .md
- max_workers: 并发调 VL 的线程数。设 4 是因为百炼免费版
qwen-vl-max 并发限制 5，留 1 个 buffer 防止 429
返回格式:
[{"content": "chunk文字", "page": 1, "chunk_type": "text"}, ...]
"""
ext = Path(file_path).suffix.lower()
# 每种格式对应三个函数：(构建units, VL描述, chunk拆分)
if ext == ".pptx":
units = pptx_to_page_units(file_path)
describe_fn, chunk_fn = describe_ppt_slide, ppt_slide_to_chunks
elif ext == ".docx":
units = docx_to_page_units(file_path)
describe_fn, chunk_fn = describe_docx_page_with_image, docx_page_to_chunks
elif ext == ".pdf":
units = pdf_to_page_units(file_path)
describe_fn, chunk_fn = describe_pdf_page, pdf_page_to_chunks
elif ext == ".md":
units = markdown_to_page_units(file_path)
describe_fn, chunk_fn = describe_markdown_images, markdown_block_to_chunks
else:
raise ValueError(f"不支持: {ext}")
vl_count = sum(1 for u in units if u.has_visual_content)
print(f"{len(units)} 页/块, 需VL: {vl_count}")
all_chunks = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {}
for unit in units:
if unit.has_visual_content:
# 需要 VL → 提交到线程池，和其他页面并发跑
futures[executor.submit(describe_fn, unit)] = unit
else:
# 纯文字页 → 直接拆分，不占线程池
all_chunks.extend(chunk_fn(unit, None))
# as_completed: 谁先完成先收谁——快的不用等慢的
for future in as_completed(futures):
unit = futures[future]
try:
vl_result = future.result(timeout=60)
except Exception as e:
print(f"[ERROR] VL失败 page={unit.page_index}: {e}")
vl_result = None
all_chunks.extend(chunk_fn(unit, vl_result))
all_chunks.sort(key=lambda c: c.get("page", 0))
return all_chunks
用法就是这么简单：
# 四种格式全是一行
ppt_chunks = process_document("计算机学院课程体系.pptx")
doc_chunks = process_document("实验指导书-数据库设计.docx")
pdf_chunks = process_document("考试管理办法2025版.pdf")
md_chunks = process_document("课程笔记-数据结构.md")
# 拿到 chunk 后，做 embedding 写向量库
for chunk in ppt_chunks + doc_chunks + pdf_chunks + md_chunks:
vec = your_embedding_fn(chunk["content"])
vector_db.insert(vector=vec, metadata=chunk)
十一、踩过的坑
DPI 别设太高
开始 PDF 渲染设了 DPI=300。一页 A4 渲染出来 ~15MB base64，传给千问 VL 的 image token 飙到快两万。30 页 PDF 处理完花了 5 块钱。
降到 200 之后，一页 ~5MB，肉眼对比 OCR 质量没差别，但 token 和费用都降了 60%。
现在所有渲染都用 200。300 费钱，150 小字模糊。
temperature=0.1 不能省
最早写 prompt 时没注意 temperature，用的默认值（0.7 或 1.0）。后果：同一页跑两次，输出的 JSON 字段名不一样。第一次 {"slide_title": "课程体系"}，第二次 {"title": "课程体系"}。下游代码按 slide_title 读，第二次读空。
设成 0.1 之后稳定了。偶尔还会有格式波动但不影响字段名。
Markdown 里相对路径的图片
 这种引用是相对路径。程序必须知道 Markdown 文件的目录才能解析。我一开始写死了绝对路径——本地跑没问题，到服务器上就找不到文件。
改成 base_dir = Path(md_path).parent，所有图片路径基于 md 文件所在目录解析。
并发太多触发 API 限流
最早设了无限制并发——ThreadPoolExecutor 来多少跑多少。30 页 PDF 同时飞 30 个 VL 请求，直接撞百炼的并发限流（免费版 5 个），一半返回 429。
设 max_workers=4（留 1 个 buffer），之后再没触过限流，整体处理反而更快——省掉了重试等待的时间。
超长图截断
有的 PDF 页是超长截图（教务系统的系统架构全景图），200 DPI 下渲染出来高度超过 8000px。千问 VL 超过一定高度后识别率明显下降——底部的文字可能漏掉。
解决：渲染时限制最大高度 4000px，超出的切成多个子图分别送 VL。
十二、成本
按量付费的价格（2026 年 5 月）：
| API | 单价 | 30页PDF典型消耗 |
|---|---|---|
| 文档解析 | 限免期 ¥0 | ¥0 |
| qwen-vl-max 输入 | ¥0.003/千token | 15页×3K = 45K ≈ ¥0.14 |
| qwen-vl-max 输出 | ¥0.012/千token | 15页×0.5K = 7.5K ≈ ¥0.09 |
| 合计 | ≈ ¥0.23 |
0.23 元把一份 30 页文档拆完。关键省钱点——纯文字页不走 VL（PDF 中约 60% 的页跳过），DPI 设 200 不设 300，temperature 压低避免废话撑 token。
整条管线约 600 行代码，按格式分了四块，每块的结构一样：解析成 PageUnit → 按需 VL 描述 → 拆 chunk。最后 process_document 统一入口收口。
在自己的百炼账号配好 DASHSCOPE_API_KEY 和 BAILIAN_APP_ID 就能跑。四种格式，一个函数，输出直接接向量库。
十三、总结一下
🔄 全局处理时序图
[你的电脑] [阿里百炼云端] [大模型 qwen-vl-max]
| | |
|-- 1. 发送PDF/Word文件 (HTTP POST) --->| |
| |-- 2. 存入临时仓库 |
|<-- 3. 返回 file_id (身份证号) --------| |
| | |
|-- 4. 提交解析任务 submit_parse_job -->| |
| (带上 file_id) |-- 5. 启动底层【文档解析引擎】 |
| | (OCR + 版面分析，不消耗大模型Token) |
| | |
|-- 6. 轮询查询任务状态 ---------------->| |
|<-- 7. 返回解析结果 JSON --------------| |
| (包含: 纯文字, 表格Markdown, | |
| 图片Base64编码, 页码等) | |
| | |
|=================== 【进入你的本地代码逻辑】 ===================== |
| | |
|--- 8. 遍历每一页数据 ---------------------------------------> |
| 判断: 这页有复杂的图表吗？ | |
| | |
| [如果有图] | |
| |-- 9. 提取图片Base64 + Prompt ----->|---------------------------------------->|
| | | |-- 10. 看图说话
| |<-- 11. 返回图片描述文本 <-----------|-----------------------------------------|
| | | |
| [如果没图] | |