夜雨聆风前两篇把骨架和入口讲完了。这一篇来到流水线的「眼睛」——
NodeMDImg。它是 PDF 路径和 MD 路径的汇流点，也是流水线里唯一一个和多模态大模型打交道的节点——把本地图片变成「VLM 生成的描述 + MinIO 公网 URL」，下游才能稳定引用、可被检索。492 行代码藏着五段工程品味：正则从
.*?演化到字符类的血泪、按章节边界给 VLM 上下文而不是按字数硬切、滑动窗口限流器、re.sub替换字符串的反向引用坑。逐行讲完你会看到：「让流水线会看图」这件事远比想象中精巧。
一、定位：汇流点 + 协议终结点
先把流水线再摆一遍，注意看 node_md_img 在哪儿：
node_entry (入口分流)
├── file_type=md ─► node_md_img ─► node_document_split ─► ...
└── file_type=pdf ─► node_pdf_to_md ─► node_md_img ─► node_document_split ─► ...
│
▼
node_item_name_recognition
node_bge_embedding
node_import_milvus
node_knowledge_graph ─► END两个细节决定这个节点的角色：
它是两条路径的汇流点——PDF 走完 node_pdf_to_md之后流到这里，MD 直接从node_entry流到这里。汇流之后所有节点拿到的都是同一形态的 MD。它是协议终结点——汇流之前，MD 里的图片还是 这种本地相对路径，离开这台机器就失效；汇流之后变成，公网可访问、描述自带。下游节点完全不用再关心图片。
一句话总结它干的活：扫描 MD 里被引用的本地图片 → 调 VLM 给每张图写描述 → 上传 MinIO → 用新引用替换旧引用。听起来四步，代码里被拆成了 process 编排 + 六个私有方法。
process 的骨架：6 步编排 + 增量返回
先看 process 全貌，调度逻辑一眼就能看完：
def process(self, state: ImportGraphState) -> dict:
# 步骤1：初始化数据，获取MD核心信息
md_content, md_path_obj, images_dir = self._step_1_get_content(state)
if not images_dir.exists():
self.logger.info("无图片文件夹，跳过图片处理")
return {"md_content": md_content}
# 步骤2：扫描并筛选MD中引用的图片
target_images = self._step_2_scan_images(md_content, images_dir)
if not target_images:
self.logger.info("未检测到MD中引用了图片，跳过图片处理")
return {"md_content": md_content}
# 步骤3：调用多模态大模型生成图片摘要
summaries = self._step_3_generate_summaries(md_path_obj.stem, target_images)
# 步骤4：上传图片至MinIO，替换MD图片路径并填充摘要
new_md_content = self._step_4_upload_and_replace(md_path_obj.stem, target_images, summaries, md_content)
# 步骤5：备份并保存新MD文件
new_md_file_path = self._step_5_backup_new_md_file(md_path_obj, new_md_content)
# 步骤6：只返回增量字段，由 LangGraph merge 进 state（与 node_pdf_to_md 一致）
return {
"md_content": new_md_content,
"md_path": new_md_file_path
}三个工程取舍先点出来：
- 两道早退闸门
——没有 images文件夹（纯文本 MD），或者images里有图但 MD 没引用（MinerU 输出的图片和正文对不上），都直接return跳过，不浪费一次 VLM 调用。这是「先过滤再做事」的早退风格，和上一节NodePDFToMD的_step_1_validate_paths是一个套路。 - 六个
_step_N_xxx私有方法——和 NodePDFToMD一样的拆分粒度：process只负责编排，每步一个私有方法，单元测试可以瞄准单个 step 写。 - 返回增量 dict 不改入参
——结尾注释专门写「与 node_pdf_to_md 一致」。同一个流水线里两个重节点用同一种状态修改风格， NodeEntry那种直接改 state 的反而成了异类（上一篇提过的债）。
记忆点：节点的「单一职责」不是只做一件事，而是只暴露一个稳定的入参契约 + 一个稳定的增量返回契约。内部拆多少 step 是私事，外部永远只看到一个 process(state) -> dict。
二、双源策略：state 优先，回退读盘
第一节说过这是汇流点——两条上游路径产生的 state 形态不一样：
PDF 路径： node_pdf_to_md把转换后的 MD 内容写进state["md_content"]，同时把新文件路径写进state["md_path"]。MD 直入路径： node_entry只写了state["md_path"]，没有md_content——还没读过盘。
_step_1_get_content 用一个回退链把这两条路收口：
# 1. 参数非空校验
md_path = state.get("md_path")
if not md_path:
raise StateFieldError(field_name="md_path", message="MD文件路径不能为空", expected_type=str)
# 2. 路径转换与有效性检查
md_path_obj = Path(md_path)
if not md_path_obj.is_file():
raise FileProcessingError(message=f"文件 {md_path_obj.name} 不存在或不是文件")
# 3. 获取MD内容：PDF 路径上游已放进 state 直接复用；
# MD 直入路径（entry 只写 md_path）state 里没有内容，需回退读盘
md_content = state.get("md_content")
if not md_content:
with open(md_path_obj, "r", encoding="utf-8") as f:
md_content = f.read()
# 4. 获取md文件的图片文件夹路径
img_dir = md_path_obj.parent / "images"
return md_content, md_path_obj, img_dir这段写得很克制，但背后的设计值得点出来：state 是缓存层，磁盘是真相层。md_content 在 state 里就直接用上游算出来的结果（省一次读盘），不在就回退到磁盘（保证功能完整）。这和上一节 NodePDFToMD 下载 ZIP 后用 Path.rename 把 MD 落盘是配套的——上游一定先落盘再写 state，下游有 state 用 state、没 state 读盘，两边都能跑。
第 4 步那个 md_path_obj.parent / "images" 是个隐式契约：图片必须放在 MD 同级目录的 images/ 子目录里。MinerU 输出的就是这个布局，所以 PDF 路径没问题；但用户手动准备 MD 时如果图片放在 assets/ 或者直接和 MD 平级，这个节点就找不到图——这是隐式契约的代价。
⚠️ 隐式契约的代价：images/ 这个目录名硬编码在代码里，既不在 config.py 也不能从 MD 内容推断。生产上没问题（MinerU 输出固定），开放给用户上传 MD 时就是坑。修法是把目录名提成配置，或者扫描 MD 里的  反推出真实图片目录。
三、扫描图片：用集合查成员、用 debug 不制造噪音
把 images/ 目录扫一遍，挑出「格式支持 + MD 真的引用了」的图片：
for image_path_obj in images_dir.iterdir():
if not image_path_obj.is_file():
continue
image_file = image_path_obj.name
# 1. 过滤无效后缀
# images 目录里出现非图片文件属正常情况（如解析附带的 json），用 debug 而非 warning
if image_path_obj.suffix.lower() not in self.config.image_extensions:
self.logger.debug(f"图片{image_file}格式不支持")
continue
# 2. 组装图片的完整路径并转成字符串
img_path = str(image_path_obj)
# 3. 查找这个图片在md文档中引用的上下文
context = self._find_image_in_md(md_content, image_file)
if not context:
self.logger.debug(f"图片 {image_file} 未在md文档中找到")
continue
# 4. 将查询到的图片组装到列表中
target_images.append((image_file, img_path, context))三个工程取舍：
iterdir()替代os.listdir()——注释明确写了「与项目其他节点风格统一」。pathlib 迭代器返回的是 Path对象，后续.is_file()/.suffix/.name都不用再os.path.join拼。- 非图片文件用
debug不用warning——MinerU 输出目录里夹杂 json、txt是正常情况不是异常。日志级别的本质是「这条信息值得不值得运维看」，warning 刷屏会让真问题被淹没。 - 三元组
(file, path, context)——把上下文和文件信息一起带着走，下一步 VLM 直接用。如果只存 (file, path)，到 VLM 那一步还得重新算一次上下文，O(n)重复扫描。把数据「在最便宜的地方算好一起带走」是流水线优化的通用套路。
正则系统：一个共用方法，三个细节，一年的坑
这是这一篇最值得慢慢看的一段代码。先看 _build_img_ref_pattern 的完整实现——只有一行，但注释比代码长得多：
@staticmethod
def _build_img_ref_pattern(image_file: str) -> re.Pattern:
"""
构建 MD 图片引用正则，匹配形如 。
筛选（_find_image_in_md）和替换（_process_md_file）共用同一构建方法，保证两处永远一致。
描述用 [^\\]]*、路径用 [^)]*：懒惰的 .*? 遇到不匹配时仍会扩张跨过相邻引用，
把两个引用连成一个匹配；文件名前必须是 ( 或 /，避免 1.jpg 误匹配 a1.jpg
"""
return re.compile(r"!\[[^\]]*\]\((?:[^)]*/)?" + re.escape(image_file) + r"\)")三个细节，一个一个拆。这是真正写过生产代码被坑过的人才能写出的正则。
细节一：为什么不用 .*?
90% 的工程师第一反应会写：
r"!\[.*?\]\(.*?" + re.escape(image_file) + r"\)"看起来很合理——懒惰匹配嘛。但拿这串去匹配一段 MinerU 输出的 MD：
正文
当你想找 fig1.jpg 时，.*? 在「找不到 fig1.jpg」时会一直扩张——直到把两个引用合并成一个匹配，从第一个 ![ 一直吃到第二个 )]。替换之后整段 MD 就乱了。
懒惰匹配不是「匹配最短的」，而是「在能让整个正则成功的所有方案里选最短的」。当正则在当前位置找不到目标时，它会回溯——继续扩张。这是 .*? 的反直觉之处。
解法就是用字符类排除：[^\]]*（描述部分不含 ]）、[^)]*（路径部分不含 )）。这种「就地封顶」的写法把每个引用锁在自己的括号里，跨不过去。
⚠️ 别迷信 .*?：懒惰量词只在「能匹配到」时取最短；匹配不到时它会一直扩张，跨过相邻同结构片段。遇到「同结构的相邻重复」时优先用字符类排除，把每段匹配锁在自己的边界里。
细节二：文件名前的边界感
正则里这一段 (?:[^)]*/)? 是处理 images/xxx.jpg 这种带目录前缀的写法。但还有个更细的坑：找 1.jpg 时，a1.jpg / b1.jpg 都会被匹配——因为 1.jpg 是它们的后缀。
解法是「文件名前必须是 ( 或 /」：( 表示引用刚开头（），/ 表示在某个目录下（）。a1.jpg 前面是字母 a，被边界卡掉。
这是正则里的「左边界」技巧——和 SQL 注入防御、shell 引号转义是同一种思维方式：不光要匹配到，还要保证匹配的位置是「真实开头」。
细节三：re.escape 是默认动作
文件名拼进正则前先过 re.escape(image_file)——这是纯防御。MinerU 输出的图片文件名通常是 fig_1.jpg 这种安全字符，但万一某天用户上传的 PDF 里有 图 (1).jpg、图+1.jpg 这种带正则元字符的文件名，不 escape 就直接语法错或乱匹配。
这是「外部输入进正则必须 escape」的硬规则——和「外部输入进 SQL 必须参数化」、「外部输入进 HTML 必须转义」是同一类防御。re.escape 的成本是零，漏了就是 bug。
共用方法的价值：注释明确写了「筛选（
_find_image_in_md）和替换（_process_md_file）共用同一构建方法，保证两处永远一致」。这一句注释是整个正则系统的灵魂——筛选到的图和被替换的图必须是同一批，否则要么替换不到，要么替换错。共用方法把这种「两处必须一致」的约束编码进了代码结构，以后改正则只改一个地方。
四、上下文章节感知：给 VLM 一个语义完整的语境
这是这一篇里最有品味的一段代码。先看朴素做法会怎么写：
# ❌ 朴素做法：图片前后各取 200 字
idx = md_content.find(image_ref)
pre_text = md_content[max(0, idx-200):idx]
post_text = md_content[idx+len(image_ref):idx+len(image_ref)+200]能跑，但有两个问题：
- 章节跨越
——如果图片正好在「第三章」末尾， post_text会把「第四章」的内容也算成上下文。VLM 看到的语境是错乱的。 - 噪音混入
——前后 200 字会把相邻图片的 引用也吃进去。VLM 看到「上文：」对生成当前图片的摘要毫无帮助。
看看这个节点是怎么做的：
# 结构感知的上下文提取用到的两个行级正则
_HEADING_RE = re.compile(r"^#{1,6}\s+") # MD 标题行
_IMG_LINE_RE = re.compile(r"^!\[[^\]]*\]\([^)]*\)$") # 独立成行的图片引用
def _find_image_in_md(self, md_content, image_file, context_len=200):
"""
按行定位图片引用，返回结构化上下文（最近上级标题, 上文, 下文）。
上下文以章节为边界（不跨标题），并剔除相邻图片引用行
"""
pattern = self._build_img_ref_pattern(image_file)
lines = md_content.split("\n")
for idx, line in enumerate(lines):
if not pattern.search(line):
continue
# 1. 向上找最近的标题行，上文只取该标题之后的正文
head_title = ""
pre_start = 0
for i in range(idx - 1, -1, -1):
if self._HEADING_RE.match(lines[i]):
head_title = lines[i]
pre_start = i + 1
break
# 2. 向下截到下一个标题为止（不含标题本身）
post_end = len(lines)
for i in range(idx + 1, post_end):
if self._HEADING_RE.match(lines[i]):
post_end = i
break
# 3. 上文取靠近图片的末尾片段，下文取靠近图片的开头片段
pre_text = self._collect_context(lines[pre_start:idx], context_len, near_end=True)
post_text = self._collect_context(lines[idx + 1:post_end], context_len, near_end=False)
return head_title, pre_text, post_text
return None三步，每一步都解决一个具体问题：
- 向上找最近的标题
——确定章节起点。 pre_start = i + 1，上文从标题下一行开始（标题本身单独存进head_title，不重复进正文）。 - 向下截到下一个标题
——确定章节终点。 post_end = i（不含标题行），保证下文不会越界进下一章。 - 切片方向不对称
——上文取末尾 context_len字（near_end=True），下文取开头context_len字（near_end=False）。两边都向图片靠拢——离图片越近的内容相关性越强。
再看 _collect_context 怎么处理噪音：
@classmethod
def _collect_context(cls, lines, max_chars, near_end):
"""
拼接正文行为上下文：剔除空行和独立图片行（相邻图片的引用标记对摘要是噪音）
"""
text = "\n".join(
line for line in lines
if line.strip() and not cls._IMG_LINE_RE.match(line.strip())
)
return text[-max_chars:] if near_end else text[:max_chars]两个过滤条件：
line.strip()过滤空行——空行对摘要零贡献。 not cls._IMG_LINE_RE.match(line.strip())过滤独立成行的图片引用——相邻图片的标记是噪音，吃进去会让 VLM 困惑。
注意 _IMG_LINE_RE 用了 ^...$ 锚定整行——只过滤「整行就是一张图片引用」的情况，行内夹带的图片（比如 这是一张示意图  看这里）不会被剔除，因为这种行有正文信息。
上下文工程的核心：给模型语义完整的语境，而不是字数对齐的窗口。按字数硬切看起来上下文长度一致，但语义上可能跨越章节、夹带噪音；按章节边界切看起来长度不一，但每段都是完整的语义单元。VLM 拿到「第三章 3.2 节 安装步骤 ...... 图片 ...... 注意事项 ......」比拿到「字数 200 字 200 字」能生成准确得多的描述。
这是这一篇里我最想强调的工程品味。Prompt 工程的精华不在 prompt 模板，而在「喂给模型什么数据」。同样的 prompt 模板，喂「图片前后 200 字」和喂「图片所在章节的纯净正文」，效果天差地别。
记忆点：上下文工程 = 语义边界（按章节不按字数）+ 噪音剔除（空行、相邻图片）+ 方向不对称（向目标靠拢）。三步都为了一个目标——让每 token 都携带相关信号。
五、限流：滑动窗口 deque，参数提成类常量
调 VLM 是花钱的，更要命的是——各家多模态 API 都有严格的速率限制，超了就 429。这一节看节点怎么做限流。先看 _step_3_generate_summaries 的骨架：
def _step_3_generate_summaries(self, doc_stem, target_images):
summaries = {}
# 模型配置全程不变，客户端建一次全批复用；原先每张图 new 一个 ChatOpenAI，纯浪费
chat_model = ChatOpenAI(
model=llm_config.vl_model,
api_key=llm_config.api_key,
base_url=llm_config.base_url,
temperature=llm_config.llm_temperature
)
# 双端队列
request_deque = deque()
for img_file, img_path, context in target_images:
# 3.1 限速
self._apply_api_rate_limit(request_deque, self.RATE_LIMIT_MAX_REQUESTS, self.RATE_LIMIT_WINDOW_SECONDS)
# 3.2 向模型发送请求
summaries[img_file] = self._summarize_image(chat_model, img_path, root_folder=doc_stem, image_content=context)
return summaries两个工程取舍：
- 客户端建一次全批复用
——注释专门写了「原先每张图 new 一个 ChatOpenAI，纯浪费」。ChatOpenAI内部维护 HTTP 连接池（httpx.Client），每张图重建一次意味着 TLS 握手、TCP 慢启动全重来。10 张图就是 10 倍延迟。 deque外部初始化跨调用复用—— request_deque = deque()在循环外创建，所有图片共享一个时间戳队列，限流才是「全批 N 次/窗口」而不是「每张图独立 N 次/窗口」。
类常量化的故事：注释藏着一次重构
看类顶部这两行——和注释一起读：
# 限流参数提成类常量：原先硬编码在调用处，既是魔法数字，又和 docstring 里
# 写的「每分钟9次」对不上；集中定义后按大模型实际限额只需改这一处
RATE_LIMIT_MAX_REQUESTS = 3
RATE_LIMIT_WINDOW_SECONDS = 10这段注释是重构纪事：原版把 max_requests=5, window=60 写死在调用处，docstring 里却写「每分钟 9 次」——代码和文档各自漂移。提成类常量是止血的第一步，但更彻底的修法是把这两个参数从 llm_config 读——不同模型的限额不同，切模型时只改配置不动代码。
注意数值本身：3 次 / 10 秒 等效于 18 次 / 分钟，比原先的 5 次 / 分钟 宽松得多。这是一个实测后调整的数字，不是拍脑袋的。
算法拆解：滑动窗口三步走
把 _apply_api_rate_limit 完整贴出来——核心算法很短：
def _apply_api_rate_limit(self, request_times, max_requests, window_seconds=60):
# 1. 记录当前时间
current_time = time.time()
# 2. 清理滑动窗口中的过期请求
while request_times and current_time - request_times[0] >= window_seconds:
request_times.popleft()
# 3. 窗口内请求数达到上限，计算需要等待的时间并阻塞
if len(request_times) >= max_requests:
sleep_duration = window_seconds - (current_time - request_times[0])
if sleep_duration > 0:
self.logger.info(f"请求被限速，等待{sleep_duration:.2f}秒...")
time.sleep(sleep_duration)
current_time = time.time()
while request_times and current_time - request_times[0] >= window_seconds:
request_times.popleft()
# 4. 记录当前请求的时间戳，新请求入队
request_times.append(current_time)三步：
- 清过期
——从队头开始，把超过窗口时长的时间戳统统 popleft()。deque是按入队顺序排列的，队头一定是最老的，从队头清一定对。 - 满了就睡
——如果剩下的还是 ≥ max_requests，就算还要等多久：window_seconds - (current_time - request_times[0])。算的是「队头那个请求离过期还有多久」，time.sleep阻塞等。睡醒之后还要再清一次过期——睡的过程中时间过去了，可能有更多请求过期了。 - 入队
——把自己的时间戳追加到队尾。下次调用会看到它。
这就是经典的滑动窗口限流（sliding window log）。和「固定窗口」相比它的优势是不会出现边界突刺——固定窗口在窗口切换的瞬间可以让请求翻倍（0:59 发 N 个 + 1:00 发 N 个 = 1 秒内 2N 个），滑动窗口随时回看过去 N 秒，这种突刺不会发生。
但代价是空间 O(N)——要存所有时间戳。对图片批处理这个量级（几十张）完全无所谓，限流到几万 QPS 时就该换令牌桶（token bucket）或滑动窗口平均数（sliding window counter）算法了。
⚠️ 老问题又来了：time.time() 测时间间隔——和上一篇 NodePDFToMD 轮询循环是同一个坑。time.time() 是系统时钟，会被 NTP 校时、夏令时、人工改时间影响——如果系统时钟往回拨，current_time - request_times[0] 可能变成负数，限流逻辑直接乱套。测时间间隔必须用 time.monotonic()——它保证单调不减，不受系统时钟影响。这是 Python 标准库里专门为「测间隔」准备的函数。
记忆点：滑动窗口限流三件套——时间戳队列（按时间排序）+ 队头清过期（O(1) 摊销）+ 满了 sleep（阻塞而非拒绝）。空间 O(N) 换精确度，单机小流量首选；分布式大流量换 Redis + 令牌桶。
六、VLM 调用：base64 编码、MIME 真实后缀、文件名兜底
_summarize_image 是真正「看图说话」的地方，几个细节都值得点：
image_path_obj = Path(image_path)
# 1. 将图片转换成base64, 发给 llm
with open(image_path, "rb") as f:
base64_image = base64.b64encode(f.read()).decode("utf-8")
# 按真实后缀生成 MIME：原先硬编码 image/jpeg，png/webp 会被错误标注
mime_type = mimetypes.types_map.get(image_path_obj.suffix.lower(), "image/jpeg")
# 拼装结构化上下文，跳过为空的部分
head_title, pre_text, post_text = image_content
context_text = "\n".join(part for part in (head_title, pre_text, post_text) if part) or "暂无可用上下文"
try:
messages = [
{
"role": "user",
"content": [
{
"type": "text",
text": f"""这是"{root_folder}"文件中的一张图片，其所在章节及上下文如下：
"{context_text}"
请结合图片视觉内容和上述上下文，用中文简要总结这张图片的内容，用于 Markdown 图片标题。"""
},
{
"type": "image_url",
"image_url": {
"url": f"data:{mime_type};base64,{base64_image}"
}
}
]
}
]
response = chat_model.invoke(messages)
return response.content.strip().replace("\n", "")
except Exception as e:
self.logger.error(f"获取图片摘要失败:{image_path}， 错误：{e}")
# 兜底用图片文件名而不是文件夹名：文件夹名和图片内容毫无关联，
# 落到 MD 里所有失败图共用同一个标题，读者无法分辨；文件名至少可辨认
return image_path_obj.stem六个细节：
base64内嵌而不是 URL——这是 OpenAI 多模态 API 的两种图片输入方式。本地图片用 data:image/xxx;base64,...直接塞进 request body；如果图片已经在公网（比如 OSS URL）就传 URL 让模型自己下载。这里图片还在本地，base64 是唯一选择。- MIME 按真实后缀生成
——注释又藏了一笔重构：原先硬编码 image/jpeg，PNG、WebP 会被错标成 JPEG。OpenAI API 对 MIME 错配的容错性其实挺好，但有些模型会严格校验，错标会直接 400。按真实后缀查mimetypes.types_map，查不到才回落到image/jpeg，这是稳妥的兜底。 - 顶部
mimetypes.add_type("image/webp", ".webp")——Windows 自带的 MIME 表不全， .webp查不到会回落image/jpeg。模块加载时手动注册一次，进程内永久生效。这是「平台差异主动抹平」的标准做法。 - 上下文空值跳过
—— "\n".join(part for part in (head_title, pre_text, post_text) if part)，生成器 +if part真值过滤，or "暂无可用上下文"兜底全空。三段都不存在时也不会让 prompt 出现空字符串。 - 返回值
.strip().replace("\n", "")——VLM 经常会返回带换行的多句描述，但 Markdown 图片标题里换行会破坏引用结构。 replace("\n", "")强制压成一行——简单粗暴但有效。 - 异常兜底用文件名不用文件夹名
——这一段注释特别有教育意义：「文件夹名和图片内容毫无关联，落到 MD 里所有失败图共用同一个标题，读者无法分辨」。失败的图全标成「万用表RS-12的使用」，读者完全分不清是哪张；标成 fig_01/fig_02至少能定位。兜底值的可辨识性 > 兜底值的相关性。
异常兜底三原则：① 兜底值必须可辨识（这值是从哪个失败来的）；② 兜底值必须可定位（出了问题能溯源）；③ 兜底之后流程继续跑而不是整批失败。
return image_path_obj.stem这一行三个原则全占——可辨识（文件名唯一）、可定位（路径在错误日志里）、不阻断（流程继续走 MinIO 上传）。
但这里有个细节值得挑：except Exception as e 是宽口径异常——网络超时、模型 5xx、API key 错、JSON 解析失败，全兜进来。优点是抗造（一张图失败不影响整批），缺点是掩盖真正的环境问题。如果 API key 配错了，你会看到一堆「fig_01 / fig_02 / fig_03」全是兜底值，不点开日志根本不知道是 key 错了。生产做法是分级兜底：httpx.TimeoutException / 5xx 这种可重试的异常重试 N 次再兜底；401 / 403 这种不可重试的异常直接 raise 让上层感知。
七、MinIO 上传 + MD 替换：清旧不阻断、协议跟随、lambda 防反向引用
_step_4_upload_and_replace 是一个六步编排里的小编排：
# 1. 获取MinIO客户端
minio_client = get_minio_client()
# 2. 获取图片上传路径
minio_dir = minio_config.img_dir
upload_dir = f"{minio_dir}/{doc_stem}".replace(" ", "")
# 3. 清理已有目录
self._clean_minio_dir(minio_client, upload_dir)
# 4. 批量上传
urls = self._upload_images_batch(minio_client, upload_dir, target_images)
# 5. 合并图片摘要和URL，过滤上传失败的图片
image_info = self._merge_summary_and_url(summaries, urls)
# 6. 替换MD中的图片引用
md_content = self._process_md_file(md_content, image_info)
return md_content六个 step 各管一件事，但里面藏着三个工程取舍值得逐个拆。
取舍一：_clean_minio_dir 失败不阻断
def _clean_minio_dir(self, minio_client, upload_dir):
try:
objects_to_delete = minio_client.list_objects(minio_config.bucket_name, upload_dir, recursive=True)
delete_list = [DeleteObject(obj.object_name) for obj in objects_to_delete]
# 目录为空（首次处理该文档）时直接返回，不发无意义的批量删除请求
if not delete_list:
return
errors = minio_client.remove_objects(minio_config.bucket_name, delete_list)
for error in errors:
self.logger.error(f"删除图片错误：{error}")
except Exception as e:
# 清理失败只影响旧图残留，不阻断后续上传，故记日志后继续
self.logger.error(f"清理MinIO目录 {upload_dir} 失败：{e}")这段的核心是「清理」和「上传」是降级关系——清理失败没关系，上传会覆盖旧图；上传失败才是真问题。except 块的注释明确写了「不阻断后续上传」，这是对故障模式的清醒分类。
两个细节：
- 目录为空直接 return
——首次处理该文档时 MinIO 上没有旧数据， remove_objects传一个空列表是浪费一次 RPC。if not delete_list: return是早退风格。 remove_objects返回错误迭代器——MinIO SDK 的设计是批量删除容忍部分失败，返回的是一个 errors迭代器而不是直接抛异常。不迭代errors就看不到错误——这是个常见的 SDK 陷阱。
取舍二：_upload_to_minio 协议跟随配置
try:
content_type = mimetypes.guess_type(local_path)[0] or "application/octet-stream"
minio_client.fput_object(
bucket_name=minio_config.bucket_name,
object_name=object_name,
file_path=local_path,
content_type=content_type,
)
# 协议跟随 secure 配置，硬编码 http 会在开启 https 后生成全部失效的链接
scheme = "https" if minio_config.secure else "http"
return f"{scheme}://{minio_config.endpoint}/{minio_config.bucket_name}/{object_name}"
except Exception as e:
self.logger.error(f"上传图片失败：{local_path}，错误：{e}")
return None两个看点：
scheme = "https" if minio_config.secure else "http"——注释又是一次重构纪事：原先硬编码 http://，开 HTTPS 后生成的 URL 全部失效（HTTP 资源嵌入 HTTPS 页面会被浏览器 block）。「协议跟随配置」是基础工程素养，但容易被忘。return None作为失败哨兵——上传失败不抛异常，返回 None，下游_merge_summary_and_url用if url := urls.get(image_file)过滤掉。用 None 做哨兵值的争议：优点是简单，缺点是调用方很容易忘记检查——某天有人新写了个消费urls的代码忘了Nonecheck 就会炸。更稳的做法是返回Optional<str>并在类型层强约束（mypy 配合）。
取舍三：_process_md_file 用 lambda 防反向引用
这是这一篇的最后一个高光点。看 _process_md_file：
def _process_md_file(self, md_content, image_info):
for image_file, (summary, new_url) in image_info.items():
# 与筛选阶段共用 _build_img_ref_pattern，保证「筛选到的图」和「替换到的图」一致
pattern = self._build_img_ref_pattern(image_file)
# 用 lambda 而非替换字符串：摘要里若含 \ 会被 re.sub 当作反向引用而报错
md_content = pattern.sub(lambda m: f"", md_content)
return md_content为什么不直接写 pattern.sub(f"", md_content)？
因为 re.sub(repl_string, ...) 的第二个参数如果传字符串，会把字符串里的反斜杠当作反向引用——\1 引用第一个分组、\g<name> 引用命名分组、\\ 是字面反斜杠。
VLM 生成的摘要里出现 \ 完全可能——比如「C:\Users\ screenshot」、「路径分隔符 \」、「图 A\B 对照」。这时 re.sub 会把 \U 当成 Unicode 转义、\1 当成反向引用——要么报错，要么替换出乱七八糟的字符串。
解法是传函数不传字符串：pattern.sub(lambda m: replacement, text)。函数的返回值直接当作字面字符串使用，不做任何转义解释。这个细节是 Python re 模块的「必须知道」冷知识，但不知道的人不在少数——很多人写了多年 Python 也没踩到，是因为他们的替换字符串刚好不含反斜杠。外部数据进 re.sub 必须用函数形式，这是个硬规则。
⚠️ re.sub 替换字符串的反向引用陷阱：re.sub(pattern, repl_string, text) 会在 repl_string 里解释 \N / \g<name> / \\。替换内容来自外部数据（用户输入、模型生成）时，必须用 pattern.sub(lambda m: ..., text) 让返回值按字面处理。
最后两笔：备份原文件 + 增量返回
def _step_5_backup_new_md_file(self, md_path_obj, md_content):
new_md_path_obj = md_path_obj.with_name(f"{md_path_obj.stem}_new.md")
with open(new_md_path_obj, "w", encoding="utf-8") as f:
f.write(md_content)
return str(new_md_path_obj)原文件不变，新内容写 _new.md——这是数据安全的兜底。万一替换逻辑出了 bug，或者 VLM 摘要全错，原文件还在，可以重新跑。这是「写不动原文件」的工程纪律——和数据库迁移永远备份原表是同一种谨慎。
然后 process 末尾返回 {"md_content": new_md_content, "md_path": new_md_file_path}——增量 dict，LangGraph 自动 merge 进 state。下游 node_document_split 拿到的是带 MinIO URL 的 MD，从此再也不用关心图片在哪。
八、业界对照：能换什么、该换什么、别换什么
盘点这个节点可以怎么进化——分「生产可换」「规模到了再换」「别换」三类。
限流：滑动窗口 → 令牌桶 + 异步并发
当前是串行 + sleep：一张图调完睡一会儿再下一张。10 张图、每张 2 秒 + 限流等待，整个节点可能跑 30 秒以上。生产可改：
import asyncio
from aiolimiter import AsyncLimiter
# 18 次 / 分钟 = 0.3 次 / 秒，平滑发放
limiter = AsyncLimiter(max_rate=18, time_period=60)
async def summarize_all(images):
async def one(img):
async with limiter:
return await async_vlm_call(img)
return await asyncio.gather(*[one(img) for img in images])三个升级点：
- 异步并发
—— asyncio.gather让多张图的 HTTP 等待并行，单张 2 秒不变但整体 10 张从 20 秒压到 2-3 秒。 - 令牌桶代替滑动窗口
—— aiolimiter.AsyncLimiter内部就是令牌桶，空间 O(1)，平滑发放令牌，单机百万 QPS 都 hold 得住。 - SDK 切异步版
—— langchain_openai.ChatOpenAI已经支持ainvoke，不用换库。
但前提是整条流水线都异步化——单独把这一个节点改异步，外层 asyncio.run 包一下也能跑，但效果打折扣。LangGraph 本身支持 async 节点，全局升级是更彻底的路。
正则：手写正则 → Markdown AST
当前用正则匹配图片引用——能跑，但扛不住所有 Markdown 方言。比如带 title 的引用 、HTML 标签 <img src="...">、引用式图片 ![alt][ref]，正则都漏。
彻底的方案是用 Markdown 解析器——mistune / markdown-it-py——把 MD 解析成 AST，遍历 image 节点替换 src 再 render 回字符串。优点是覆盖所有方言，缺点是多一层依赖 + 性能开销。MinerU 输出的 MD 格式相对固定，正则目前够用——但代码里至少要加一行注释说「仅支持  标准语法」，否则后人接手时不知道边界。
VLM 调用：可观测性升级
当前可观测性靠 logging.info/error，能跑但排查问题全靠 grep 日志。生产推荐三件套：
structlog——结构化日志，每条日志带 task_id/img_file/elapsed_ms，ELK / Loki 直接查询。- OpenTelemetry tracing
——每张图一个 span，链路追踪看哪张图慢、哪张图失败、整体耗时分布。 - metrics（Prometheus）
——VLM 调用成功率、p50/p99 延迟、限流触发次数，Grafana 大盘实时看。
这一块不是这一篇的重点，但必须提——「能跑」和「能运维」之间差的就是可观测性。
九、收尾：492 行里的五条工程纪律
这一篇走完了。回头看，NodeMDImg 的 492 行藏着五条可以背下来的工程纪律：
- 正则遇到「同结构相邻重复」用字符类排除，不用
.*?——懒惰量词匹配不到时会回溯扩张， [^\]]*/[^)]*把匹配锁在边界里。 - 外部数据进正则必须 escape，进
re.sub必须用 lambda—— re.escape防文件名元字符，lambda防替换字符串里的反斜杠被反向引用。 - 上下文工程的精华在「喂什么」不在「怎么问」
——按章节边界切 + 剔除噪音 + 向目标靠拢，比按字数硬切有效得多。 - 限流器算法选型看规模
——单机小流量用滑动窗口（精确、简单），分布式大流量用令牌桶（空间 O(1)、平滑）。 - 异常兜底要可辨识、可定位、不阻断
——文件名兜底比文件夹名兜底好，因为它唯一；但宽口径 except Exception在生产里要分级。
必改清单（按投入产出比排序）
time.time()→time.monotonic()——限流器和上一篇的轮询循环是同一个坑，全局替换 5 分钟搞定。 - 限流参数从
llm_config读——切模型时改一处配置，不动代码。 _upload_to_minio返回Optional[str]+ mypy 强约束——防「忘了 check None」。 - VLM 异常分级
——可重试（超时 / 5xx）用 tenacity重试，不可重试（401 / 403）直接 raise。 - 正则边界写进 docstring
——「仅支持 标准语法」，后人接手时知道边界在哪。
可选改清单（规模到了再做）
限流器换 aiolimiter.AsyncLimiter+ 全节点async化——单 PDF 图片数到 50+ 时收益明显。正则换 markdown-it-py——支持用户上传自定义 MD 时考虑。images/目录名提成配置或从 MD 反推——开放用户上传 MD 时考虑。 加 OpenTelemetry + structlog——多实例部署、需要链路追踪时考虑。
这一篇的核心 takeaway：正则不是写出来的是改出来的——三个细节 (
.*?→字符类、左边界、re.escape) 每一个都是真实坑的痕迹。上下文工程的精华在「喂什么」不在「怎么问」——按章节边界给 VLM 语境，比按字数硬切有效得多。外部数据进re.sub必须用 lambda——这是一条值得刻进肌肉记忆的规则。
「项目骨架精读」系列，每一处结论都基于真实仓库源码逐行核实。这一篇把 NodeMDImg 492 行里的正则演化、章节感知上下文、限流算法、VLM 兜底、MinIO 上传、re.sub 反向引用坑全拆了一遍。要是帮你少踩一个坑，点个赞、点个在看，转给写流水线的朋友。关注我，下一篇把切分节点 node_document_split 的递归切分、重叠窗口、token 计数拆透。
